1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.regionserver;
20
21 import java.io.EOFException;
22 import java.io.FileNotFoundException;
23 import java.io.IOException;
24 import java.io.InterruptedIOException;
25 import java.lang.reflect.Constructor;
26 import java.text.ParseException;
27 import java.util.AbstractList;
28 import java.util.ArrayList;
29 import java.util.Arrays;
30 import java.util.Collection;
31 import java.util.Collections;
32 import java.util.Comparator;
33 import java.util.HashMap;
34 import java.util.HashSet;
35 import java.util.Iterator;
36 import java.util.List;
37 import java.util.Map;
38 import java.util.Map.Entry;
39 import java.util.NavigableMap;
40 import java.util.NavigableSet;
41 import java.util.RandomAccess;
42 import java.util.Set;
43 import java.util.TreeMap;
44 import java.util.concurrent.Callable;
45 import java.util.concurrent.CompletionService;
46 import java.util.concurrent.ConcurrentHashMap;
47 import java.util.concurrent.ConcurrentMap;
48 import java.util.concurrent.ConcurrentSkipListMap;
49 import java.util.concurrent.ExecutionException;
50 import java.util.concurrent.ExecutorCompletionService;
51 import java.util.concurrent.ExecutorService;
52 import java.util.concurrent.Executors;
53 import java.util.concurrent.Future;
54 import java.util.concurrent.FutureTask;
55 import java.util.concurrent.ThreadFactory;
56 import java.util.concurrent.ThreadPoolExecutor;
57 import java.util.concurrent.TimeUnit;
58 import java.util.concurrent.TimeoutException;
59 import java.util.concurrent.atomic.AtomicBoolean;
60 import java.util.concurrent.atomic.AtomicInteger;
61 import java.util.concurrent.atomic.AtomicLong;
62 import java.util.concurrent.locks.Lock;
63 import java.util.concurrent.locks.ReadWriteLock;
64 import java.util.concurrent.locks.ReentrantReadWriteLock;
65
66 import org.apache.commons.lang.RandomStringUtils;
67 import org.apache.commons.logging.Log;
68 import org.apache.commons.logging.LogFactory;
69 import org.apache.hadoop.conf.Configuration;
70 import org.apache.hadoop.fs.FileStatus;
71 import org.apache.hadoop.fs.FileSystem;
72 import org.apache.hadoop.fs.Path;
73 import org.apache.hadoop.hbase.Cell;
74 import org.apache.hadoop.hbase.CellScanner;
75 import org.apache.hadoop.hbase.CellUtil;
76 import org.apache.hadoop.hbase.CompoundConfiguration;
77 import org.apache.hadoop.hbase.DoNotRetryIOException;
78 import org.apache.hadoop.hbase.DroppedSnapshotException;
79 import org.apache.hadoop.hbase.HBaseConfiguration;
80 import org.apache.hadoop.hbase.HColumnDescriptor;
81 import org.apache.hadoop.hbase.HConstants;
82 import org.apache.hadoop.hbase.HConstants.OperationStatusCode;
83 import org.apache.hadoop.hbase.HDFSBlocksDistribution;
84 import org.apache.hadoop.hbase.HRegionInfo;
85 import org.apache.hadoop.hbase.HTableDescriptor;
86 import org.apache.hadoop.hbase.KeyValue;
87 import org.apache.hadoop.hbase.KeyValueUtil;
88 import org.apache.hadoop.hbase.NamespaceDescriptor;
89 import org.apache.hadoop.hbase.NotServingRegionException;
90 import org.apache.hadoop.hbase.RegionTooBusyException;
91 import org.apache.hadoop.hbase.TableName;
92 import org.apache.hadoop.hbase.Tag;
93 import org.apache.hadoop.hbase.TagType;
94 import org.apache.hadoop.hbase.UnknownScannerException;
95 import org.apache.hadoop.hbase.backup.HFileArchiver;
96 import org.apache.hadoop.hbase.classification.InterfaceAudience;
97 import org.apache.hadoop.hbase.client.Append;
98 import org.apache.hadoop.hbase.client.Delete;
99 import org.apache.hadoop.hbase.client.Durability;
100 import org.apache.hadoop.hbase.client.Get;
101 import org.apache.hadoop.hbase.client.Increment;
102 import org.apache.hadoop.hbase.client.IsolationLevel;
103 import org.apache.hadoop.hbase.client.Mutation;
104 import org.apache.hadoop.hbase.client.Put;
105 import org.apache.hadoop.hbase.client.RegionReplicaUtil;
106 import org.apache.hadoop.hbase.client.Result;
107 import org.apache.hadoop.hbase.client.RowMutations;
108 import org.apache.hadoop.hbase.client.Scan;
109 import org.apache.hadoop.hbase.conf.ConfigurationManager;
110 import org.apache.hadoop.hbase.conf.PropagatingConfigurationObserver;
111 import org.apache.hadoop.hbase.coprocessor.RegionObserver;
112 import org.apache.hadoop.hbase.errorhandling.ForeignExceptionSnare;
113 import org.apache.hadoop.hbase.exceptions.FailedSanityCheckException;
114 import org.apache.hadoop.hbase.exceptions.RegionInRecoveryException;
115 import org.apache.hadoop.hbase.exceptions.UnknownProtocolException;
116 import org.apache.hadoop.hbase.filter.ByteArrayComparable;
117 import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
118 import org.apache.hadoop.hbase.filter.FilterWrapper;
119 import org.apache.hadoop.hbase.filter.IncompatibleFilterException;
120 import org.apache.hadoop.hbase.io.HeapSize;
121 import org.apache.hadoop.hbase.io.TimeRange;
122 import org.apache.hadoop.hbase.io.hfile.BlockCache;
123 import org.apache.hadoop.hbase.io.hfile.CacheConfig;
124 import org.apache.hadoop.hbase.io.hfile.HFile;
125 import org.apache.hadoop.hbase.ipc.CallerDisconnectedException;
126 import org.apache.hadoop.hbase.ipc.RpcCallContext;
127 import org.apache.hadoop.hbase.ipc.RpcServer;
128 import org.apache.hadoop.hbase.mob.MobUtils;
129 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
130 import org.apache.hadoop.hbase.monitoring.TaskMonitor;
131 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
132 import org.apache.hadoop.hbase.protobuf.ResponseConverter;
133 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetRegionInfoResponse.CompactionState;
134 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
135 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.CoprocessorServiceCall;
136 import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.RegionLoad;
137 import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.StoreSequenceId;
138 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.SnapshotDescription;
139 import org.apache.hadoop.hbase.protobuf.generated.WALProtos;
140 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.CompactionDescriptor;
141 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.FlushDescriptor;
142 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.FlushDescriptor.FlushAction;
143 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.FlushDescriptor.StoreFlushDescriptor;
144 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.RegionEventDescriptor;
145 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.RegionEventDescriptor.EventType;
146 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.StoreDescriptor;
147 import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl.WriteEntry;
148 import org.apache.hadoop.hbase.regionserver.ScannerContext.LimitScope;
149 import org.apache.hadoop.hbase.regionserver.ScannerContext.NextState;
150 import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
151 import org.apache.hadoop.hbase.regionserver.compactions.CompactionThroughputController;
152 import org.apache.hadoop.hbase.regionserver.compactions.CompactionThroughputControllerFactory;
153 import org.apache.hadoop.hbase.regionserver.compactions.NoLimitCompactionThroughputController;
154 import org.apache.hadoop.hbase.regionserver.wal.HLogKey;
155 import org.apache.hadoop.hbase.regionserver.wal.MetricsWAL;
156 import org.apache.hadoop.hbase.regionserver.wal.ReplayHLogKey;
157 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
158 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
159 import org.apache.hadoop.hbase.regionserver.wal.WALUtil;
160 import org.apache.hadoop.hbase.security.User;
161 import org.apache.hadoop.hbase.snapshot.SnapshotDescriptionUtils;
162 import org.apache.hadoop.hbase.snapshot.SnapshotManifest;
163 import org.apache.hadoop.hbase.util.ByteStringer;
164 import org.apache.hadoop.hbase.util.Bytes;
165 import org.apache.hadoop.hbase.util.CancelableProgressable;
166 import org.apache.hadoop.hbase.util.ClassSize;
167 import org.apache.hadoop.hbase.util.CompressionTest;
168 import org.apache.hadoop.hbase.util.Counter;
169 import org.apache.hadoop.hbase.util.EncryptionTest;
170 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
171 import org.apache.hadoop.hbase.util.FSTableDescriptors;
172 import org.apache.hadoop.hbase.util.FSUtils;
173 import org.apache.hadoop.hbase.util.HashedBytes;
174 import org.apache.hadoop.hbase.util.Pair;
175 import org.apache.hadoop.hbase.util.ServerRegionReplicaUtil;
176 import org.apache.hadoop.hbase.util.Threads;
177 import org.apache.hadoop.hbase.wal.WAL;
178 import org.apache.hadoop.hbase.wal.WALFactory;
179 import org.apache.hadoop.hbase.wal.WALKey;
180 import org.apache.hadoop.hbase.wal.WALSplitter;
181 import org.apache.hadoop.hbase.wal.WALSplitter.MutationReplay;
182 import org.apache.hadoop.io.MultipleIOException;
183 import org.apache.hadoop.util.StringUtils;
184 import org.apache.htrace.Trace;
185 import org.apache.htrace.TraceScope;
186
187 import com.google.common.annotations.VisibleForTesting;
188 import com.google.common.base.Optional;
189 import com.google.common.base.Preconditions;
190 import com.google.common.collect.Lists;
191 import com.google.common.collect.Maps;
192 import com.google.common.io.Closeables;
193 import com.google.protobuf.ByteString;
194 import com.google.protobuf.Descriptors;
195 import com.google.protobuf.Message;
196 import com.google.protobuf.RpcCallback;
197 import com.google.protobuf.RpcController;
198 import com.google.protobuf.Service;
199 import com.google.protobuf.TextFormat;
200
201 @InterfaceAudience.Private
202 public class HRegion implements HeapSize, PropagatingConfigurationObserver, Region {
203 private static final Log LOG = LogFactory.getLog(HRegion.class);
204
205 public static final String LOAD_CFS_ON_DEMAND_CONFIG_KEY =
206 "hbase.hregion.scan.loadColumnFamiliesOnDemand";
207
208
209
210
211
212
213
214
215
216 private final int maxWaitForSeqId;
217 private static final String MAX_WAIT_FOR_SEQ_ID_KEY = "hbase.hregion.max.wait.for.sequenceid.ms";
218 private static final int DEFAULT_MAX_WAIT_FOR_SEQ_ID = 30000;
219
220
221
222
223
224 private static final Durability DEFAULT_DURABILITY = Durability.SYNC_WAL;
225
226 final AtomicBoolean closed = new AtomicBoolean(false);
227
228
229
230
231
232
233 final AtomicBoolean closing = new AtomicBoolean(false);
234
235
236
237
238
239 private volatile long maxFlushedSeqId = HConstants.NO_SEQNUM;
240
241
242
243
244
245
246 private volatile long lastFlushOpSeqId = HConstants.NO_SEQNUM;
247
248
249
250
251
252
253 protected volatile long lastReplayedOpenRegionSeqId = -1L;
254 protected volatile long lastReplayedCompactionSeqId = -1L;
255
256
257
258
259
260
261
262
263
264
265 private final ConcurrentHashMap<HashedBytes, RowLockContext> lockedRows =
266 new ConcurrentHashMap<HashedBytes, RowLockContext>();
267
268 protected final Map<byte[], Store> stores = new ConcurrentSkipListMap<byte[], Store>(
269 Bytes.BYTES_RAWCOMPARATOR);
270
271
272 private Map<String, Service> coprocessorServiceHandlers = Maps.newHashMap();
273
274 private final AtomicLong memstoreSize = new AtomicLong(0);
275
276
277 final Counter numMutationsWithoutWAL = new Counter();
278 final Counter dataInMemoryWithoutWAL = new Counter();
279
280
281 final Counter checkAndMutateChecksPassed = new Counter();
282 final Counter checkAndMutateChecksFailed = new Counter();
283
284
285 final Counter readRequestsCount = new Counter();
286 final Counter writeRequestsCount = new Counter();
287
288
289 private final Counter blockedRequestsCount = new Counter();
290
291
292 final AtomicLong compactionsFinished = new AtomicLong(0L);
293 final AtomicLong compactionNumFilesCompacted = new AtomicLong(0L);
294 final AtomicLong compactionNumBytesCompacted = new AtomicLong(0L);
295
296 private final WAL wal;
297 private final HRegionFileSystem fs;
298 protected final Configuration conf;
299 private final Configuration baseConf;
300 private final KeyValue.KVComparator comparator;
301 private final int rowLockWaitDuration;
302 static final int DEFAULT_ROWLOCK_WAIT_DURATION = 30000;
303
304
305
306
307
308
309
310 final long busyWaitDuration;
311 static final long DEFAULT_BUSY_WAIT_DURATION = HConstants.DEFAULT_HBASE_RPC_TIMEOUT;
312
313
314
315
316 final int maxBusyWaitMultiplier;
317
318
319
320 final long maxBusyWaitDuration;
321
322
323 static final long DEFAULT_ROW_PROCESSOR_TIMEOUT = 60 * 1000L;
324 final ExecutorService rowProcessorExecutor = Executors.newCachedThreadPool();
325
326 private final ConcurrentHashMap<RegionScanner, Long> scannerReadPoints;
327
328
329
330
331 private long openSeqNum = HConstants.NO_SEQNUM;
332
333
334
335
336
337 private boolean isLoadingCfsOnDemandDefault = false;
338
339 private final AtomicInteger majorInProgress = new AtomicInteger(0);
340 private final AtomicInteger minorInProgress = new AtomicInteger(0);
341
342
343
344
345
346
347
348 Map<byte[], Long> maxSeqIdInStores = new TreeMap<byte[], Long>(Bytes.BYTES_COMPARATOR);
349
350
351 private PrepareFlushResult prepareFlushResult = null;
352
353
354
355
356 private boolean disallowWritesInRecovering = false;
357
358
359 private volatile boolean recovering = false;
360
361 private volatile Optional<ConfigurationManager> configurationManager;
362
363
364
365
366
367
368 public long getSmallestReadPoint() {
369 long minimumReadPoint;
370
371
372
373 synchronized(scannerReadPoints) {
374 minimumReadPoint = mvcc.getReadPoint();
375
376 for (Long readPoint: this.scannerReadPoints.values()) {
377 if (readPoint < minimumReadPoint) {
378 minimumReadPoint = readPoint;
379 }
380 }
381 }
382 return minimumReadPoint;
383 }
384
385
386
387
388
389 static class WriteState {
390
391 volatile boolean flushing = false;
392
393 volatile boolean flushRequested = false;
394
395 AtomicInteger compacting = new AtomicInteger(0);
396
397 volatile boolean writesEnabled = true;
398
399 volatile boolean readOnly = false;
400
401
402 volatile boolean readsEnabled = true;
403
404
405
406
407
408
409 synchronized void setReadOnly(final boolean onOff) {
410 this.writesEnabled = !onOff;
411 this.readOnly = onOff;
412 }
413
414 boolean isReadOnly() {
415 return this.readOnly;
416 }
417
418 boolean isFlushRequested() {
419 return this.flushRequested;
420 }
421
422 void setReadsEnabled(boolean readsEnabled) {
423 this.readsEnabled = readsEnabled;
424 }
425
426 static final long HEAP_SIZE = ClassSize.align(
427 ClassSize.OBJECT + 5 * Bytes.SIZEOF_BOOLEAN);
428 }
429
430
431
432
433
434
435
436 public static class FlushResultImpl implements FlushResult {
437 final Result result;
438 final String failureReason;
439 final long flushSequenceId;
440 final boolean wroteFlushWalMarker;
441
442
443
444
445
446
447
448
449 FlushResultImpl(Result result, long flushSequenceId) {
450 this(result, flushSequenceId, null, false);
451 assert result == Result.FLUSHED_NO_COMPACTION_NEEDED || result == Result
452 .FLUSHED_COMPACTION_NEEDED;
453 }
454
455
456
457
458
459
460 FlushResultImpl(Result result, String failureReason, boolean wroteFlushMarker) {
461 this(result, -1, failureReason, wroteFlushMarker);
462 assert result == Result.CANNOT_FLUSH_MEMSTORE_EMPTY || result == Result.CANNOT_FLUSH;
463 }
464
465
466
467
468
469
470
471 FlushResultImpl(Result result, long flushSequenceId, String failureReason,
472 boolean wroteFlushMarker) {
473 this.result = result;
474 this.flushSequenceId = flushSequenceId;
475 this.failureReason = failureReason;
476 this.wroteFlushWalMarker = wroteFlushMarker;
477 }
478
479
480
481
482
483
484 @Override
485 public boolean isFlushSucceeded() {
486 return result == Result.FLUSHED_NO_COMPACTION_NEEDED || result == Result
487 .FLUSHED_COMPACTION_NEEDED;
488 }
489
490
491
492
493
494 @Override
495 public boolean isCompactionNeeded() {
496 return result == Result.FLUSHED_COMPACTION_NEEDED;
497 }
498
499 @Override
500 public String toString() {
501 return new StringBuilder()
502 .append("flush result:").append(result).append(", ")
503 .append("failureReason:").append(failureReason).append(",")
504 .append("flush seq id").append(flushSequenceId).toString();
505 }
506
507 @Override
508 public Result getResult() {
509 return result;
510 }
511 }
512
513
514 @VisibleForTesting
515 static class PrepareFlushResult {
516 final FlushResult result;
517 final TreeMap<byte[], StoreFlushContext> storeFlushCtxs;
518 final TreeMap<byte[], List<Path>> committedFiles;
519 final TreeMap<byte[], Long> storeFlushableSize;
520 final long startTime;
521 final long flushOpSeqId;
522 final long flushedSeqId;
523 final long totalFlushableSize;
524
525
526 PrepareFlushResult(FlushResult result, long flushSeqId) {
527 this(result, null, null, null, Math.max(0, flushSeqId), 0, 0, 0);
528 }
529
530
531 PrepareFlushResult(
532 TreeMap<byte[], StoreFlushContext> storeFlushCtxs,
533 TreeMap<byte[], List<Path>> committedFiles,
534 TreeMap<byte[], Long> storeFlushableSize, long startTime, long flushSeqId,
535 long flushedSeqId, long totalFlushableSize) {
536 this(null, storeFlushCtxs, committedFiles, storeFlushableSize, startTime,
537 flushSeqId, flushedSeqId, totalFlushableSize);
538 }
539
540 private PrepareFlushResult(
541 FlushResult result,
542 TreeMap<byte[], StoreFlushContext> storeFlushCtxs,
543 TreeMap<byte[], List<Path>> committedFiles,
544 TreeMap<byte[], Long> storeFlushableSize, long startTime, long flushSeqId,
545 long flushedSeqId, long totalFlushableSize) {
546 this.result = result;
547 this.storeFlushCtxs = storeFlushCtxs;
548 this.committedFiles = committedFiles;
549 this.storeFlushableSize = storeFlushableSize;
550 this.startTime = startTime;
551 this.flushOpSeqId = flushSeqId;
552 this.flushedSeqId = flushedSeqId;
553 this.totalFlushableSize = totalFlushableSize;
554 }
555
556 public FlushResult getResult() {
557 return this.result;
558 }
559 }
560
561 final WriteState writestate = new WriteState();
562
563 long memstoreFlushSize;
564 final long timestampSlop;
565 final long rowProcessorTimeout;
566
567
568 private final ConcurrentMap<Store, Long> lastStoreFlushTimeMap =
569 new ConcurrentHashMap<Store, Long>();
570
571 final RegionServerServices rsServices;
572 private RegionServerAccounting rsAccounting;
573 private long flushCheckInterval;
574
575 private long flushPerChanges;
576 private long blockingMemStoreSize;
577 final long threadWakeFrequency;
578
579 final ReentrantReadWriteLock lock =
580 new ReentrantReadWriteLock();
581
582
583 private final ReentrantReadWriteLock updatesLock =
584 new ReentrantReadWriteLock();
585 private boolean splitRequest;
586 private byte[] explicitSplitPoint = null;
587
588 private final MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl();
589
590
591 private RegionCoprocessorHost coprocessorHost;
592
593 private HTableDescriptor htableDescriptor = null;
594 private RegionSplitPolicy splitPolicy;
595 private FlushPolicy flushPolicy;
596
597 private final MetricsRegion metricsRegion;
598 private final MetricsRegionWrapperImpl metricsRegionWrapper;
599 private final Durability durability;
600 private final boolean regionStatsEnabled;
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623 @Deprecated
624 @VisibleForTesting
625 public HRegion(final Path tableDir, final WAL wal, final FileSystem fs,
626 final Configuration confParam, final HRegionInfo regionInfo,
627 final HTableDescriptor htd, final RegionServerServices rsServices) {
628 this(new HRegionFileSystem(confParam, fs, tableDir, regionInfo),
629 wal, confParam, htd, rsServices);
630 }
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648 public HRegion(final HRegionFileSystem fs, final WAL wal, final Configuration confParam,
649 final HTableDescriptor htd, final RegionServerServices rsServices) {
650 if (htd == null) {
651 throw new IllegalArgumentException("Need table descriptor");
652 }
653
654 if (confParam instanceof CompoundConfiguration) {
655 throw new IllegalArgumentException("Need original base configuration");
656 }
657
658 this.comparator = fs.getRegionInfo().getComparator();
659 this.wal = wal;
660 this.fs = fs;
661
662
663 this.baseConf = confParam;
664 this.conf = new CompoundConfiguration()
665 .add(confParam)
666 .addStringMap(htd.getConfiguration())
667 .addWritableMap(htd.getValues());
668 this.flushCheckInterval = conf.getInt(MEMSTORE_PERIODIC_FLUSH_INTERVAL,
669 DEFAULT_CACHE_FLUSH_INTERVAL);
670 this.flushPerChanges = conf.getLong(MEMSTORE_FLUSH_PER_CHANGES, DEFAULT_FLUSH_PER_CHANGES);
671 if (this.flushPerChanges > MAX_FLUSH_PER_CHANGES) {
672 throw new IllegalArgumentException(MEMSTORE_FLUSH_PER_CHANGES + " can not exceed "
673 + MAX_FLUSH_PER_CHANGES);
674 }
675 this.rowLockWaitDuration = conf.getInt("hbase.rowlock.wait.duration",
676 DEFAULT_ROWLOCK_WAIT_DURATION);
677
678 this.maxWaitForSeqId = conf.getInt(MAX_WAIT_FOR_SEQ_ID_KEY, DEFAULT_MAX_WAIT_FOR_SEQ_ID);
679 this.isLoadingCfsOnDemandDefault = conf.getBoolean(LOAD_CFS_ON_DEMAND_CONFIG_KEY, true);
680 this.htableDescriptor = htd;
681 this.rsServices = rsServices;
682 this.threadWakeFrequency = conf.getLong(HConstants.THREAD_WAKE_FREQUENCY, 10 * 1000);
683 setHTableSpecificConf();
684 this.scannerReadPoints = new ConcurrentHashMap<RegionScanner, Long>();
685
686 this.busyWaitDuration = conf.getLong(
687 "hbase.busy.wait.duration", DEFAULT_BUSY_WAIT_DURATION);
688 this.maxBusyWaitMultiplier = conf.getInt("hbase.busy.wait.multiplier.max", 2);
689 if (busyWaitDuration * maxBusyWaitMultiplier <= 0L) {
690 throw new IllegalArgumentException("Invalid hbase.busy.wait.duration ("
691 + busyWaitDuration + ") or hbase.busy.wait.multiplier.max ("
692 + maxBusyWaitMultiplier + "). Their product should be positive");
693 }
694 this.maxBusyWaitDuration = conf.getLong("hbase.ipc.client.call.purge.timeout",
695 2 * HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
696
697
698
699
700
701
702
703 this.timestampSlop = conf.getLong(
704 "hbase.hregion.keyvalue.timestamp.slop.millisecs",
705 HConstants.LATEST_TIMESTAMP);
706
707
708
709
710
711 this.rowProcessorTimeout = conf.getLong(
712 "hbase.hregion.row.processor.timeout", DEFAULT_ROW_PROCESSOR_TIMEOUT);
713 this.durability = htd.getDurability() == Durability.USE_DEFAULT
714 ? DEFAULT_DURABILITY
715 : htd.getDurability();
716 if (rsServices != null) {
717 this.rsAccounting = this.rsServices.getRegionServerAccounting();
718
719
720 this.coprocessorHost = new RegionCoprocessorHost(this, rsServices, conf);
721 this.metricsRegionWrapper = new MetricsRegionWrapperImpl(this);
722 this.metricsRegion = new MetricsRegion(this.metricsRegionWrapper);
723
724 Map<String, Region> recoveringRegions = rsServices.getRecoveringRegions();
725 String encodedName = getRegionInfo().getEncodedName();
726 if (recoveringRegions != null && recoveringRegions.containsKey(encodedName)) {
727 this.recovering = true;
728 recoveringRegions.put(encodedName, this);
729 }
730 } else {
731 this.metricsRegionWrapper = null;
732 this.metricsRegion = null;
733 }
734 if (LOG.isDebugEnabled()) {
735
736 LOG.debug("Instantiated " + this);
737 }
738
739
740 this.disallowWritesInRecovering =
741 conf.getBoolean(HConstants.DISALLOW_WRITES_IN_RECOVERING,
742 HConstants.DEFAULT_DISALLOW_WRITES_IN_RECOVERING_CONFIG);
743 configurationManager = Optional.absent();
744
745
746 this.regionStatsEnabled = htd.getTableName().getNamespaceAsString().equals(
747 NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR) ?
748 false :
749 conf.getBoolean(HConstants.ENABLE_CLIENT_BACKPRESSURE,
750 HConstants.DEFAULT_ENABLE_CLIENT_BACKPRESSURE);
751 }
752
753 void setHTableSpecificConf() {
754 if (this.htableDescriptor == null) return;
755 long flushSize = this.htableDescriptor.getMemStoreFlushSize();
756
757 if (flushSize <= 0) {
758 flushSize = conf.getLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE,
759 HTableDescriptor.DEFAULT_MEMSTORE_FLUSH_SIZE);
760 }
761 this.memstoreFlushSize = flushSize;
762 this.blockingMemStoreSize = this.memstoreFlushSize *
763 conf.getLong(HConstants.HREGION_MEMSTORE_BLOCK_MULTIPLIER,
764 HConstants.DEFAULT_HREGION_MEMSTORE_BLOCK_MULTIPLIER);
765 }
766
767
768
769
770
771
772
773
774
775 @Deprecated
776 public long initialize() throws IOException {
777 return initialize(null);
778 }
779
780
781
782
783
784
785
786
787 private long initialize(final CancelableProgressable reporter) throws IOException {
788
789
790 if (htableDescriptor.getColumnFamilies().length == 0) {
791 throw new DoNotRetryIOException("Table " + htableDescriptor.getNameAsString() +
792 " should have at least one column family.");
793 }
794
795 MonitoredTask status = TaskMonitor.get().createStatus("Initializing region " + this);
796 long nextSeqId = -1;
797 try {
798 nextSeqId = initializeRegionInternals(reporter, status);
799 return nextSeqId;
800 } finally {
801
802
803 if (nextSeqId == -1) {
804 status.abort("Exception during region " + getRegionInfo().getRegionNameAsString() +
805 " initialization.");
806 }
807 }
808 }
809
810 private long initializeRegionInternals(final CancelableProgressable reporter,
811 final MonitoredTask status) throws IOException {
812 if (coprocessorHost != null) {
813 status.setStatus("Running coprocessor pre-open hook");
814 coprocessorHost.preOpen();
815 }
816
817
818 status.setStatus("Writing region info on filesystem");
819 fs.checkRegionInfoOnFilesystem();
820
821
822 status.setStatus("Initializing all the Stores");
823 long maxSeqId = initializeStores(reporter, status);
824 this.mvcc.advanceTo(maxSeqId);
825 if (ServerRegionReplicaUtil.shouldReplayRecoveredEdits(this)) {
826
827 maxSeqId = Math.max(maxSeqId,
828 replayRecoveredEditsIfAny(this.fs.getRegionDir(), maxSeqIdInStores, reporter, status));
829
830 this.mvcc.advanceTo(maxSeqId);
831 }
832 this.lastReplayedOpenRegionSeqId = maxSeqId;
833
834 this.writestate.setReadOnly(ServerRegionReplicaUtil.isReadOnly(this));
835 this.writestate.flushRequested = false;
836 this.writestate.compacting.set(0);
837
838 if (this.writestate.writesEnabled) {
839
840 status.setStatus("Cleaning up temporary data from old regions");
841 fs.cleanupTempDir();
842 }
843
844 if (this.writestate.writesEnabled) {
845 status.setStatus("Cleaning up detritus from prior splits");
846
847
848
849 fs.cleanupAnySplitDetritus();
850 fs.cleanupMergesDir();
851 }
852
853
854 this.splitPolicy = RegionSplitPolicy.create(this, conf);
855
856
857 this.flushPolicy = FlushPolicyFactory.create(this, conf);
858
859 long lastFlushTime = EnvironmentEdgeManager.currentTime();
860 for (Store store: stores.values()) {
861 this.lastStoreFlushTimeMap.put(store, lastFlushTime);
862 }
863
864
865
866 long nextSeqid = maxSeqId;
867
868
869
870
871 if (this.writestate.writesEnabled) {
872 nextSeqid = WALSplitter.writeRegionSequenceIdFile(this.fs.getFileSystem(), this.fs
873 .getRegionDir(), nextSeqid, (this.recovering ? (this.flushPerChanges + 10000000) : 1));
874 } else {
875 nextSeqid++;
876 }
877
878 LOG.info("Onlined " + this.getRegionInfo().getShortNameToLog() +
879 "; next sequenceid=" + nextSeqid);
880
881
882 this.closing.set(false);
883 this.closed.set(false);
884
885 if (coprocessorHost != null) {
886 status.setStatus("Running coprocessor post-open hooks");
887 coprocessorHost.postOpen();
888 }
889
890 status.markComplete("Region opened successfully");
891 return nextSeqid;
892 }
893
894
895
896
897
898
899
900
901 private long initializeStores(final CancelableProgressable reporter, MonitoredTask status)
902 throws IOException {
903
904
905 long maxSeqId = -1;
906
907 long maxMemstoreTS = -1;
908
909 if (!htableDescriptor.getFamilies().isEmpty()) {
910
911 ThreadPoolExecutor storeOpenerThreadPool =
912 getStoreOpenAndCloseThreadPool("StoreOpener-" + this.getRegionInfo().getShortNameToLog());
913 CompletionService<HStore> completionService =
914 new ExecutorCompletionService<HStore>(storeOpenerThreadPool);
915
916
917 for (final HColumnDescriptor family : htableDescriptor.getFamilies()) {
918 status.setStatus("Instantiating store for column family " + family);
919 completionService.submit(new Callable<HStore>() {
920 @Override
921 public HStore call() throws IOException {
922 return instantiateHStore(family);
923 }
924 });
925 }
926 boolean allStoresOpened = false;
927 try {
928 for (int i = 0; i < htableDescriptor.getFamilies().size(); i++) {
929 Future<HStore> future = completionService.take();
930 HStore store = future.get();
931 this.stores.put(store.getFamily().getName(), store);
932
933 long storeMaxSequenceId = store.getMaxSequenceId();
934 maxSeqIdInStores.put(store.getColumnFamilyName().getBytes(),
935 storeMaxSequenceId);
936 if (maxSeqId == -1 || storeMaxSequenceId > maxSeqId) {
937 maxSeqId = storeMaxSequenceId;
938 }
939 long maxStoreMemstoreTS = store.getMaxMemstoreTS();
940 if (maxStoreMemstoreTS > maxMemstoreTS) {
941 maxMemstoreTS = maxStoreMemstoreTS;
942 }
943 }
944 allStoresOpened = true;
945 } catch (InterruptedException e) {
946 throw (InterruptedIOException)new InterruptedIOException().initCause(e);
947 } catch (ExecutionException e) {
948 throw new IOException(e.getCause());
949 } finally {
950 storeOpenerThreadPool.shutdownNow();
951 if (!allStoresOpened) {
952
953 LOG.error("Could not initialize all stores for the region=" + this);
954 for (Store store : this.stores.values()) {
955 try {
956 store.close();
957 } catch (IOException e) {
958 LOG.warn(e.getMessage());
959 }
960 }
961 }
962 }
963 }
964 return Math.max(maxSeqId, maxMemstoreTS + 1);
965 }
966
967 private void initializeWarmup(final CancelableProgressable reporter) throws IOException {
968 MonitoredTask status = TaskMonitor.get().createStatus("Initializing region " + this);
969
970
971 status.setStatus("Warming up all the Stores");
972 initializeStores(reporter, status);
973 }
974
975
976
977
978 private NavigableMap<byte[], List<Path>> getStoreFiles() {
979 NavigableMap<byte[], List<Path>> allStoreFiles =
980 new TreeMap<byte[], List<Path>>(Bytes.BYTES_COMPARATOR);
981 for (Store store: getStores()) {
982 Collection<StoreFile> storeFiles = store.getStorefiles();
983 if (storeFiles == null) continue;
984 List<Path> storeFileNames = new ArrayList<Path>();
985 for (StoreFile storeFile: storeFiles) {
986 storeFileNames.add(storeFile.getPath());
987 }
988 allStoreFiles.put(store.getFamily().getName(), storeFileNames);
989 }
990 return allStoreFiles;
991 }
992
993 private void writeRegionOpenMarker(WAL wal, long openSeqId) throws IOException {
994 Map<byte[], List<Path>> storeFiles = getStoreFiles();
995 RegionEventDescriptor regionOpenDesc = ProtobufUtil.toRegionEventDescriptor(
996 RegionEventDescriptor.EventType.REGION_OPEN, getRegionInfo(), openSeqId,
997 getRegionServerServices().getServerName(), storeFiles);
998 WALUtil.writeRegionEventMarker(wal, getTableDesc(), getRegionInfo(), regionOpenDesc, mvcc);
999 }
1000
1001 private void writeRegionCloseMarker(WAL wal) throws IOException {
1002 Map<byte[], List<Path>> storeFiles = getStoreFiles();
1003 RegionEventDescriptor regionEventDesc = ProtobufUtil.toRegionEventDescriptor(
1004 RegionEventDescriptor.EventType.REGION_CLOSE, getRegionInfo(), mvcc.getReadPoint(),
1005 getRegionServerServices().getServerName(), storeFiles);
1006 WALUtil.writeRegionEventMarker(wal, getTableDesc(), getRegionInfo(), regionEventDesc, mvcc);
1007
1008
1009
1010
1011 if (this.fs.getFileSystem().exists(this.fs.getRegionDir())) {
1012 WALSplitter.writeRegionSequenceIdFile(this.fs.getFileSystem(), this.fs.getRegionDir(),
1013 mvcc.getReadPoint(), 0);
1014 }
1015 }
1016
1017
1018
1019
1020 public boolean hasReferences() {
1021 for (Store store : this.stores.values()) {
1022 if (store.hasReferences()) return true;
1023 }
1024 return false;
1025 }
1026
1027 @Override
1028 public HDFSBlocksDistribution getHDFSBlocksDistribution() {
1029 HDFSBlocksDistribution hdfsBlocksDistribution =
1030 new HDFSBlocksDistribution();
1031 synchronized (this.stores) {
1032 for (Store store : this.stores.values()) {
1033 Collection<StoreFile> storeFiles = store.getStorefiles();
1034 if (storeFiles == null) continue;
1035 for (StoreFile sf : storeFiles) {
1036 HDFSBlocksDistribution storeFileBlocksDistribution =
1037 sf.getHDFSBlockDistribution();
1038 hdfsBlocksDistribution.add(storeFileBlocksDistribution);
1039 }
1040 }
1041 }
1042 return hdfsBlocksDistribution;
1043 }
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053 public static HDFSBlocksDistribution computeHDFSBlocksDistribution(final Configuration conf,
1054 final HTableDescriptor tableDescriptor, final HRegionInfo regionInfo) throws IOException {
1055 Path tablePath = FSUtils.getTableDir(FSUtils.getRootDir(conf), tableDescriptor.getTableName());
1056 return computeHDFSBlocksDistribution(conf, tableDescriptor, regionInfo, tablePath);
1057 }
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068 public static HDFSBlocksDistribution computeHDFSBlocksDistribution(final Configuration conf,
1069 final HTableDescriptor tableDescriptor, final HRegionInfo regionInfo, Path tablePath)
1070 throws IOException {
1071 HDFSBlocksDistribution hdfsBlocksDistribution = new HDFSBlocksDistribution();
1072 FileSystem fs = tablePath.getFileSystem(conf);
1073
1074 HRegionFileSystem regionFs = new HRegionFileSystem(conf, fs, tablePath, regionInfo);
1075 for (HColumnDescriptor family: tableDescriptor.getFamilies()) {
1076 Collection<StoreFileInfo> storeFiles = regionFs.getStoreFiles(family.getNameAsString());
1077 if (storeFiles == null) continue;
1078 for (StoreFileInfo storeFileInfo : storeFiles) {
1079 try {
1080 hdfsBlocksDistribution.add(storeFileInfo.computeHDFSBlocksDistribution(fs));
1081 } catch (IOException ioe) {
1082 LOG.warn("Error getting hdfs block distribution for " + storeFileInfo);
1083 }
1084 }
1085 }
1086 return hdfsBlocksDistribution;
1087 }
1088
1089
1090
1091
1092
1093
1094 public long addAndGetGlobalMemstoreSize(long memStoreSize) {
1095 if (this.rsAccounting != null) {
1096 rsAccounting.addAndGetGlobalMemstoreSize(memStoreSize);
1097 }
1098 return this.memstoreSize.addAndGet(memStoreSize);
1099 }
1100
1101 @Override
1102 public HRegionInfo getRegionInfo() {
1103 return this.fs.getRegionInfo();
1104 }
1105
1106
1107
1108
1109
1110 RegionServerServices getRegionServerServices() {
1111 return this.rsServices;
1112 }
1113
1114 @Override
1115 public long getReadRequestsCount() {
1116 return readRequestsCount.get();
1117 }
1118
1119 @Override
1120 public void updateReadRequestsCount(long i) {
1121 readRequestsCount.add(i);
1122 }
1123
1124 @Override
1125 public long getWriteRequestsCount() {
1126 return writeRequestsCount.get();
1127 }
1128
1129 @Override
1130 public void updateWriteRequestsCount(long i) {
1131 writeRequestsCount.add(i);
1132 }
1133
1134 @Override
1135 public long getMemstoreSize() {
1136 return memstoreSize.get();
1137 }
1138
1139 @Override
1140 public long getNumMutationsWithoutWAL() {
1141 return numMutationsWithoutWAL.get();
1142 }
1143
1144 @Override
1145 public long getDataInMemoryWithoutWAL() {
1146 return dataInMemoryWithoutWAL.get();
1147 }
1148
1149 @Override
1150 public long getBlockedRequestsCount() {
1151 return blockedRequestsCount.get();
1152 }
1153
1154 @Override
1155 public long getCheckAndMutateChecksPassed() {
1156 return checkAndMutateChecksPassed.get();
1157 }
1158
1159 @Override
1160 public long getCheckAndMutateChecksFailed() {
1161 return checkAndMutateChecksFailed.get();
1162 }
1163
1164 @Override
1165 public MetricsRegion getMetrics() {
1166 return metricsRegion;
1167 }
1168
1169 @Override
1170 public boolean isClosed() {
1171 return this.closed.get();
1172 }
1173
1174 @Override
1175 public boolean isClosing() {
1176 return this.closing.get();
1177 }
1178
1179 @Override
1180 public boolean isReadOnly() {
1181 return this.writestate.isReadOnly();
1182 }
1183
1184
1185
1186
1187 public void setRecovering(boolean newState) {
1188 boolean wasRecovering = this.recovering;
1189
1190
1191 if (wal != null && getRegionServerServices() != null && !writestate.readOnly
1192 && wasRecovering && !newState) {
1193
1194
1195 boolean forceFlush = getTableDesc().getRegionReplication() > 1;
1196
1197
1198 MonitoredTask status = TaskMonitor.get().createStatus(
1199 "Flushing region " + this + " because recovery is finished");
1200 try {
1201 if (forceFlush) {
1202 internalFlushcache(status);
1203 }
1204
1205 status.setStatus("Writing region open event marker to WAL because recovery is finished");
1206 try {
1207 long seqId = openSeqNum;
1208
1209 if (wal != null) {
1210 seqId = getNextSequenceId(wal);
1211 }
1212 writeRegionOpenMarker(wal, seqId);
1213 } catch (IOException e) {
1214
1215
1216 LOG.warn(getRegionInfo().getEncodedName() + " : was not able to write region opening "
1217 + "event to WAL, continueing", e);
1218 }
1219 } catch (IOException ioe) {
1220
1221
1222 LOG.warn(getRegionInfo().getEncodedName() + " : was not able to flush "
1223 + "event to WAL, continueing", ioe);
1224 } finally {
1225 status.cleanup();
1226 }
1227 }
1228
1229 this.recovering = newState;
1230 if (wasRecovering && !recovering) {
1231
1232 coprocessorHost.postLogReplay();
1233 }
1234 }
1235
1236 @Override
1237 public boolean isRecovering() {
1238 return this.recovering;
1239 }
1240
1241 @Override
1242 public boolean isAvailable() {
1243 return !isClosed() && !isClosing();
1244 }
1245
1246
1247 public boolean isSplittable() {
1248 return isAvailable() && !hasReferences();
1249 }
1250
1251
1252
1253
1254 public boolean isMergeable() {
1255 if (!isAvailable()) {
1256 LOG.debug("Region " + getRegionInfo().getRegionNameAsString()
1257 + " is not mergeable because it is closing or closed");
1258 return false;
1259 }
1260 if (hasReferences()) {
1261 LOG.debug("Region " + getRegionInfo().getRegionNameAsString()
1262 + " is not mergeable because it has references");
1263 return false;
1264 }
1265
1266 return true;
1267 }
1268
1269 public boolean areWritesEnabled() {
1270 synchronized(this.writestate) {
1271 return this.writestate.writesEnabled;
1272 }
1273 }
1274
1275 public MultiVersionConcurrencyControl getMVCC() {
1276 return mvcc;
1277 }
1278
1279 @Override
1280 public long getMaxFlushedSeqId() {
1281 return maxFlushedSeqId;
1282 }
1283
1284 @Override
1285 public long getReadpoint(IsolationLevel isolationLevel) {
1286 if (isolationLevel == IsolationLevel.READ_UNCOMMITTED) {
1287
1288 return Long.MAX_VALUE;
1289 }
1290 return mvcc.getReadPoint();
1291 }
1292
1293 @Override
1294 public boolean isLoadingCfsOnDemandDefault() {
1295 return this.isLoadingCfsOnDemandDefault;
1296 }
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314 public Map<byte[], List<StoreFile>> close() throws IOException {
1315 return close(false);
1316 }
1317
1318 private final Object closeLock = new Object();
1319
1320
1321 public static final String MEMSTORE_PERIODIC_FLUSH_INTERVAL =
1322 "hbase.regionserver.optionalcacheflushinterval";
1323
1324 public static final int DEFAULT_CACHE_FLUSH_INTERVAL = 3600000;
1325
1326 public static final int SYSTEM_CACHE_FLUSH_INTERVAL = 300000;
1327
1328
1329 public static final String MEMSTORE_FLUSH_PER_CHANGES =
1330 "hbase.regionserver.flush.per.changes";
1331 public static final long DEFAULT_FLUSH_PER_CHANGES = 30000000;
1332
1333
1334
1335
1336 public static final long MAX_FLUSH_PER_CHANGES = 1000000000;
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355 public Map<byte[], List<StoreFile>> close(final boolean abort) throws IOException {
1356
1357
1358 MonitoredTask status = TaskMonitor.get().createStatus(
1359 "Closing region " + this +
1360 (abort ? " due to abort" : ""));
1361
1362 status.setStatus("Waiting for close lock");
1363 try {
1364 synchronized (closeLock) {
1365 return doClose(abort, status);
1366 }
1367 } finally {
1368 status.cleanup();
1369 }
1370 }
1371
1372
1373
1374
1375 @VisibleForTesting
1376 public void setClosing(boolean closing) {
1377 this.closing.set(closing);
1378 }
1379
1380 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="UL_UNRELEASED_LOCK_EXCEPTION_PATH",
1381 justification="I think FindBugs is confused")
1382 private Map<byte[], List<StoreFile>> doClose(final boolean abort, MonitoredTask status)
1383 throws IOException {
1384 if (isClosed()) {
1385 LOG.warn("Region " + this + " already closed");
1386 return null;
1387 }
1388
1389 if (coprocessorHost != null) {
1390 status.setStatus("Running coprocessor pre-close hooks");
1391 this.coprocessorHost.preClose(abort);
1392 }
1393
1394 status.setStatus("Disabling compacts and flushes for region");
1395 boolean canFlush = true;
1396 synchronized (writestate) {
1397
1398
1399 canFlush = !writestate.readOnly;
1400 writestate.writesEnabled = false;
1401 LOG.debug("Closing " + this + ": disabling compactions & flushes");
1402 waitForFlushesAndCompactions();
1403 }
1404
1405
1406
1407 if (!abort && worthPreFlushing() && canFlush) {
1408 status.setStatus("Pre-flushing region before close");
1409 LOG.info("Running close preflush of " + getRegionInfo().getRegionNameAsString());
1410 try {
1411 internalFlushcache(status);
1412 } catch (IOException ioe) {
1413
1414 status.setStatus("Failed pre-flush " + this + "; " + ioe.getMessage());
1415 }
1416 }
1417
1418
1419 lock.writeLock().lock();
1420 this.closing.set(true);
1421 status.setStatus("Disabling writes for close");
1422 try {
1423 if (this.isClosed()) {
1424 status.abort("Already got closed by another process");
1425
1426 return null;
1427 }
1428 LOG.debug("Updates disabled for region " + this);
1429
1430 if (!abort && canFlush) {
1431 int flushCount = 0;
1432 while (this.memstoreSize.get() > 0) {
1433 try {
1434 if (flushCount++ > 0) {
1435 int actualFlushes = flushCount - 1;
1436 if (actualFlushes > 5) {
1437
1438
1439 throw new DroppedSnapshotException("Failed clearing memory after " +
1440 actualFlushes + " attempts on region: " +
1441 Bytes.toStringBinary(getRegionInfo().getRegionName()));
1442 }
1443 LOG.info("Running extra flush, " + actualFlushes +
1444 " (carrying snapshot?) " + this);
1445 }
1446 internalFlushcache(status);
1447 } catch (IOException ioe) {
1448 status.setStatus("Failed flush " + this + ", putting online again");
1449 synchronized (writestate) {
1450 writestate.writesEnabled = true;
1451 }
1452
1453 throw ioe;
1454 }
1455 }
1456 }
1457
1458 Map<byte[], List<StoreFile>> result =
1459 new TreeMap<byte[], List<StoreFile>>(Bytes.BYTES_COMPARATOR);
1460 if (!stores.isEmpty()) {
1461
1462 ThreadPoolExecutor storeCloserThreadPool =
1463 getStoreOpenAndCloseThreadPool("StoreCloserThread-" +
1464 getRegionInfo().getRegionNameAsString());
1465 CompletionService<Pair<byte[], Collection<StoreFile>>> completionService =
1466 new ExecutorCompletionService<Pair<byte[], Collection<StoreFile>>>(storeCloserThreadPool);
1467
1468
1469 for (final Store store : stores.values()) {
1470 long flushableSize = store.getFlushableSize();
1471 if (!(abort || flushableSize == 0 || writestate.readOnly)) {
1472 if (getRegionServerServices() != null) {
1473 getRegionServerServices().abort("Assertion failed while closing store "
1474 + getRegionInfo().getRegionNameAsString() + " " + store
1475 + ". flushableSize expected=0, actual= " + flushableSize
1476 + ". Current memstoreSize=" + getMemstoreSize() + ". Maybe a coprocessor "
1477 + "operation failed and left the memstore in a partially updated state.", null);
1478 }
1479 }
1480 completionService
1481 .submit(new Callable<Pair<byte[], Collection<StoreFile>>>() {
1482 @Override
1483 public Pair<byte[], Collection<StoreFile>> call() throws IOException {
1484 return new Pair<byte[], Collection<StoreFile>>(
1485 store.getFamily().getName(), store.close());
1486 }
1487 });
1488 }
1489 try {
1490 for (int i = 0; i < stores.size(); i++) {
1491 Future<Pair<byte[], Collection<StoreFile>>> future = completionService.take();
1492 Pair<byte[], Collection<StoreFile>> storeFiles = future.get();
1493 List<StoreFile> familyFiles = result.get(storeFiles.getFirst());
1494 if (familyFiles == null) {
1495 familyFiles = new ArrayList<StoreFile>();
1496 result.put(storeFiles.getFirst(), familyFiles);
1497 }
1498 familyFiles.addAll(storeFiles.getSecond());
1499 }
1500 } catch (InterruptedException e) {
1501 throw (InterruptedIOException)new InterruptedIOException().initCause(e);
1502 } catch (ExecutionException e) {
1503 throw new IOException(e.getCause());
1504 } finally {
1505 storeCloserThreadPool.shutdownNow();
1506 }
1507 }
1508
1509 status.setStatus("Writing region close event to WAL");
1510 if (!abort && wal != null && getRegionServerServices() != null && !writestate.readOnly) {
1511 writeRegionCloseMarker(wal);
1512 }
1513
1514 this.closed.set(true);
1515 if (!canFlush) {
1516 addAndGetGlobalMemstoreSize(-memstoreSize.get());
1517 } else if (memstoreSize.get() != 0) {
1518 LOG.error("Memstore size is " + memstoreSize.get());
1519 }
1520 if (coprocessorHost != null) {
1521 status.setStatus("Running coprocessor post-close hooks");
1522 this.coprocessorHost.postClose(abort);
1523 }
1524 if (this.metricsRegion != null) {
1525 this.metricsRegion.close();
1526 }
1527 if (this.metricsRegionWrapper != null) {
1528 Closeables.closeQuietly(this.metricsRegionWrapper);
1529 }
1530 status.markComplete("Closed");
1531 LOG.info("Closed " + this);
1532 return result;
1533 } finally {
1534 lock.writeLock().unlock();
1535 }
1536 }
1537
1538 @Override
1539 public void waitForFlushesAndCompactions() {
1540 synchronized (writestate) {
1541 if (this.writestate.readOnly) {
1542
1543
1544 return;
1545 }
1546 boolean interrupted = false;
1547 try {
1548 while (writestate.compacting.get() > 0 || writestate.flushing) {
1549 LOG.debug("waiting for " + writestate.compacting + " compactions"
1550 + (writestate.flushing ? " & cache flush" : "") + " to complete for region " + this);
1551 try {
1552 writestate.wait();
1553 } catch (InterruptedException iex) {
1554
1555 LOG.warn("Interrupted while waiting");
1556 interrupted = true;
1557 }
1558 }
1559 } finally {
1560 if (interrupted) {
1561 Thread.currentThread().interrupt();
1562 }
1563 }
1564 }
1565 }
1566
1567 protected ThreadPoolExecutor getStoreOpenAndCloseThreadPool(
1568 final String threadNamePrefix) {
1569 int numStores = Math.max(1, this.htableDescriptor.getFamilies().size());
1570 int maxThreads = Math.min(numStores,
1571 conf.getInt(HConstants.HSTORE_OPEN_AND_CLOSE_THREADS_MAX,
1572 HConstants.DEFAULT_HSTORE_OPEN_AND_CLOSE_THREADS_MAX));
1573 return getOpenAndCloseThreadPool(maxThreads, threadNamePrefix);
1574 }
1575
1576 protected ThreadPoolExecutor getStoreFileOpenAndCloseThreadPool(
1577 final String threadNamePrefix) {
1578 int numStores = Math.max(1, this.htableDescriptor.getFamilies().size());
1579 int maxThreads = Math.max(1,
1580 conf.getInt(HConstants.HSTORE_OPEN_AND_CLOSE_THREADS_MAX,
1581 HConstants.DEFAULT_HSTORE_OPEN_AND_CLOSE_THREADS_MAX)
1582 / numStores);
1583 return getOpenAndCloseThreadPool(maxThreads, threadNamePrefix);
1584 }
1585
1586 static ThreadPoolExecutor getOpenAndCloseThreadPool(int maxThreads,
1587 final String threadNamePrefix) {
1588 return Threads.getBoundedCachedThreadPool(maxThreads, 30L, TimeUnit.SECONDS,
1589 new ThreadFactory() {
1590 private int count = 1;
1591
1592 @Override
1593 public Thread newThread(Runnable r) {
1594 return new Thread(r, threadNamePrefix + "-" + count++);
1595 }
1596 });
1597 }
1598
1599
1600
1601
1602 private boolean worthPreFlushing() {
1603 return this.memstoreSize.get() >
1604 this.conf.getLong("hbase.hregion.preclose.flush.size", 1024 * 1024 * 5);
1605 }
1606
1607
1608
1609
1610
1611 @Override
1612 public HTableDescriptor getTableDesc() {
1613 return this.htableDescriptor;
1614 }
1615
1616
1617 public WAL getWAL() {
1618 return this.wal;
1619 }
1620
1621
1622
1623
1624
1625
1626
1627
1628 Configuration getBaseConf() {
1629 return this.baseConf;
1630 }
1631
1632
1633 public FileSystem getFilesystem() {
1634 return fs.getFileSystem();
1635 }
1636
1637
1638 public HRegionFileSystem getRegionFileSystem() {
1639 return this.fs;
1640 }
1641
1642 @Override
1643 public long getEarliestFlushTimeForAllStores() {
1644 return lastStoreFlushTimeMap.isEmpty() ? Long.MAX_VALUE : Collections.min(lastStoreFlushTimeMap
1645 .values());
1646 }
1647
1648 @Override
1649 public long getOldestHfileTs(boolean majorCompactioOnly) throws IOException {
1650 long result = Long.MAX_VALUE;
1651 for (Store store : getStores()) {
1652 Collection<StoreFile> storeFiles = store.getStorefiles();
1653 if (storeFiles == null) continue;
1654 for (StoreFile file : storeFiles) {
1655 StoreFile.Reader sfReader = file.getReader();
1656 if (sfReader == null) continue;
1657 HFile.Reader reader = sfReader.getHFileReader();
1658 if (reader == null) continue;
1659 if (majorCompactioOnly) {
1660 byte[] val = reader.loadFileInfo().get(StoreFile.MAJOR_COMPACTION_KEY);
1661 if (val == null) continue;
1662 if (val == null || !Bytes.toBoolean(val)) {
1663 continue;
1664 }
1665 }
1666 result = Math.min(result, reader.getFileContext().getFileCreateTime());
1667 }
1668 }
1669 return result == Long.MAX_VALUE ? 0 : result;
1670 }
1671
1672 RegionLoad.Builder setCompleteSequenceId(RegionLoad.Builder regionLoadBldr) {
1673 long lastFlushOpSeqIdLocal = this.lastFlushOpSeqId;
1674 byte[] encodedRegionName = this.getRegionInfo().getEncodedNameAsBytes();
1675 regionLoadBldr.clearStoreCompleteSequenceId();
1676 for (byte[] familyName : this.stores.keySet()) {
1677 long earliest = this.wal.getEarliestMemstoreSeqNum(encodedRegionName, familyName);
1678
1679
1680
1681 long csid = (earliest == HConstants.NO_SEQNUM)? lastFlushOpSeqIdLocal: earliest - 1;
1682 regionLoadBldr.addStoreCompleteSequenceId(StoreSequenceId.
1683 newBuilder().setFamilyName(ByteString.copyFrom(familyName)).setSequenceId(csid).build());
1684 }
1685 return regionLoadBldr.setCompleteSequenceId(getMaxFlushedSeqId());
1686 }
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696 public long getLargestHStoreSize() {
1697 long size = 0;
1698 for (Store h : stores.values()) {
1699 long storeSize = h.getSize();
1700 if (storeSize > size) {
1701 size = storeSize;
1702 }
1703 }
1704 return size;
1705 }
1706
1707
1708
1709
1710 public KeyValue.KVComparator getComparator() {
1711 return this.comparator;
1712 }
1713
1714
1715
1716
1717
1718 protected void doRegionCompactionPrep() throws IOException {
1719 }
1720
1721 @Override
1722 public void triggerMajorCompaction() throws IOException {
1723 for (Store s : getStores()) {
1724 s.triggerMajorCompaction();
1725 }
1726 }
1727
1728 @Override
1729 public void compact(final boolean majorCompaction) throws IOException {
1730 if (majorCompaction) {
1731 triggerMajorCompaction();
1732 }
1733 for (Store s : getStores()) {
1734 CompactionContext compaction = s.requestCompaction();
1735 if (compaction != null) {
1736 CompactionThroughputController controller = null;
1737 if (rsServices != null) {
1738 controller = CompactionThroughputControllerFactory.create(rsServices, conf);
1739 }
1740 if (controller == null) {
1741 controller = NoLimitCompactionThroughputController.INSTANCE;
1742 }
1743 compact(compaction, s, controller, null);
1744 }
1745 }
1746 }
1747
1748
1749
1750
1751
1752
1753
1754 public void compactStores() throws IOException {
1755 for (Store s : getStores()) {
1756 CompactionContext compaction = s.requestCompaction();
1757 if (compaction != null) {
1758 compact(compaction, s, NoLimitCompactionThroughputController.INSTANCE, null);
1759 }
1760 }
1761 }
1762
1763
1764
1765
1766
1767
1768
1769 @VisibleForTesting
1770 void compactStore(byte[] family, CompactionThroughputController throughputController)
1771 throws IOException {
1772 Store s = getStore(family);
1773 CompactionContext compaction = s.requestCompaction();
1774 if (compaction != null) {
1775 compact(compaction, s, throughputController, null);
1776 }
1777 }
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794 public boolean compact(CompactionContext compaction, Store store,
1795 CompactionThroughputController throughputController) throws IOException {
1796 return compact(compaction, store, throughputController, null);
1797 }
1798
1799 public boolean compact(CompactionContext compaction, Store store,
1800 CompactionThroughputController throughputController, User user) throws IOException {
1801 assert compaction != null && compaction.hasSelection();
1802 assert !compaction.getRequest().getFiles().isEmpty();
1803 if (this.closing.get() || this.closed.get()) {
1804 LOG.debug("Skipping compaction on " + this + " because closing/closed");
1805 store.cancelRequestedCompaction(compaction);
1806 return false;
1807 }
1808 MonitoredTask status = null;
1809 boolean requestNeedsCancellation = true;
1810
1811 lock.readLock().lock();
1812 try {
1813 byte[] cf = Bytes.toBytes(store.getColumnFamilyName());
1814 if (stores.get(cf) != store) {
1815 LOG.warn("Store " + store.getColumnFamilyName() + " on region " + this
1816 + " has been re-instantiated, cancel this compaction request. "
1817 + " It may be caused by the roll back of split transaction");
1818 return false;
1819 }
1820
1821 status = TaskMonitor.get().createStatus("Compacting " + store + " in " + this);
1822 if (this.closed.get()) {
1823 String msg = "Skipping compaction on " + this + " because closed";
1824 LOG.debug(msg);
1825 status.abort(msg);
1826 return false;
1827 }
1828 boolean wasStateSet = false;
1829 try {
1830 synchronized (writestate) {
1831 if (writestate.writesEnabled) {
1832 wasStateSet = true;
1833 writestate.compacting.incrementAndGet();
1834 } else {
1835 String msg = "NOT compacting region " + this + ". Writes disabled.";
1836 LOG.info(msg);
1837 status.abort(msg);
1838 return false;
1839 }
1840 }
1841 LOG.info("Starting compaction on " + store + " in region " + this
1842 + (compaction.getRequest().isOffPeak()?" as an off-peak compaction":""));
1843 doRegionCompactionPrep();
1844 try {
1845 status.setStatus("Compacting store " + store);
1846
1847
1848 requestNeedsCancellation = false;
1849 store.compact(compaction, throughputController, user);
1850 } catch (InterruptedIOException iioe) {
1851 String msg = "compaction interrupted";
1852 LOG.info(msg, iioe);
1853 status.abort(msg);
1854 return false;
1855 }
1856 } finally {
1857 if (wasStateSet) {
1858 synchronized (writestate) {
1859 writestate.compacting.decrementAndGet();
1860 if (writestate.compacting.get() <= 0) {
1861 writestate.notifyAll();
1862 }
1863 }
1864 }
1865 }
1866 status.markComplete("Compaction complete");
1867 return true;
1868 } finally {
1869 try {
1870 if (requestNeedsCancellation) store.cancelRequestedCompaction(compaction);
1871 if (status != null) status.cleanup();
1872 } finally {
1873 lock.readLock().unlock();
1874 }
1875 }
1876 }
1877
1878 @Override
1879 public FlushResult flush(boolean force) throws IOException {
1880 return flushcache(force, false);
1881 }
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
1892
1893
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
1905 public FlushResult flushcache(boolean forceFlushAllStores, boolean writeFlushRequestWalMarker)
1906 throws IOException {
1907
1908 if (this.closing.get()) {
1909 String msg = "Skipping flush on " + this + " because closing";
1910 LOG.debug(msg);
1911 return new FlushResultImpl(FlushResult.Result.CANNOT_FLUSH, msg, false);
1912 }
1913 MonitoredTask status = TaskMonitor.get().createStatus("Flushing " + this);
1914 status.setStatus("Acquiring readlock on region");
1915
1916 lock.readLock().lock();
1917 try {
1918 if (this.closed.get()) {
1919 String msg = "Skipping flush on " + this + " because closed";
1920 LOG.debug(msg);
1921 status.abort(msg);
1922 return new FlushResultImpl(FlushResult.Result.CANNOT_FLUSH, msg, false);
1923 }
1924 if (coprocessorHost != null) {
1925 status.setStatus("Running coprocessor pre-flush hooks");
1926 coprocessorHost.preFlush();
1927 }
1928
1929
1930 if (numMutationsWithoutWAL.get() > 0) {
1931 numMutationsWithoutWAL.set(0);
1932 dataInMemoryWithoutWAL.set(0);
1933 }
1934 synchronized (writestate) {
1935 if (!writestate.flushing && writestate.writesEnabled) {
1936 this.writestate.flushing = true;
1937 } else {
1938 if (LOG.isDebugEnabled()) {
1939 LOG.debug("NOT flushing memstore for region " + this
1940 + ", flushing=" + writestate.flushing + ", writesEnabled="
1941 + writestate.writesEnabled);
1942 }
1943 String msg = "Not flushing since "
1944 + (writestate.flushing ? "already flushing"
1945 : "writes not enabled");
1946 status.abort(msg);
1947 return new FlushResultImpl(FlushResult.Result.CANNOT_FLUSH, msg, false);
1948 }
1949 }
1950
1951 try {
1952 Collection<Store> specificStoresToFlush =
1953 forceFlushAllStores ? stores.values() : flushPolicy.selectStoresToFlush();
1954 FlushResult fs = internalFlushcache(specificStoresToFlush,
1955 status, writeFlushRequestWalMarker);
1956
1957 if (coprocessorHost != null) {
1958 status.setStatus("Running post-flush coprocessor hooks");
1959 coprocessorHost.postFlush();
1960 }
1961
1962 status.markComplete("Flush successful");
1963 return fs;
1964 } finally {
1965 synchronized (writestate) {
1966 writestate.flushing = false;
1967 this.writestate.flushRequested = false;
1968 writestate.notifyAll();
1969 }
1970 }
1971 } finally {
1972 lock.readLock().unlock();
1973 status.cleanup();
1974 }
1975 }
1976
1977
1978
1979
1980
1981
1982
1983
1984 boolean shouldFlushStore(Store store) {
1985 long earliest = this.wal.getEarliestMemstoreSeqNum(getRegionInfo().getEncodedNameAsBytes(),
1986 store.getFamily().getName()) - 1;
1987 if (earliest > 0 && earliest + flushPerChanges < mvcc.getReadPoint()) {
1988 if (LOG.isDebugEnabled()) {
1989 LOG.debug("Flush column family " + store.getColumnFamilyName() + " of " +
1990 getRegionInfo().getEncodedName() + " because unflushed sequenceid=" + earliest +
1991 " is > " + this.flushPerChanges + " from current=" + mvcc.getReadPoint());
1992 }
1993 return true;
1994 }
1995 if (this.flushCheckInterval <= 0) {
1996 return false;
1997 }
1998 long now = EnvironmentEdgeManager.currentTime();
1999 if (store.timeOfOldestEdit() < now - this.flushCheckInterval) {
2000 if (LOG.isDebugEnabled()) {
2001 LOG.debug("Flush column family: " + store.getColumnFamilyName() + " of " +
2002 getRegionInfo().getEncodedName() + " because time of oldest edit=" +
2003 store.timeOfOldestEdit() + " is > " + this.flushCheckInterval + " from now =" + now);
2004 }
2005 return true;
2006 }
2007 return false;
2008 }
2009
2010
2011
2012
2013 boolean shouldFlush(final StringBuffer whyFlush) {
2014 whyFlush.setLength(0);
2015
2016 if (this.maxFlushedSeqId > 0
2017 && (this.maxFlushedSeqId + this.flushPerChanges < this.mvcc.getReadPoint())) {
2018 whyFlush.append("more than max edits, " + this.flushPerChanges + ", since last flush");
2019 return true;
2020 }
2021 long modifiedFlushCheckInterval = flushCheckInterval;
2022 if (getRegionInfo().isSystemTable() &&
2023 getRegionInfo().getReplicaId() == HRegionInfo.DEFAULT_REPLICA_ID) {
2024 modifiedFlushCheckInterval = SYSTEM_CACHE_FLUSH_INTERVAL;
2025 }
2026 if (modifiedFlushCheckInterval <= 0) {
2027 return false;
2028 }
2029 long now = EnvironmentEdgeManager.currentTime();
2030
2031 if ((now - getEarliestFlushTimeForAllStores() < modifiedFlushCheckInterval)) {
2032 return false;
2033 }
2034
2035
2036 for (Store s : getStores()) {
2037 if (s.timeOfOldestEdit() < now - modifiedFlushCheckInterval) {
2038
2039 whyFlush.append(s.toString() + " has an old edit so flush to free WALs");
2040 return true;
2041 }
2042 }
2043 return false;
2044 }
2045
2046
2047
2048
2049
2050
2051 private FlushResult internalFlushcache(MonitoredTask status)
2052 throws IOException {
2053 return internalFlushcache(stores.values(), status, false);
2054 }
2055
2056
2057
2058
2059
2060
2061 private FlushResult internalFlushcache(final Collection<Store> storesToFlush,
2062 MonitoredTask status, boolean writeFlushWalMarker) throws IOException {
2063 return internalFlushcache(this.wal, HConstants.NO_SEQNUM, storesToFlush,
2064 status, writeFlushWalMarker);
2065 }
2066
2067
2068
2069
2070
2071
2072
2073
2074
2075
2076
2077
2078
2079
2080
2081
2082
2083
2084
2085
2086
2087
2088
2089
2090
2091
2092
2093
2094
2095 protected FlushResult internalFlushcache(final WAL wal, final long myseqid,
2096 final Collection<Store> storesToFlush, MonitoredTask status, boolean writeFlushWalMarker)
2097 throws IOException {
2098 PrepareFlushResult result
2099 = internalPrepareFlushCache(wal, myseqid, storesToFlush, status, writeFlushWalMarker);
2100 if (result.result == null) {
2101 return internalFlushCacheAndCommit(wal, status, result, storesToFlush);
2102 } else {
2103 return result.result;
2104 }
2105 }
2106
2107 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="DLS_DEAD_LOCAL_STORE",
2108 justification="FindBugs seems confused about trxId")
2109 protected PrepareFlushResult internalPrepareFlushCache(final WAL wal, final long myseqid,
2110 final Collection<Store> storesToFlush, MonitoredTask status, boolean writeFlushWalMarker)
2111 throws IOException {
2112 if (this.rsServices != null && this.rsServices.isAborted()) {
2113
2114 throw new IOException("Aborting flush because server is aborted...");
2115 }
2116 final long startTime = EnvironmentEdgeManager.currentTime();
2117
2118 if (this.memstoreSize.get() <= 0) {
2119
2120
2121 MultiVersionConcurrencyControl.WriteEntry writeEntry = null;
2122 this.updatesLock.writeLock().lock();
2123 try {
2124 if (this.memstoreSize.get() <= 0) {
2125
2126
2127
2128
2129
2130
2131 if (wal != null) {
2132 writeEntry = mvcc.begin();
2133 long flushOpSeqId = writeEntry.getWriteNumber();
2134 FlushResult flushResult = new FlushResultImpl(
2135 FlushResult.Result.CANNOT_FLUSH_MEMSTORE_EMPTY,
2136 flushOpSeqId,
2137 "Nothing to flush",
2138 writeFlushRequestMarkerToWAL(wal, writeFlushWalMarker));
2139
2140
2141 mvcc.completeAndWait(writeEntry);
2142 writeEntry = null;
2143 return new PrepareFlushResult(flushResult, myseqid);
2144 } else {
2145 return new PrepareFlushResult(
2146 new FlushResultImpl(
2147 FlushResult.Result.CANNOT_FLUSH_MEMSTORE_EMPTY,
2148 "Nothing to flush",
2149 false),
2150 myseqid);
2151 }
2152 }
2153 } finally {
2154 this.updatesLock.writeLock().unlock();
2155 if (writeEntry != null) {
2156 mvcc.complete(writeEntry);
2157 }
2158 }
2159 }
2160
2161 if (LOG.isInfoEnabled()) {
2162
2163 StringBuilder perCfExtras = null;
2164 if (!isAllFamilies(storesToFlush)) {
2165 perCfExtras = new StringBuilder();
2166 for (Store store: storesToFlush) {
2167 perCfExtras.append("; ").append(store.getColumnFamilyName());
2168 perCfExtras.append("=").append(StringUtils.byteDesc(store.getMemStoreSize()));
2169 }
2170 }
2171 LOG.info("Flushing " + + storesToFlush.size() + "/" + stores.size() +
2172 " column families, memstore=" + StringUtils.byteDesc(this.memstoreSize.get()) +
2173 ((perCfExtras != null && perCfExtras.length() > 0)? perCfExtras.toString(): "") +
2174 ((wal != null) ? "" : "; WAL is null, using passed sequenceid=" + myseqid));
2175 }
2176
2177
2178
2179
2180
2181
2182
2183 status.setStatus("Obtaining lock to block concurrent updates");
2184
2185 this.updatesLock.writeLock().lock();
2186 status.setStatus("Preparing to flush by snapshotting stores in " +
2187 getRegionInfo().getEncodedName());
2188 long totalFlushableSizeOfFlushableStores = 0;
2189
2190 Set<byte[]> flushedFamilyNames = new HashSet<byte[]>();
2191 for (Store store: storesToFlush) {
2192 flushedFamilyNames.add(store.getFamily().getName());
2193 }
2194
2195 TreeMap<byte[], StoreFlushContext> storeFlushCtxs
2196 = new TreeMap<byte[], StoreFlushContext>(Bytes.BYTES_COMPARATOR);
2197 TreeMap<byte[], List<Path>> committedFiles = new TreeMap<byte[], List<Path>>(
2198 Bytes.BYTES_COMPARATOR);
2199 TreeMap<byte[], Long> storeFlushableSize
2200 = new TreeMap<byte[], Long>(Bytes.BYTES_COMPARATOR);
2201
2202
2203
2204 long flushOpSeqId = HConstants.NO_SEQNUM;
2205
2206
2207 long flushedSeqId = HConstants.NO_SEQNUM;
2208 byte[] encodedRegionName = getRegionInfo().getEncodedNameAsBytes();
2209
2210 long trxId = 0;
2211 MultiVersionConcurrencyControl.WriteEntry writeEntry = mvcc.begin();
2212 try {
2213 try {
2214 if (wal != null) {
2215 Long earliestUnflushedSequenceIdForTheRegion =
2216 wal.startCacheFlush(encodedRegionName, flushedFamilyNames);
2217 if (earliestUnflushedSequenceIdForTheRegion == null) {
2218
2219 String msg = this.getRegionInfo().getEncodedName() + " flush aborted; WAL closing.";
2220 status.setStatus(msg);
2221 return new PrepareFlushResult(
2222 new FlushResultImpl(FlushResult.Result.CANNOT_FLUSH, msg, false),
2223 myseqid);
2224 }
2225 flushOpSeqId = getNextSequenceId(wal);
2226
2227 flushedSeqId =
2228 earliestUnflushedSequenceIdForTheRegion.longValue() == HConstants.NO_SEQNUM?
2229 flushOpSeqId: earliestUnflushedSequenceIdForTheRegion.longValue() - 1;
2230 } else {
2231
2232 flushedSeqId = flushOpSeqId = myseqid;
2233 }
2234
2235 for (Store s : storesToFlush) {
2236 totalFlushableSizeOfFlushableStores += s.getFlushableSize();
2237 storeFlushCtxs.put(s.getFamily().getName(), s.createFlushContext(flushOpSeqId));
2238 committedFiles.put(s.getFamily().getName(), null);
2239 storeFlushableSize.put(s.getFamily().getName(), s.getFlushableSize());
2240 }
2241
2242
2243 if (wal != null && !writestate.readOnly) {
2244 FlushDescriptor desc = ProtobufUtil.toFlushDescriptor(FlushAction.START_FLUSH,
2245 getRegionInfo(), flushOpSeqId, committedFiles);
2246
2247 trxId = WALUtil.writeFlushMarker(wal, this.htableDescriptor, getRegionInfo(),
2248 desc, false, mvcc);
2249 }
2250
2251
2252 for (StoreFlushContext flush : storeFlushCtxs.values()) {
2253 flush.prepare();
2254 }
2255 } catch (IOException ex) {
2256 if (wal != null) {
2257 if (trxId > 0) {
2258 try {
2259 FlushDescriptor desc = ProtobufUtil.toFlushDescriptor(FlushAction.ABORT_FLUSH,
2260 getRegionInfo(), flushOpSeqId, committedFiles);
2261 WALUtil.writeFlushMarker(wal, this.htableDescriptor, getRegionInfo(),
2262 desc, false, mvcc);
2263 } catch (Throwable t) {
2264 LOG.warn("Received unexpected exception trying to write ABORT_FLUSH marker to WAL:" +
2265 StringUtils.stringifyException(t));
2266
2267 }
2268 }
2269
2270 wal.abortCacheFlush(this.getRegionInfo().getEncodedNameAsBytes());
2271 throw ex;
2272 }
2273 } finally {
2274 this.updatesLock.writeLock().unlock();
2275 }
2276 String s = "Finished memstore snapshotting " + this +
2277 ", syncing WAL and waiting on mvcc, flushsize=" + totalFlushableSizeOfFlushableStores;
2278 status.setStatus(s);
2279 if (LOG.isTraceEnabled()) LOG.trace(s);
2280
2281
2282 if (wal != null) {
2283 try {
2284 wal.sync();
2285 } catch (IOException ioe) {
2286 wal.abortCacheFlush(this.getRegionInfo().getEncodedNameAsBytes());
2287 throw ioe;
2288 }
2289 }
2290
2291
2292
2293
2294
2295
2296 mvcc.completeAndWait(writeEntry);
2297
2298
2299 writeEntry = null;
2300 } finally {
2301 if (writeEntry != null) {
2302
2303 mvcc.complete(writeEntry);
2304 }
2305 }
2306 return new PrepareFlushResult(storeFlushCtxs, committedFiles, storeFlushableSize, startTime,
2307 flushOpSeqId, flushedSeqId, totalFlushableSizeOfFlushableStores);
2308 }
2309
2310
2311
2312
2313
2314 private boolean isAllFamilies(final Collection<Store> families) {
2315 return families == null || this.stores.size() == families.size();
2316 }
2317
2318
2319
2320
2321
2322
2323
2324 private boolean writeFlushRequestMarkerToWAL(WAL wal, boolean writeFlushWalMarker) {
2325 if (writeFlushWalMarker && wal != null && !writestate.readOnly) {
2326 FlushDescriptor desc = ProtobufUtil.toFlushDescriptor(FlushAction.CANNOT_FLUSH,
2327 getRegionInfo(), -1, new TreeMap<byte[], List<Path>>(Bytes.BYTES_COMPARATOR));
2328 try {
2329 WALUtil.writeFlushMarker(wal, this.htableDescriptor, getRegionInfo(),
2330 desc, true, mvcc);
2331 return true;
2332 } catch (IOException e) {
2333 LOG.warn(getRegionInfo().getEncodedName() + " : "
2334 + "Received exception while trying to write the flush request to wal", e);
2335 }
2336 }
2337 return false;
2338 }
2339
2340 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="NN_NAKED_NOTIFY",
2341 justification="Intentional; notify is about completed flush")
2342 protected FlushResult internalFlushCacheAndCommit(
2343 final WAL wal, MonitoredTask status, final PrepareFlushResult prepareResult,
2344 final Collection<Store> storesToFlush)
2345 throws IOException {
2346
2347
2348 TreeMap<byte[], StoreFlushContext> storeFlushCtxs = prepareResult.storeFlushCtxs;
2349 TreeMap<byte[], List<Path>> committedFiles = prepareResult.committedFiles;
2350 long startTime = prepareResult.startTime;
2351 long flushOpSeqId = prepareResult.flushOpSeqId;
2352 long flushedSeqId = prepareResult.flushedSeqId;
2353 long totalFlushableSizeOfFlushableStores = prepareResult.totalFlushableSize;
2354
2355 String s = "Flushing stores of " + this;
2356 status.setStatus(s);
2357 if (LOG.isTraceEnabled()) LOG.trace(s);
2358
2359
2360
2361
2362
2363 boolean compactionRequested = false;
2364 try {
2365
2366
2367
2368
2369
2370 for (StoreFlushContext flush : storeFlushCtxs.values()) {
2371 flush.flushCache(status);
2372 }
2373
2374
2375
2376 Iterator<Store> it = storesToFlush.iterator();
2377
2378 for (StoreFlushContext flush : storeFlushCtxs.values()) {
2379 boolean needsCompaction = flush.commit(status);
2380 if (needsCompaction) {
2381 compactionRequested = true;
2382 }
2383 byte[] storeName = it.next().getFamily().getName();
2384 List<Path> storeCommittedFiles = flush.getCommittedFiles();
2385 committedFiles.put(storeName, storeCommittedFiles);
2386
2387 if (storeCommittedFiles == null || storeCommittedFiles.isEmpty()) {
2388 totalFlushableSizeOfFlushableStores -= prepareResult.storeFlushableSize.get(storeName);
2389 }
2390 }
2391 storeFlushCtxs.clear();
2392
2393
2394 this.addAndGetGlobalMemstoreSize(-totalFlushableSizeOfFlushableStores);
2395
2396 if (wal != null) {
2397
2398 FlushDescriptor desc = ProtobufUtil.toFlushDescriptor(FlushAction.COMMIT_FLUSH,
2399 getRegionInfo(), flushOpSeqId, committedFiles);
2400 WALUtil.writeFlushMarker(wal, this.htableDescriptor, getRegionInfo(),
2401 desc, true, mvcc);
2402 }
2403 } catch (Throwable t) {
2404
2405
2406
2407
2408
2409
2410 if (wal != null) {
2411 try {
2412 FlushDescriptor desc = ProtobufUtil.toFlushDescriptor(FlushAction.ABORT_FLUSH,
2413 getRegionInfo(), flushOpSeqId, committedFiles);
2414 WALUtil.writeFlushMarker(wal, this.htableDescriptor, getRegionInfo(),
2415 desc, false, mvcc);
2416 } catch (Throwable ex) {
2417 LOG.warn(getRegionInfo().getEncodedName() + " : "
2418 + "failed writing ABORT_FLUSH marker to WAL", ex);
2419
2420 }
2421 wal.abortCacheFlush(this.getRegionInfo().getEncodedNameAsBytes());
2422 }
2423 DroppedSnapshotException dse = new DroppedSnapshotException("region: " +
2424 Bytes.toStringBinary(getRegionInfo().getRegionName()));
2425 dse.initCause(t);
2426 status.abort("Flush failed: " + StringUtils.stringifyException(t));
2427
2428
2429
2430
2431
2432 this.closing.set(true);
2433
2434 if (rsServices != null) {
2435
2436 rsServices.abort("Replay of WAL required. Forcing server shutdown", dse);
2437 }
2438
2439 throw dse;
2440 }
2441
2442
2443 if (wal != null) {
2444 wal.completeCacheFlush(this.getRegionInfo().getEncodedNameAsBytes());
2445 }
2446
2447
2448 for (Store store: storesToFlush) {
2449 this.lastStoreFlushTimeMap.put(store, startTime);
2450 }
2451
2452 this.maxFlushedSeqId = flushedSeqId;
2453 this.lastFlushOpSeqId = flushOpSeqId;
2454
2455
2456
2457 synchronized (this) {
2458 notifyAll();
2459 }
2460
2461 long time = EnvironmentEdgeManager.currentTime() - startTime;
2462 long memstoresize = this.memstoreSize.get();
2463 String msg = "Finished memstore flush of ~"
2464 + StringUtils.byteDesc(totalFlushableSizeOfFlushableStores) + "/"
2465 + totalFlushableSizeOfFlushableStores + ", currentsize="
2466 + StringUtils.byteDesc(memstoresize) + "/" + memstoresize
2467 + " for region " + this + " in " + time + "ms, sequenceid="
2468 + flushOpSeqId + ", compaction requested=" + compactionRequested
2469 + ((wal == null) ? "; wal=null" : "");
2470 LOG.info(msg);
2471 status.setStatus(msg);
2472
2473 return new FlushResultImpl(compactionRequested ?
2474 FlushResult.Result.FLUSHED_COMPACTION_NEEDED :
2475 FlushResult.Result.FLUSHED_NO_COMPACTION_NEEDED,
2476 flushOpSeqId);
2477 }
2478
2479
2480
2481
2482
2483
2484 @VisibleForTesting
2485 protected long getNextSequenceId(final WAL wal) throws IOException {
2486
2487
2488
2489
2490
2491
2492 WALKey key = this.appendEmptyEdit(wal);
2493 mvcc.complete(key.getWriteEntry());
2494 return key.getSequenceId(this.maxWaitForSeqId);
2495 }
2496
2497
2498
2499
2500
2501 @Override
2502 public Result getClosestRowBefore(final byte [] row, final byte [] family) throws IOException {
2503 if (coprocessorHost != null) {
2504 Result result = new Result();
2505 if (coprocessorHost.preGetClosestRowBefore(row, family, result)) {
2506 return result;
2507 }
2508 }
2509
2510
2511 checkRow(row, "getClosestRowBefore");
2512 startRegionOperation(Operation.GET);
2513 this.readRequestsCount.increment();
2514 try {
2515 Store store = getStore(family);
2516
2517 Cell key = store.getRowKeyAtOrBefore(row);
2518 Result result = null;
2519 if (key != null) {
2520 Get get = new Get(CellUtil.cloneRow(key));
2521 get.addFamily(family);
2522 result = get(get);
2523 }
2524 if (coprocessorHost != null) {
2525 coprocessorHost.postGetClosestRowBefore(row, family, result);
2526 }
2527 return result;
2528 } finally {
2529 closeRegionOperation(Operation.GET);
2530 }
2531 }
2532
2533 @Override
2534 public RegionScanner getScanner(Scan scan) throws IOException {
2535 return getScanner(scan, null);
2536 }
2537
2538 @Override
2539 public RegionScanner getScanner(Scan scan,
2540 List<KeyValueScanner> additionalScanners) throws IOException {
2541 startRegionOperation(Operation.SCAN);
2542 try {
2543
2544 if (!scan.hasFamilies()) {
2545
2546 for (byte[] family: this.htableDescriptor.getFamiliesKeys()) {
2547 scan.addFamily(family);
2548 }
2549 } else {
2550 for (byte [] family : scan.getFamilyMap().keySet()) {
2551 checkFamily(family);
2552 }
2553 }
2554 return instantiateRegionScanner(scan, additionalScanners);
2555 } finally {
2556 closeRegionOperation(Operation.SCAN);
2557 }
2558 }
2559
2560 protected RegionScanner instantiateRegionScanner(Scan scan,
2561 List<KeyValueScanner> additionalScanners) throws IOException {
2562 if (scan.isReversed()) {
2563 if (scan.getFilter() != null) {
2564 scan.getFilter().setReversed(true);
2565 }
2566 return new ReversedRegionScannerImpl(scan, additionalScanners, this);
2567 }
2568 return new RegionScannerImpl(scan, additionalScanners, this);
2569 }
2570
2571 @Override
2572 public void prepareDelete(Delete delete) throws IOException {
2573
2574 if(delete.getFamilyCellMap().isEmpty()){
2575 for(byte [] family : this.htableDescriptor.getFamiliesKeys()){
2576
2577 delete.addFamily(family, delete.getTimeStamp());
2578 }
2579 } else {
2580 for(byte [] family : delete.getFamilyCellMap().keySet()) {
2581 if(family == null) {
2582 throw new NoSuchColumnFamilyException("Empty family is invalid");
2583 }
2584 checkFamily(family);
2585 }
2586 }
2587 }
2588
2589 @Override
2590 public void delete(Delete delete) throws IOException {
2591 checkReadOnly();
2592 checkResources();
2593 startRegionOperation(Operation.DELETE);
2594 try {
2595 delete.getRow();
2596
2597 doBatchMutate(delete);
2598 } finally {
2599 closeRegionOperation(Operation.DELETE);
2600 }
2601 }
2602
2603
2604
2605
2606 private static final byte [] FOR_UNIT_TESTS_ONLY = Bytes.toBytes("ForUnitTestsOnly");
2607
2608
2609
2610
2611
2612
2613 void delete(NavigableMap<byte[], List<Cell>> familyMap,
2614 Durability durability) throws IOException {
2615 Delete delete = new Delete(FOR_UNIT_TESTS_ONLY);
2616 delete.setFamilyCellMap(familyMap);
2617 delete.setDurability(durability);
2618 doBatchMutate(delete);
2619 }
2620
2621 @Override
2622 public void prepareDeleteTimestamps(Mutation mutation, Map<byte[], List<Cell>> familyMap,
2623 byte[] byteNow) throws IOException {
2624 for (Map.Entry<byte[], List<Cell>> e : familyMap.entrySet()) {
2625
2626 byte[] family = e.getKey();
2627 List<Cell> cells = e.getValue();
2628 assert cells instanceof RandomAccess;
2629
2630 Map<byte[], Integer> kvCount = new TreeMap<byte[], Integer>(Bytes.BYTES_COMPARATOR);
2631 int listSize = cells.size();
2632 for (int i=0; i < listSize; i++) {
2633 Cell cell = cells.get(i);
2634
2635
2636 if (cell.getTimestamp() == HConstants.LATEST_TIMESTAMP && CellUtil.isDeleteType(cell)) {
2637 byte[] qual = CellUtil.cloneQualifier(cell);
2638 if (qual == null) qual = HConstants.EMPTY_BYTE_ARRAY;
2639
2640 Integer count = kvCount.get(qual);
2641 if (count == null) {
2642 kvCount.put(qual, 1);
2643 } else {
2644 kvCount.put(qual, count + 1);
2645 }
2646 count = kvCount.get(qual);
2647
2648 Get get = new Get(CellUtil.cloneRow(cell));
2649 get.setMaxVersions(count);
2650 get.addColumn(family, qual);
2651 if (coprocessorHost != null) {
2652 if (!coprocessorHost.prePrepareTimeStampForDeleteVersion(mutation, cell,
2653 byteNow, get)) {
2654 updateDeleteLatestVersionTimeStamp(cell, get, count, byteNow);
2655 }
2656 } else {
2657 updateDeleteLatestVersionTimeStamp(cell, get, count, byteNow);
2658 }
2659 } else {
2660 CellUtil.updateLatestStamp(cell, byteNow, 0);
2661 }
2662 }
2663 }
2664 }
2665
2666 void updateDeleteLatestVersionTimeStamp(Cell cell, Get get, int count, byte[] byteNow)
2667 throws IOException {
2668 List<Cell> result = get(get, false);
2669
2670 if (result.size() < count) {
2671
2672 CellUtil.updateLatestStamp(cell, byteNow, 0);
2673 return;
2674 }
2675 if (result.size() > count) {
2676 throw new RuntimeException("Unexpected size: " + result.size());
2677 }
2678 Cell getCell = result.get(count - 1);
2679 CellUtil.setTimestamp(cell, getCell.getTimestamp());
2680 }
2681
2682 @Override
2683 public void put(Put put) throws IOException {
2684 checkReadOnly();
2685
2686
2687
2688
2689
2690 checkResources();
2691 startRegionOperation(Operation.PUT);
2692 try {
2693
2694 doBatchMutate(put);
2695 } finally {
2696 closeRegionOperation(Operation.PUT);
2697 }
2698 }
2699
2700
2701
2702
2703
2704
2705 private abstract static class BatchOperationInProgress<T> {
2706 T[] operations;
2707 int nextIndexToProcess = 0;
2708 OperationStatus[] retCodeDetails;
2709 WALEdit[] walEditsFromCoprocessors;
2710
2711 public BatchOperationInProgress(T[] operations) {
2712 this.operations = operations;
2713 this.retCodeDetails = new OperationStatus[operations.length];
2714 this.walEditsFromCoprocessors = new WALEdit[operations.length];
2715 Arrays.fill(this.retCodeDetails, OperationStatus.NOT_RUN);
2716 }
2717
2718 public abstract Mutation getMutation(int index);
2719 public abstract long getNonceGroup(int index);
2720 public abstract long getNonce(int index);
2721
2722 public abstract Mutation[] getMutationsForCoprocs();
2723 public abstract boolean isInReplay();
2724 public abstract long getReplaySequenceId();
2725
2726 public boolean isDone() {
2727 return nextIndexToProcess == operations.length;
2728 }
2729 }
2730
2731 private static class MutationBatch extends BatchOperationInProgress<Mutation> {
2732 private long nonceGroup;
2733 private long nonce;
2734 public MutationBatch(Mutation[] operations, long nonceGroup, long nonce) {
2735 super(operations);
2736 this.nonceGroup = nonceGroup;
2737 this.nonce = nonce;
2738 }
2739
2740 @Override
2741 public Mutation getMutation(int index) {
2742 return this.operations[index];
2743 }
2744
2745 @Override
2746 public long getNonceGroup(int index) {
2747 return nonceGroup;
2748 }
2749
2750 @Override
2751 public long getNonce(int index) {
2752 return nonce;
2753 }
2754
2755 @Override
2756 public Mutation[] getMutationsForCoprocs() {
2757 return this.operations;
2758 }
2759
2760 @Override
2761 public boolean isInReplay() {
2762 return false;
2763 }
2764
2765 @Override
2766 public long getReplaySequenceId() {
2767 return 0;
2768 }
2769 }
2770
2771 private static class ReplayBatch extends BatchOperationInProgress<MutationReplay> {
2772 private long replaySeqId = 0;
2773 public ReplayBatch(MutationReplay[] operations, long seqId) {
2774 super(operations);
2775 this.replaySeqId = seqId;
2776 }
2777
2778 @Override
2779 public Mutation getMutation(int index) {
2780 return this.operations[index].mutation;
2781 }
2782
2783 @Override
2784 public long getNonceGroup(int index) {
2785 return this.operations[index].nonceGroup;
2786 }
2787
2788 @Override
2789 public long getNonce(int index) {
2790 return this.operations[index].nonce;
2791 }
2792
2793 @Override
2794 public Mutation[] getMutationsForCoprocs() {
2795 assert false;
2796 throw new RuntimeException("Should not be called for replay batch");
2797 }
2798
2799 @Override
2800 public boolean isInReplay() {
2801 return true;
2802 }
2803
2804 @Override
2805 public long getReplaySequenceId() {
2806 return this.replaySeqId;
2807 }
2808 }
2809
2810 @Override
2811 public OperationStatus[] batchMutate(Mutation[] mutations, long nonceGroup, long nonce)
2812 throws IOException {
2813
2814
2815
2816
2817 return batchMutate(new MutationBatch(mutations, nonceGroup, nonce));
2818 }
2819
2820 public OperationStatus[] batchMutate(Mutation[] mutations) throws IOException {
2821 return batchMutate(mutations, HConstants.NO_NONCE, HConstants.NO_NONCE);
2822 }
2823
2824 @Override
2825 public OperationStatus[] batchReplay(MutationReplay[] mutations, long replaySeqId)
2826 throws IOException {
2827 if (!RegionReplicaUtil.isDefaultReplica(getRegionInfo())
2828 && replaySeqId < lastReplayedOpenRegionSeqId) {
2829
2830
2831 if (LOG.isTraceEnabled()) {
2832 LOG.trace(getRegionInfo().getEncodedName() + " : "
2833 + "Skipping " + mutations.length + " mutations with replaySeqId=" + replaySeqId
2834 + " which is < than lastReplayedOpenRegionSeqId=" + lastReplayedOpenRegionSeqId);
2835 for (MutationReplay mut : mutations) {
2836 LOG.trace(getRegionInfo().getEncodedName() + " : Skipping : " + mut.mutation);
2837 }
2838 }
2839
2840 OperationStatus[] statuses = new OperationStatus[mutations.length];
2841 for (int i = 0; i < statuses.length; i++) {
2842 statuses[i] = OperationStatus.SUCCESS;
2843 }
2844 return statuses;
2845 }
2846 return batchMutate(new ReplayBatch(mutations, replaySeqId));
2847 }
2848
2849
2850
2851
2852
2853
2854
2855
2856
2857 OperationStatus[] batchMutate(BatchOperationInProgress<?> batchOp) throws IOException {
2858 boolean initialized = false;
2859 Operation op = batchOp.isInReplay() ? Operation.REPLAY_BATCH_MUTATE : Operation.BATCH_MUTATE;
2860 startRegionOperation(op);
2861 try {
2862 while (!batchOp.isDone()) {
2863 if (!batchOp.isInReplay()) {
2864 checkReadOnly();
2865 }
2866 checkResources();
2867
2868 if (!initialized) {
2869 this.writeRequestsCount.add(batchOp.operations.length);
2870 if (!batchOp.isInReplay()) {
2871 doPreMutationHook(batchOp);
2872 }
2873 initialized = true;
2874 }
2875 long addedSize = doMiniBatchMutation(batchOp);
2876 long newSize = this.addAndGetGlobalMemstoreSize(addedSize);
2877 if (isFlushSize(newSize)) {
2878 requestFlush();
2879 }
2880 }
2881 } finally {
2882 closeRegionOperation(op);
2883 }
2884 return batchOp.retCodeDetails;
2885 }
2886
2887
2888 private void doPreMutationHook(BatchOperationInProgress<?> batchOp)
2889 throws IOException {
2890
2891 WALEdit walEdit = new WALEdit();
2892 if (coprocessorHost != null) {
2893 for (int i = 0 ; i < batchOp.operations.length; i++) {
2894 Mutation m = batchOp.getMutation(i);
2895 if (m instanceof Put) {
2896 if (coprocessorHost.prePut((Put) m, walEdit, m.getDurability())) {
2897
2898
2899 batchOp.retCodeDetails[i] = OperationStatus.SUCCESS;
2900 }
2901 } else if (m instanceof Delete) {
2902 Delete curDel = (Delete) m;
2903 if (curDel.getFamilyCellMap().isEmpty()) {
2904
2905 prepareDelete(curDel);
2906 }
2907 if (coprocessorHost.preDelete(curDel, walEdit, m.getDurability())) {
2908
2909
2910 batchOp.retCodeDetails[i] = OperationStatus.SUCCESS;
2911 }
2912 } else {
2913
2914
2915
2916 batchOp.retCodeDetails[i] = new OperationStatus(OperationStatusCode.FAILURE,
2917 "Put/Delete mutations only supported in batchMutate() now");
2918 }
2919 if (!walEdit.isEmpty()) {
2920 batchOp.walEditsFromCoprocessors[i] = walEdit;
2921 walEdit = new WALEdit();
2922 }
2923 }
2924 }
2925 }
2926
2927 @SuppressWarnings("unchecked")
2928 private long doMiniBatchMutation(BatchOperationInProgress<?> batchOp) throws IOException {
2929 boolean isInReplay = batchOp.isInReplay();
2930
2931 boolean putsCfSetConsistent = true;
2932
2933 Set<byte[]> putsCfSet = null;
2934
2935 boolean deletesCfSetConsistent = true;
2936
2937 Set<byte[]> deletesCfSet = null;
2938
2939 long currentNonceGroup = HConstants.NO_NONCE, currentNonce = HConstants.NO_NONCE;
2940 WALEdit walEdit = new WALEdit(isInReplay);
2941 MultiVersionConcurrencyControl.WriteEntry writeEntry = null;
2942 long txid = 0;
2943 boolean doRollBackMemstore = false;
2944 boolean locked = false;
2945
2946
2947 List<RowLock> acquiredRowLocks = Lists.newArrayListWithCapacity(batchOp.operations.length);
2948
2949 Map<byte[], List<Cell>>[] familyMaps = new Map[batchOp.operations.length];
2950
2951 int firstIndex = batchOp.nextIndexToProcess;
2952 int lastIndexExclusive = firstIndex;
2953 boolean success = false;
2954 int noOfPuts = 0, noOfDeletes = 0;
2955 WALKey walKey = null;
2956 long mvccNum = 0;
2957 try {
2958
2959
2960
2961
2962 int numReadyToWrite = 0;
2963 long now = EnvironmentEdgeManager.currentTime();
2964 while (lastIndexExclusive < batchOp.operations.length) {
2965 Mutation mutation = batchOp.getMutation(lastIndexExclusive);
2966 boolean isPutMutation = mutation instanceof Put;
2967
2968 Map<byte[], List<Cell>> familyMap = mutation.getFamilyCellMap();
2969
2970 familyMaps[lastIndexExclusive] = familyMap;
2971
2972
2973 if (batchOp.retCodeDetails[lastIndexExclusive].getOperationStatusCode()
2974 != OperationStatusCode.NOT_RUN) {
2975 lastIndexExclusive++;
2976 continue;
2977 }
2978
2979 try {
2980 if (isPutMutation) {
2981
2982 if (isInReplay) {
2983 removeNonExistentColumnFamilyForReplay(familyMap);
2984 } else {
2985 checkFamilies(familyMap.keySet());
2986 }
2987 checkTimestamps(mutation.getFamilyCellMap(), now);
2988 } else {
2989 prepareDelete((Delete) mutation);
2990 }
2991 checkRow(mutation.getRow(), "doMiniBatchMutation");
2992 } catch (NoSuchColumnFamilyException nscf) {
2993 LOG.warn("No such column family in batch mutation", nscf);
2994 batchOp.retCodeDetails[lastIndexExclusive] = new OperationStatus(
2995 OperationStatusCode.BAD_FAMILY, nscf.getMessage());
2996 lastIndexExclusive++;
2997 continue;
2998 } catch (FailedSanityCheckException fsce) {
2999 LOG.warn("Batch Mutation did not pass sanity check", fsce);
3000 batchOp.retCodeDetails[lastIndexExclusive] = new OperationStatus(
3001 OperationStatusCode.SANITY_CHECK_FAILURE, fsce.getMessage());
3002 lastIndexExclusive++;
3003 continue;
3004 } catch (WrongRegionException we) {
3005 LOG.warn("Batch mutation had a row that does not belong to this region", we);
3006 batchOp.retCodeDetails[lastIndexExclusive] = new OperationStatus(
3007 OperationStatusCode.SANITY_CHECK_FAILURE, we.getMessage());
3008 lastIndexExclusive++;
3009 continue;
3010 }
3011
3012
3013
3014 RowLock rowLock = null;
3015 try {
3016 rowLock = getRowLock(mutation.getRow(), true);
3017 } catch (IOException ioe) {
3018 LOG.warn("Failed getting lock in batch put, row="
3019 + Bytes.toStringBinary(mutation.getRow()), ioe);
3020 }
3021 if (rowLock == null) {
3022
3023 break;
3024 } else {
3025 acquiredRowLocks.add(rowLock);
3026 }
3027
3028 lastIndexExclusive++;
3029 numReadyToWrite++;
3030
3031 if (isPutMutation) {
3032
3033
3034
3035 if (putsCfSet == null) {
3036 putsCfSet = mutation.getFamilyCellMap().keySet();
3037 } else {
3038 putsCfSetConsistent = putsCfSetConsistent
3039 && mutation.getFamilyCellMap().keySet().equals(putsCfSet);
3040 }
3041 } else {
3042 if (deletesCfSet == null) {
3043 deletesCfSet = mutation.getFamilyCellMap().keySet();
3044 } else {
3045 deletesCfSetConsistent = deletesCfSetConsistent
3046 && mutation.getFamilyCellMap().keySet().equals(deletesCfSet);
3047 }
3048 }
3049 }
3050
3051
3052
3053 now = EnvironmentEdgeManager.currentTime();
3054 byte[] byteNow = Bytes.toBytes(now);
3055
3056
3057 if (numReadyToWrite <= 0) return 0L;
3058
3059
3060
3061
3062
3063
3064 for (int i = firstIndex; !isInReplay && i < lastIndexExclusive; i++) {
3065
3066 if (batchOp.retCodeDetails[i].getOperationStatusCode()
3067 != OperationStatusCode.NOT_RUN) continue;
3068
3069 Mutation mutation = batchOp.getMutation(i);
3070 if (mutation instanceof Put) {
3071 updateCellTimestamps(familyMaps[i].values(), byteNow);
3072 noOfPuts++;
3073 } else {
3074 prepareDeleteTimestamps(mutation, familyMaps[i], byteNow);
3075 noOfDeletes++;
3076 }
3077 rewriteCellTags(familyMaps[i], mutation);
3078 }
3079
3080 lock(this.updatesLock.readLock(), numReadyToWrite);
3081 locked = true;
3082
3083
3084 if (!isInReplay && coprocessorHost != null) {
3085 MiniBatchOperationInProgress<Mutation> miniBatchOp =
3086 new MiniBatchOperationInProgress<Mutation>(batchOp.getMutationsForCoprocs(),
3087 batchOp.retCodeDetails, batchOp.walEditsFromCoprocessors, firstIndex, lastIndexExclusive);
3088 if (coprocessorHost.preBatchMutate(miniBatchOp)) return 0L;
3089 }
3090
3091
3092
3093
3094 Durability durability = Durability.USE_DEFAULT;
3095 for (int i = firstIndex; i < lastIndexExclusive; i++) {
3096
3097 if (batchOp.retCodeDetails[i].getOperationStatusCode() != OperationStatusCode.NOT_RUN) {
3098 continue;
3099 }
3100
3101 Mutation m = batchOp.getMutation(i);
3102 Durability tmpDur = getEffectiveDurability(m.getDurability());
3103 if (tmpDur.ordinal() > durability.ordinal()) {
3104 durability = tmpDur;
3105 }
3106 if (tmpDur == Durability.SKIP_WAL) {
3107 recordMutationWithoutWal(m.getFamilyCellMap());
3108 continue;
3109 }
3110
3111 long nonceGroup = batchOp.getNonceGroup(i), nonce = batchOp.getNonce(i);
3112
3113
3114
3115 if (nonceGroup != currentNonceGroup || nonce != currentNonce) {
3116 if (walEdit.size() > 0) {
3117 assert isInReplay;
3118 if (!isInReplay) {
3119 throw new IOException("Multiple nonces per batch and not in replay");
3120 }
3121
3122
3123 walKey = new ReplayHLogKey(this.getRegionInfo().getEncodedNameAsBytes(),
3124 this.htableDescriptor.getTableName(), now, m.getClusterIds(),
3125 currentNonceGroup, currentNonce, mvcc);
3126 txid = this.wal.append(this.htableDescriptor, this.getRegionInfo(), walKey,
3127 walEdit, true);
3128 walEdit = new WALEdit(isInReplay);
3129 walKey = null;
3130 }
3131 currentNonceGroup = nonceGroup;
3132 currentNonce = nonce;
3133 }
3134
3135
3136 WALEdit fromCP = batchOp.walEditsFromCoprocessors[i];
3137 if (fromCP != null) {
3138 for (Cell cell : fromCP.getCells()) {
3139 walEdit.add(cell);
3140 }
3141 }
3142 addFamilyMapToWALEdit(familyMaps[i], walEdit);
3143 }
3144
3145
3146
3147
3148 Mutation mutation = batchOp.getMutation(firstIndex);
3149 if (isInReplay) {
3150
3151 walKey = new ReplayHLogKey(this.getRegionInfo().getEncodedNameAsBytes(),
3152 this.htableDescriptor.getTableName(), WALKey.NO_SEQUENCE_ID, now,
3153 mutation.getClusterIds(), currentNonceGroup, currentNonce, mvcc);
3154 long replaySeqId = batchOp.getReplaySequenceId();
3155 walKey.setOrigLogSeqNum(replaySeqId);
3156 }
3157 if (walEdit.size() > 0) {
3158 if (!isInReplay) {
3159
3160 walKey = new HLogKey(this.getRegionInfo().getEncodedNameAsBytes(),
3161 this.htableDescriptor.getTableName(), WALKey.NO_SEQUENCE_ID, now,
3162 mutation.getClusterIds(), currentNonceGroup, currentNonce, mvcc);
3163 }
3164 txid = this.wal.append(this.htableDescriptor, this.getRegionInfo(), walKey, walEdit, true);
3165 }
3166
3167
3168
3169 if (walKey == null) {
3170
3171 walKey = this.appendEmptyEdit(this.wal);
3172 }
3173 if (!isInReplay) {
3174 writeEntry = walKey.getWriteEntry();
3175 mvccNum = writeEntry.getWriteNumber();
3176 } else {
3177 mvccNum = batchOp.getReplaySequenceId();
3178 }
3179
3180
3181
3182
3183
3184
3185
3186
3187
3188
3189 long addedSize = 0;
3190 for (int i = firstIndex; i < lastIndexExclusive; i++) {
3191 if (batchOp.retCodeDetails[i].getOperationStatusCode()
3192 != OperationStatusCode.NOT_RUN) {
3193 continue;
3194 }
3195 doRollBackMemstore = true;
3196 addedSize += applyFamilyMapToMemstore(familyMaps[i], mvccNum, isInReplay);
3197 }
3198
3199
3200
3201
3202 if (locked) {
3203 this.updatesLock.readLock().unlock();
3204 locked = false;
3205 }
3206 releaseRowLocks(acquiredRowLocks);
3207
3208
3209
3210
3211 if (txid != 0) {
3212 syncOrDefer(txid, durability);
3213 }
3214
3215 doRollBackMemstore = false;
3216
3217 if (!isInReplay && coprocessorHost != null) {
3218 MiniBatchOperationInProgress<Mutation> miniBatchOp =
3219 new MiniBatchOperationInProgress<Mutation>(batchOp.getMutationsForCoprocs(),
3220 batchOp.retCodeDetails, batchOp.walEditsFromCoprocessors, firstIndex, lastIndexExclusive);
3221 coprocessorHost.postBatchMutate(miniBatchOp);
3222 }
3223
3224
3225
3226
3227 if (writeEntry != null) {
3228 mvcc.completeAndWait(writeEntry);
3229 writeEntry = null;
3230 } else if (isInReplay) {
3231
3232 mvcc.advanceTo(mvccNum);
3233 }
3234
3235 for (int i = firstIndex; i < lastIndexExclusive; i ++) {
3236 if (batchOp.retCodeDetails[i] == OperationStatus.NOT_RUN) {
3237 batchOp.retCodeDetails[i] = OperationStatus.SUCCESS;
3238 }
3239 }
3240
3241
3242
3243
3244
3245 if (!isInReplay && coprocessorHost != null) {
3246 for (int i = firstIndex; i < lastIndexExclusive; i++) {
3247
3248 if (batchOp.retCodeDetails[i].getOperationStatusCode()
3249 != OperationStatusCode.SUCCESS) {
3250 continue;
3251 }
3252 Mutation m = batchOp.getMutation(i);
3253 if (m instanceof Put) {
3254 coprocessorHost.postPut((Put) m, walEdit, m.getDurability());
3255 } else {
3256 coprocessorHost.postDelete((Delete) m, walEdit, m.getDurability());
3257 }
3258 }
3259 }
3260
3261 success = true;
3262 return addedSize;
3263 } finally {
3264
3265 if (doRollBackMemstore) {
3266 for (int j = 0; j < familyMaps.length; j++) {
3267 for(List<Cell> cells:familyMaps[j].values()) {
3268 rollbackMemstore(cells);
3269 }
3270 }
3271 if (writeEntry != null) mvcc.complete(writeEntry);
3272 } else if (writeEntry != null) {
3273 mvcc.completeAndWait(writeEntry);
3274 }
3275
3276 if (locked) {
3277 this.updatesLock.readLock().unlock();
3278 }
3279 releaseRowLocks(acquiredRowLocks);
3280
3281
3282
3283
3284
3285
3286
3287 if (noOfPuts > 0) {
3288
3289 if (this.metricsRegion != null) {
3290 this.metricsRegion.updatePut();
3291 }
3292 }
3293 if (noOfDeletes > 0) {
3294
3295 if (this.metricsRegion != null) {
3296 this.metricsRegion.updateDelete();
3297 }
3298 }
3299 if (!success) {
3300 for (int i = firstIndex; i < lastIndexExclusive; i++) {
3301 if (batchOp.retCodeDetails[i].getOperationStatusCode() == OperationStatusCode.NOT_RUN) {
3302 batchOp.retCodeDetails[i] = OperationStatus.FAILURE;
3303 }
3304 }
3305 }
3306 if (coprocessorHost != null && !batchOp.isInReplay()) {
3307
3308
3309 MiniBatchOperationInProgress<Mutation> miniBatchOp =
3310 new MiniBatchOperationInProgress<Mutation>(batchOp.getMutationsForCoprocs(),
3311 batchOp.retCodeDetails, batchOp.walEditsFromCoprocessors, firstIndex,
3312 lastIndexExclusive);
3313 coprocessorHost.postBatchMutateIndispensably(miniBatchOp, success);
3314 }
3315
3316 batchOp.nextIndexToProcess = lastIndexExclusive;
3317 }
3318 }
3319
3320
3321
3322
3323
3324 protected Durability getEffectiveDurability(Durability d) {
3325 return d == Durability.USE_DEFAULT ? this.durability : d;
3326 }
3327
3328
3329
3330
3331
3332
3333 @Override
3334 public boolean checkAndMutate(byte [] row, byte [] family, byte [] qualifier,
3335 CompareOp compareOp, ByteArrayComparable comparator, Mutation w,
3336 boolean writeToWAL)
3337 throws IOException{
3338 checkReadOnly();
3339
3340
3341 checkResources();
3342 boolean isPut = w instanceof Put;
3343 if (!isPut && !(w instanceof Delete))
3344 throw new org.apache.hadoop.hbase.DoNotRetryIOException("Action must " +
3345 "be Put or Delete");
3346 if (!Bytes.equals(row, w.getRow())) {
3347 throw new org.apache.hadoop.hbase.DoNotRetryIOException("Action's " +
3348 "getRow must match the passed row");
3349 }
3350
3351 startRegionOperation();
3352 try {
3353 Get get = new Get(row);
3354 checkFamily(family);
3355 get.addColumn(family, qualifier);
3356
3357
3358 RowLock rowLock = getRowLock(get.getRow());
3359
3360 mvcc.await();
3361 try {
3362 if (this.getCoprocessorHost() != null) {
3363 Boolean processed = null;
3364 if (w instanceof Put) {
3365 processed = this.getCoprocessorHost().preCheckAndPutAfterRowLock(row, family,
3366 qualifier, compareOp, comparator, (Put) w);
3367 } else if (w instanceof Delete) {
3368 processed = this.getCoprocessorHost().preCheckAndDeleteAfterRowLock(row, family,
3369 qualifier, compareOp, comparator, (Delete) w);
3370 }
3371 if (processed != null) {
3372 return processed;
3373 }
3374 }
3375 List<Cell> result = get(get, false);
3376
3377 boolean valueIsNull = comparator.getValue() == null ||
3378 comparator.getValue().length == 0;
3379 boolean matches = false;
3380 long cellTs = 0;
3381 if (result.size() == 0 && valueIsNull) {
3382 matches = true;
3383 } else if (result.size() > 0 && result.get(0).getValueLength() == 0 &&
3384 valueIsNull) {
3385 matches = true;
3386 cellTs = result.get(0).getTimestamp();
3387 } else if (result.size() == 1 && !valueIsNull) {
3388 Cell kv = result.get(0);
3389 cellTs = kv.getTimestamp();
3390 int compareResult = comparator.compareTo(kv.getValueArray(),
3391 kv.getValueOffset(), kv.getValueLength());
3392 switch (compareOp) {
3393 case LESS:
3394 matches = compareResult < 0;
3395 break;
3396 case LESS_OR_EQUAL:
3397 matches = compareResult <= 0;
3398 break;
3399 case EQUAL:
3400 matches = compareResult == 0;
3401 break;
3402 case NOT_EQUAL:
3403 matches = compareResult != 0;
3404 break;
3405 case GREATER_OR_EQUAL:
3406 matches = compareResult >= 0;
3407 break;
3408 case GREATER:
3409 matches = compareResult > 0;
3410 break;
3411 default:
3412 throw new RuntimeException("Unknown Compare op " + compareOp.name());
3413 }
3414 }
3415
3416 if (matches) {
3417
3418
3419
3420
3421 long now = EnvironmentEdgeManager.currentTime();
3422 long ts = Math.max(now, cellTs);
3423 byte[] byteTs = Bytes.toBytes(ts);
3424
3425 if (w instanceof Put) {
3426 updateCellTimestamps(w.getFamilyCellMap().values(), byteTs);
3427 }
3428
3429
3430
3431
3432
3433 doBatchMutate(w);
3434 this.checkAndMutateChecksPassed.increment();
3435 return true;
3436 }
3437 this.checkAndMutateChecksFailed.increment();
3438 return false;
3439 } finally {
3440 rowLock.release();
3441 }
3442 } finally {
3443 closeRegionOperation();
3444 }
3445 }
3446
3447
3448
3449
3450
3451
3452 @Override
3453 public boolean checkAndRowMutate(byte [] row, byte [] family, byte [] qualifier,
3454 CompareOp compareOp, ByteArrayComparable comparator, RowMutations rm,
3455 boolean writeToWAL) throws IOException {
3456 checkReadOnly();
3457
3458
3459 checkResources();
3460
3461 startRegionOperation();
3462 try {
3463 Get get = new Get(row);
3464 checkFamily(family);
3465 get.addColumn(family, qualifier);
3466
3467
3468 RowLock rowLock = getRowLock(get.getRow());
3469
3470 mvcc.await();
3471 try {
3472 List<Cell> result = get(get, false);
3473
3474 boolean valueIsNull = comparator.getValue() == null ||
3475 comparator.getValue().length == 0;
3476 boolean matches = false;
3477 long cellTs = 0;
3478 if (result.size() == 0 && valueIsNull) {
3479 matches = true;
3480 } else if (result.size() > 0 && result.get(0).getValueLength() == 0 &&
3481 valueIsNull) {
3482 matches = true;
3483 cellTs = result.get(0).getTimestamp();
3484 } else if (result.size() == 1 && !valueIsNull) {
3485 Cell kv = result.get(0);
3486 cellTs = kv.getTimestamp();
3487 int compareResult = comparator.compareTo(kv.getValueArray(),
3488 kv.getValueOffset(), kv.getValueLength());
3489 switch (compareOp) {
3490 case LESS:
3491 matches = compareResult < 0;
3492 break;
3493 case LESS_OR_EQUAL:
3494 matches = compareResult <= 0;
3495 break;
3496 case EQUAL:
3497 matches = compareResult == 0;
3498 break;
3499 case NOT_EQUAL:
3500 matches = compareResult != 0;
3501 break;
3502 case GREATER_OR_EQUAL:
3503 matches = compareResult >= 0;
3504 break;
3505 case GREATER:
3506 matches = compareResult > 0;
3507 break;
3508 default:
3509 throw new RuntimeException("Unknown Compare op " + compareOp.name());
3510 }
3511 }
3512
3513 if (matches) {
3514
3515
3516
3517
3518 long now = EnvironmentEdgeManager.currentTime();
3519 long ts = Math.max(now, cellTs);
3520 byte[] byteTs = Bytes.toBytes(ts);
3521
3522 for (Mutation w : rm.getMutations()) {
3523 if (w instanceof Put) {
3524 updateCellTimestamps(w.getFamilyCellMap().values(), byteTs);
3525 }
3526
3527
3528 }
3529
3530
3531
3532 mutateRow(rm);
3533 this.checkAndMutateChecksPassed.increment();
3534 return true;
3535 }
3536 this.checkAndMutateChecksFailed.increment();
3537 return false;
3538 } finally {
3539 rowLock.release();
3540 }
3541 } finally {
3542 closeRegionOperation();
3543 }
3544 }
3545
3546 private void doBatchMutate(Mutation mutation) throws IOException {
3547
3548 OperationStatus[] batchMutate = this.batchMutate(new Mutation[]{mutation});
3549 if (batchMutate[0].getOperationStatusCode().equals(OperationStatusCode.SANITY_CHECK_FAILURE)) {
3550 throw new FailedSanityCheckException(batchMutate[0].getExceptionMsg());
3551 } else if (batchMutate[0].getOperationStatusCode().equals(OperationStatusCode.BAD_FAMILY)) {
3552 throw new NoSuchColumnFamilyException(batchMutate[0].getExceptionMsg());
3553 }
3554 }
3555
3556
3557
3558
3559
3560
3561
3562
3563
3564
3565
3566
3567
3568
3569 public void addRegionToSnapshot(SnapshotDescription desc,
3570 ForeignExceptionSnare exnSnare) throws IOException {
3571 Path rootDir = FSUtils.getRootDir(conf);
3572 Path snapshotDir = SnapshotDescriptionUtils.getWorkingSnapshotDir(desc, rootDir);
3573
3574 SnapshotManifest manifest = SnapshotManifest.create(conf, getFilesystem(),
3575 snapshotDir, desc, exnSnare);
3576 manifest.addRegion(this);
3577
3578
3579
3580 if (!Bytes.equals(getRegionInfo().getStartKey(), HConstants.EMPTY_START_ROW))
3581 return;
3582
3583
3584 List<Store> stores = getStores();
3585 for (Store store : stores) {
3586 boolean hasMobStore = store.getFamily().isMobEnabled();
3587 if (hasMobStore) {
3588
3589 HRegionInfo mobRegionInfo = MobUtils.getMobRegionInfo(this.getTableDesc().getTableName());
3590 mobRegionInfo.setOffline(true);
3591 manifest.addMobRegion(mobRegionInfo, this.getTableDesc().getColumnFamilies());
3592 return;
3593 }
3594 }
3595 }
3596
3597 @Override
3598 public void updateCellTimestamps(final Iterable<List<Cell>> cellItr, final byte[] now)
3599 throws IOException {
3600 for (List<Cell> cells: cellItr) {
3601 if (cells == null) continue;
3602 assert cells instanceof RandomAccess;
3603 int listSize = cells.size();
3604 for (int i = 0; i < listSize; i++) {
3605 CellUtil.updateLatestStamp(cells.get(i), now, 0);
3606 }
3607 }
3608 }
3609
3610
3611
3612
3613 void rewriteCellTags(Map<byte[], List<Cell>> familyMap, final Mutation m) {
3614
3615
3616
3617 if (m.getTTL() == Long.MAX_VALUE) {
3618 return;
3619 }
3620
3621
3622
3623 for (Map.Entry<byte[], List<Cell>> e: familyMap.entrySet()) {
3624 List<Cell> cells = e.getValue();
3625 assert cells instanceof RandomAccess;
3626 int listSize = cells.size();
3627 for (int i = 0; i < listSize; i++) {
3628 Cell cell = cells.get(i);
3629 List<Tag> newTags = Tag.carryForwardTags(null, cell);
3630 newTags = carryForwardTTLTag(newTags, m);
3631
3632
3633 cells.set(i, new KeyValue(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength(),
3634 cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength(),
3635 cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength(),
3636 cell.getTimestamp(), KeyValue.Type.codeToType(cell.getTypeByte()),
3637 cell.getValueArray(), cell.getValueOffset(), cell.getValueLength(),
3638 newTags));
3639 }
3640 }
3641 }
3642
3643
3644
3645
3646
3647
3648
3649 private void checkResources() throws RegionTooBusyException {
3650
3651 if (this.getRegionInfo().isMetaRegion()) return;
3652
3653 if (this.memstoreSize.get() > this.blockingMemStoreSize) {
3654 blockedRequestsCount.increment();
3655 requestFlush();
3656 throw new RegionTooBusyException("Above memstore limit, " +
3657 "regionName=" + (this.getRegionInfo() == null ? "unknown" :
3658 this.getRegionInfo().getRegionNameAsString()) +
3659 ", server=" + (this.getRegionServerServices() == null ? "unknown" :
3660 this.getRegionServerServices().getServerName()) +
3661 ", memstoreSize=" + memstoreSize.get() +
3662 ", blockingMemStoreSize=" + blockingMemStoreSize);
3663 }
3664 }
3665
3666
3667
3668
3669 protected void checkReadOnly() throws IOException {
3670 if (isReadOnly()) {
3671 throw new DoNotRetryIOException("region is read only");
3672 }
3673 }
3674
3675 protected void checkReadsEnabled() throws IOException {
3676 if (!this.writestate.readsEnabled) {
3677 throw new IOException(getRegionInfo().getEncodedName()
3678 + ": The region's reads are disabled. Cannot serve the request");
3679 }
3680 }
3681
3682 public void setReadsEnabled(boolean readsEnabled) {
3683 if (readsEnabled && !this.writestate.readsEnabled) {
3684 LOG.info(getRegionInfo().getEncodedName() + " : Enabling reads for region.");
3685 }
3686 this.writestate.setReadsEnabled(readsEnabled);
3687 }
3688
3689
3690
3691
3692
3693
3694
3695 private void put(final byte [] row, byte [] family, List<Cell> edits)
3696 throws IOException {
3697 NavigableMap<byte[], List<Cell>> familyMap;
3698 familyMap = new TreeMap<byte[], List<Cell>>(Bytes.BYTES_COMPARATOR);
3699
3700 familyMap.put(family, edits);
3701 Put p = new Put(row);
3702 p.setFamilyCellMap(familyMap);
3703 doBatchMutate(p);
3704 }
3705
3706
3707
3708
3709
3710
3711
3712
3713
3714
3715
3716
3717
3718
3719
3720
3721 private long applyFamilyMapToMemstore(Map<byte[], List<Cell>> familyMap,
3722 long mvccNum, boolean isInReplay) throws IOException {
3723 long size = 0;
3724
3725 for (Map.Entry<byte[], List<Cell>> e : familyMap.entrySet()) {
3726 byte[] family = e.getKey();
3727 List<Cell> cells = e.getValue();
3728 assert cells instanceof RandomAccess;
3729 Store store = getStore(family);
3730 int listSize = cells.size();
3731 for (int i=0; i < listSize; i++) {
3732 Cell cell = cells.get(i);
3733 if (cell.getSequenceId() == 0 || isInReplay) {
3734 CellUtil.setSequenceId(cell, mvccNum);
3735 }
3736 size += store.add(cell);
3737 }
3738 }
3739
3740 return size;
3741 }
3742
3743
3744
3745
3746
3747
3748 private void rollbackMemstore(List<Cell> memstoreCells) {
3749 int kvsRolledback = 0;
3750
3751 for (Cell cell : memstoreCells) {
3752 byte[] family = CellUtil.cloneFamily(cell);
3753 Store store = getStore(family);
3754 store.rollback(cell);
3755 kvsRolledback++;
3756 }
3757 LOG.debug("rollbackMemstore rolled back " + kvsRolledback);
3758 }
3759
3760 @Override
3761 public void checkFamilies(Collection<byte[]> families) throws NoSuchColumnFamilyException {
3762 for (byte[] family : families) {
3763 checkFamily(family);
3764 }
3765 }
3766
3767
3768
3769
3770
3771 private void removeNonExistentColumnFamilyForReplay(
3772 final Map<byte[], List<Cell>> familyMap) {
3773 List<byte[]> nonExistentList = null;
3774 for (byte[] family : familyMap.keySet()) {
3775 if (!this.htableDescriptor.hasFamily(family)) {
3776 if (nonExistentList == null) {
3777 nonExistentList = new ArrayList<byte[]>();
3778 }
3779 nonExistentList.add(family);
3780 }
3781 }
3782 if (nonExistentList != null) {
3783 for (byte[] family : nonExistentList) {
3784
3785 LOG.info("No family for " + Bytes.toString(family) + " omit from reply.");
3786 familyMap.remove(family);
3787 }
3788 }
3789 }
3790
3791 @Override
3792 public void checkTimestamps(final Map<byte[], List<Cell>> familyMap, long now)
3793 throws FailedSanityCheckException {
3794 if (timestampSlop == HConstants.LATEST_TIMESTAMP) {
3795 return;
3796 }
3797 long maxTs = now + timestampSlop;
3798 for (List<Cell> kvs : familyMap.values()) {
3799 assert kvs instanceof RandomAccess;
3800 int listSize = kvs.size();
3801 for (int i=0; i < listSize; i++) {
3802 Cell cell = kvs.get(i);
3803
3804 long ts = cell.getTimestamp();
3805 if (ts != HConstants.LATEST_TIMESTAMP && ts > maxTs) {
3806 throw new FailedSanityCheckException("Timestamp for KV out of range "
3807 + cell + " (too.new=" + timestampSlop + ")");
3808 }
3809 }
3810 }
3811 }
3812
3813
3814
3815
3816
3817
3818
3819 private void addFamilyMapToWALEdit(Map<byte[], List<Cell>> familyMap,
3820 WALEdit walEdit) {
3821 for (List<Cell> edits : familyMap.values()) {
3822 assert edits instanceof RandomAccess;
3823 int listSize = edits.size();
3824 for (int i=0; i < listSize; i++) {
3825 Cell cell = edits.get(i);
3826 walEdit.add(cell);
3827 }
3828 }
3829 }
3830
3831 private void requestFlush() {
3832 if (this.rsServices == null) {
3833 return;
3834 }
3835 synchronized (writestate) {
3836 if (this.writestate.isFlushRequested()) {
3837 return;
3838 }
3839 writestate.flushRequested = true;
3840 }
3841
3842 this.rsServices.getFlushRequester().requestFlush(this, false);
3843 if (LOG.isDebugEnabled()) {
3844 LOG.debug("Flush requested on " + this.getRegionInfo().getEncodedName());
3845 }
3846 }
3847
3848
3849
3850
3851
3852 private boolean isFlushSize(final long size) {
3853 return size > this.memstoreFlushSize;
3854 }
3855
3856
3857
3858
3859
3860
3861
3862
3863
3864
3865
3866
3867
3868
3869
3870
3871
3872
3873
3874
3875
3876
3877
3878
3879
3880
3881
3882
3883
3884
3885
3886
3887
3888
3889 protected long replayRecoveredEditsIfAny(final Path regiondir,
3890 Map<byte[], Long> maxSeqIdInStores,
3891 final CancelableProgressable reporter, final MonitoredTask status)
3892 throws IOException {
3893 long minSeqIdForTheRegion = -1;
3894 for (Long maxSeqIdInStore : maxSeqIdInStores.values()) {
3895 if (maxSeqIdInStore < minSeqIdForTheRegion || minSeqIdForTheRegion == -1) {
3896 minSeqIdForTheRegion = maxSeqIdInStore;
3897 }
3898 }
3899 long seqid = minSeqIdForTheRegion;
3900
3901 FileSystem fs = this.fs.getFileSystem();
3902 NavigableSet<Path> files = WALSplitter.getSplitEditFilesSorted(fs, regiondir);
3903 if (LOG.isDebugEnabled()) {
3904 LOG.debug("Found " + (files == null ? 0 : files.size())
3905 + " recovered edits file(s) under " + regiondir);
3906 }
3907
3908 if (files == null || files.isEmpty()) return seqid;
3909
3910 for (Path edits: files) {
3911 if (edits == null || !fs.exists(edits)) {
3912 LOG.warn("Null or non-existent edits file: " + edits);
3913 continue;
3914 }
3915 if (isZeroLengthThenDelete(fs, edits)) continue;
3916
3917 long maxSeqId;
3918 String fileName = edits.getName();
3919 maxSeqId = Math.abs(Long.parseLong(fileName));
3920 if (maxSeqId <= minSeqIdForTheRegion) {
3921 if (LOG.isDebugEnabled()) {
3922 String msg = "Maximum sequenceid for this wal is " + maxSeqId
3923 + " and minimum sequenceid for the region is " + minSeqIdForTheRegion
3924 + ", skipped the whole file, path=" + edits;
3925 LOG.debug(msg);
3926 }
3927 continue;
3928 }
3929
3930 try {
3931
3932
3933 seqid = Math.max(seqid, replayRecoveredEdits(edits, maxSeqIdInStores, reporter));
3934 } catch (IOException e) {
3935 boolean skipErrors = conf.getBoolean(
3936 HConstants.HREGION_EDITS_REPLAY_SKIP_ERRORS,
3937 conf.getBoolean(
3938 "hbase.skip.errors",
3939 HConstants.DEFAULT_HREGION_EDITS_REPLAY_SKIP_ERRORS));
3940 if (conf.get("hbase.skip.errors") != null) {
3941 LOG.warn(
3942 "The property 'hbase.skip.errors' has been deprecated. Please use " +
3943 HConstants.HREGION_EDITS_REPLAY_SKIP_ERRORS + " instead.");
3944 }
3945 if (skipErrors) {
3946 Path p = WALSplitter.moveAsideBadEditsFile(fs, edits);
3947 LOG.error(HConstants.HREGION_EDITS_REPLAY_SKIP_ERRORS
3948 + "=true so continuing. Renamed " + edits +
3949 " as " + p, e);
3950 } else {
3951 throw e;
3952 }
3953 }
3954 }
3955
3956
3957 if (this.rsAccounting != null) {
3958 this.rsAccounting.clearRegionReplayEditsSize(getRegionInfo().getRegionName());
3959 }
3960 if (seqid > minSeqIdForTheRegion) {
3961
3962 internalFlushcache(null, seqid, stores.values(), status, false);
3963 }
3964
3965 if (files.size() > 0 && this.conf.getBoolean("hbase.region.archive.recovered.edits", false)) {
3966
3967
3968
3969 String fakeFamilyName = WALSplitter.getRegionDirRecoveredEditsDir(regiondir).getName();
3970 Set<StoreFile> fakeStoreFiles = new HashSet<StoreFile>(files.size());
3971 for (Path file: files) {
3972 fakeStoreFiles.add(new StoreFile(getRegionFileSystem().getFileSystem(), file, this.conf,
3973 null, null));
3974 }
3975 getRegionFileSystem().removeStoreFiles(fakeFamilyName, fakeStoreFiles);
3976 } else {
3977 for (Path file: files) {
3978 if (!fs.delete(file, false)) {
3979 LOG.error("Failed delete of " + file);
3980 } else {
3981 LOG.debug("Deleted recovered.edits file=" + file);
3982 }
3983 }
3984 }
3985 return seqid;
3986 }
3987
3988
3989
3990
3991
3992
3993
3994
3995
3996
3997 private long replayRecoveredEdits(final Path edits,
3998 Map<byte[], Long> maxSeqIdInStores, final CancelableProgressable reporter)
3999 throws IOException {
4000 String msg = "Replaying edits from " + edits;
4001 LOG.info(msg);
4002 MonitoredTask status = TaskMonitor.get().createStatus(msg);
4003 FileSystem fs = this.fs.getFileSystem();
4004
4005 status.setStatus("Opening recovered edits");
4006 WAL.Reader reader = null;
4007 try {
4008 reader = WALFactory.createReader(fs, edits, conf);
4009 long currentEditSeqId = -1;
4010 long currentReplaySeqId = -1;
4011 long firstSeqIdInLog = -1;
4012 long skippedEdits = 0;
4013 long editsCount = 0;
4014 long intervalEdits = 0;
4015 WAL.Entry entry;
4016 Store store = null;
4017 boolean reported_once = false;
4018 ServerNonceManager ng = this.rsServices == null ? null : this.rsServices.getNonceManager();
4019
4020 try {
4021
4022 int interval = this.conf.getInt("hbase.hstore.report.interval.edits", 2000);
4023
4024 int period = this.conf.getInt("hbase.hstore.report.period", 300000);
4025 long lastReport = EnvironmentEdgeManager.currentTime();
4026
4027 while ((entry = reader.next()) != null) {
4028 WALKey key = entry.getKey();
4029 WALEdit val = entry.getEdit();
4030
4031 if (ng != null) {
4032 ng.reportOperationFromWal(key.getNonceGroup(), key.getNonce(), key.getWriteTime());
4033 }
4034
4035 if (reporter != null) {
4036 intervalEdits += val.size();
4037 if (intervalEdits >= interval) {
4038
4039 intervalEdits = 0;
4040 long cur = EnvironmentEdgeManager.currentTime();
4041 if (lastReport + period <= cur) {
4042 status.setStatus("Replaying edits..." +
4043 " skipped=" + skippedEdits +
4044 " edits=" + editsCount);
4045
4046 if(!reporter.progress()) {
4047 msg = "Progressable reporter failed, stopping replay";
4048 LOG.warn(msg);
4049 status.abort(msg);
4050 throw new IOException(msg);
4051 }
4052 reported_once = true;
4053 lastReport = cur;
4054 }
4055 }
4056 }
4057
4058 if (firstSeqIdInLog == -1) {
4059 firstSeqIdInLog = key.getLogSeqNum();
4060 }
4061 if (currentEditSeqId > key.getLogSeqNum()) {
4062
4063
4064 LOG.error(getRegionInfo().getEncodedName() + " : "
4065 + "Found decreasing SeqId. PreId=" + currentEditSeqId + " key=" + key
4066 + "; edit=" + val);
4067 } else {
4068 currentEditSeqId = key.getLogSeqNum();
4069 }
4070 currentReplaySeqId = (key.getOrigLogSeqNum() > 0) ?
4071 key.getOrigLogSeqNum() : currentEditSeqId;
4072
4073
4074
4075 if (coprocessorHost != null) {
4076 status.setStatus("Running pre-WAL-restore hook in coprocessors");
4077 if (coprocessorHost.preWALRestore(this.getRegionInfo(), key, val)) {
4078
4079 continue;
4080 }
4081 }
4082 boolean checkRowWithinBoundary = false;
4083
4084 if (!Bytes.equals(key.getEncodedRegionName(),
4085 this.getRegionInfo().getEncodedNameAsBytes())) {
4086 checkRowWithinBoundary = true;
4087 }
4088
4089 boolean flush = false;
4090 for (Cell cell: val.getCells()) {
4091
4092
4093 if (CellUtil.matchingFamily(cell, WALEdit.METAFAMILY)) {
4094
4095 if (!checkRowWithinBoundary) {
4096
4097 CompactionDescriptor compaction = WALEdit.getCompaction(cell);
4098 if (compaction != null) {
4099
4100 replayWALCompactionMarker(compaction, false, true, Long.MAX_VALUE);
4101 }
4102 }
4103 skippedEdits++;
4104 continue;
4105 }
4106
4107 if (store == null || !CellUtil.matchingFamily(cell, store.getFamily().getName())) {
4108 store = getStore(cell);
4109 }
4110 if (store == null) {
4111
4112
4113 LOG.warn("No family for " + cell);
4114 skippedEdits++;
4115 continue;
4116 }
4117 if (checkRowWithinBoundary && !rowIsInRange(this.getRegionInfo(),
4118 cell.getRowArray(), cell.getRowOffset(), cell.getRowLength())) {
4119 LOG.warn("Row of " + cell + " is not within region boundary");
4120 skippedEdits++;
4121 continue;
4122 }
4123
4124 if (key.getLogSeqNum() <= maxSeqIdInStores.get(store.getFamily()
4125 .getName())) {
4126 skippedEdits++;
4127 continue;
4128 }
4129 CellUtil.setSequenceId(cell, currentReplaySeqId);
4130
4131
4132
4133
4134 flush |= restoreEdit(store, cell);
4135 editsCount++;
4136 }
4137 if (flush) {
4138 internalFlushcache(null, currentEditSeqId, stores.values(), status, false);
4139 }
4140
4141 if (coprocessorHost != null) {
4142 coprocessorHost.postWALRestore(this.getRegionInfo(), key, val);
4143 }
4144 }
4145 } catch (EOFException eof) {
4146 Path p = WALSplitter.moveAsideBadEditsFile(fs, edits);
4147 msg = "Encountered EOF. Most likely due to Master failure during " +
4148 "wal splitting, so we have this data in another edit. " +
4149 "Continuing, but renaming " + edits + " as " + p;
4150 LOG.warn(msg, eof);
4151 status.abort(msg);
4152 } catch (IOException ioe) {
4153
4154
4155 if (ioe.getCause() instanceof ParseException) {
4156 Path p = WALSplitter.moveAsideBadEditsFile(fs, edits);
4157 msg = "File corruption encountered! " +
4158 "Continuing, but renaming " + edits + " as " + p;
4159 LOG.warn(msg, ioe);
4160 status.setStatus(msg);
4161 } else {
4162 status.abort(StringUtils.stringifyException(ioe));
4163
4164
4165 throw ioe;
4166 }
4167 }
4168 if (reporter != null && !reported_once) {
4169 reporter.progress();
4170 }
4171 msg = "Applied " + editsCount + ", skipped " + skippedEdits +
4172 ", firstSequenceIdInLog=" + firstSeqIdInLog +
4173 ", maxSequenceIdInLog=" + currentEditSeqId + ", path=" + edits;
4174 status.markComplete(msg);
4175 LOG.debug(msg);
4176 return currentEditSeqId;
4177 } finally {
4178 status.cleanup();
4179 if (reader != null) {
4180 reader.close();
4181 }
4182 }
4183 }
4184
4185
4186
4187
4188
4189
4190 void replayWALCompactionMarker(CompactionDescriptor compaction, boolean pickCompactionFiles,
4191 boolean removeFiles, long replaySeqId)
4192 throws IOException {
4193 try {
4194 checkTargetRegion(compaction.getEncodedRegionName().toByteArray(),
4195 "Compaction marker from WAL ", compaction);
4196 } catch (WrongRegionException wre) {
4197 if (RegionReplicaUtil.isDefaultReplica(this.getRegionInfo())) {
4198
4199 return;
4200 }
4201 throw wre;
4202 }
4203
4204 synchronized (writestate) {
4205 if (replaySeqId < lastReplayedOpenRegionSeqId) {
4206 LOG.warn(getRegionInfo().getEncodedName() + " : "
4207 + "Skipping replaying compaction event :" + TextFormat.shortDebugString(compaction)
4208 + " because its sequence id " + replaySeqId + " is smaller than this regions "
4209 + "lastReplayedOpenRegionSeqId of " + lastReplayedOpenRegionSeqId);
4210 return;
4211 }
4212 if (replaySeqId < lastReplayedCompactionSeqId) {
4213 LOG.warn(getRegionInfo().getEncodedName() + " : "
4214 + "Skipping replaying compaction event :" + TextFormat.shortDebugString(compaction)
4215 + " because its sequence id " + replaySeqId + " is smaller than this regions "
4216 + "lastReplayedCompactionSeqId of " + lastReplayedCompactionSeqId);
4217 return;
4218 } else {
4219 lastReplayedCompactionSeqId = replaySeqId;
4220 }
4221
4222 if (LOG.isDebugEnabled()) {
4223 LOG.debug(getRegionInfo().getEncodedName() + " : "
4224 + "Replaying compaction marker " + TextFormat.shortDebugString(compaction)
4225 + " with seqId=" + replaySeqId + " and lastReplayedOpenRegionSeqId="
4226 + lastReplayedOpenRegionSeqId);
4227 }
4228
4229 startRegionOperation(Operation.REPLAY_EVENT);
4230 try {
4231 Store store = this.getStore(compaction.getFamilyName().toByteArray());
4232 if (store == null) {
4233 LOG.warn(getRegionInfo().getEncodedName() + " : "
4234 + "Found Compaction WAL edit for deleted family:"
4235 + Bytes.toString(compaction.getFamilyName().toByteArray()));
4236 return;
4237 }
4238 store.replayCompactionMarker(compaction, pickCompactionFiles, removeFiles);
4239 logRegionFiles();
4240 } catch (FileNotFoundException ex) {
4241 LOG.warn(getRegionInfo().getEncodedName() + " : "
4242 + "At least one of the store files in compaction: "
4243 + TextFormat.shortDebugString(compaction)
4244 + " doesn't exist any more. Skip loading the file(s)", ex);
4245 } finally {
4246 closeRegionOperation(Operation.REPLAY_EVENT);
4247 }
4248 }
4249 }
4250
4251 void replayWALFlushMarker(FlushDescriptor flush, long replaySeqId) throws IOException {
4252 checkTargetRegion(flush.getEncodedRegionName().toByteArray(),
4253 "Flush marker from WAL ", flush);
4254
4255 if (ServerRegionReplicaUtil.isDefaultReplica(this.getRegionInfo())) {
4256 return;
4257 }
4258
4259 if (LOG.isDebugEnabled()) {
4260 LOG.debug(getRegionInfo().getEncodedName() + " : "
4261 + "Replaying flush marker " + TextFormat.shortDebugString(flush));
4262 }
4263
4264 startRegionOperation(Operation.REPLAY_EVENT);
4265 try {
4266 FlushAction action = flush.getAction();
4267 switch (action) {
4268 case START_FLUSH:
4269 replayWALFlushStartMarker(flush);
4270 break;
4271 case COMMIT_FLUSH:
4272 replayWALFlushCommitMarker(flush);
4273 break;
4274 case ABORT_FLUSH:
4275 replayWALFlushAbortMarker(flush);
4276 break;
4277 case CANNOT_FLUSH:
4278 replayWALFlushCannotFlushMarker(flush, replaySeqId);
4279 break;
4280 default:
4281 LOG.warn(getRegionInfo().getEncodedName() + " : " +
4282 "Received a flush event with unknown action, ignoring. " +
4283 TextFormat.shortDebugString(flush));
4284 break;
4285 }
4286
4287 logRegionFiles();
4288 } finally {
4289 closeRegionOperation(Operation.REPLAY_EVENT);
4290 }
4291 }
4292
4293
4294
4295
4296
4297 @VisibleForTesting
4298 PrepareFlushResult replayWALFlushStartMarker(FlushDescriptor flush) throws IOException {
4299 long flushSeqId = flush.getFlushSequenceNumber();
4300
4301 HashSet<Store> storesToFlush = new HashSet<Store>();
4302 for (StoreFlushDescriptor storeFlush : flush.getStoreFlushesList()) {
4303 byte[] family = storeFlush.getFamilyName().toByteArray();
4304 Store store = getStore(family);
4305 if (store == null) {
4306 LOG.warn(getRegionInfo().getEncodedName() + " : "
4307 + "Received a flush start marker from primary, but the family is not found. Ignoring"
4308 + " StoreFlushDescriptor:" + TextFormat.shortDebugString(storeFlush));
4309 continue;
4310 }
4311 storesToFlush.add(store);
4312 }
4313
4314 MonitoredTask status = TaskMonitor.get().createStatus("Preparing flush " + this);
4315
4316
4317
4318 synchronized (writestate) {
4319 try {
4320 if (flush.getFlushSequenceNumber() < lastReplayedOpenRegionSeqId) {
4321 LOG.warn(getRegionInfo().getEncodedName() + " : "
4322 + "Skipping replaying flush event :" + TextFormat.shortDebugString(flush)
4323 + " because its sequence id is smaller than this regions lastReplayedOpenRegionSeqId "
4324 + " of " + lastReplayedOpenRegionSeqId);
4325 return null;
4326 }
4327 if (numMutationsWithoutWAL.get() > 0) {
4328 numMutationsWithoutWAL.set(0);
4329 dataInMemoryWithoutWAL.set(0);
4330 }
4331
4332 if (!writestate.flushing) {
4333
4334
4335
4336
4337 PrepareFlushResult prepareResult = internalPrepareFlushCache(null,
4338 flushSeqId, storesToFlush, status, false);
4339 if (prepareResult.result == null) {
4340
4341 this.writestate.flushing = true;
4342 this.prepareFlushResult = prepareResult;
4343 status.markComplete("Flush prepare successful");
4344 if (LOG.isDebugEnabled()) {
4345 LOG.debug(getRegionInfo().getEncodedName() + " : "
4346 + " Prepared flush with seqId:" + flush.getFlushSequenceNumber());
4347 }
4348 } else {
4349
4350
4351 if (prepareResult.getResult().getResult() ==
4352 FlushResult.Result.CANNOT_FLUSH_MEMSTORE_EMPTY) {
4353 this.writestate.flushing = true;
4354 this.prepareFlushResult = prepareResult;
4355 if (LOG.isDebugEnabled()) {
4356 LOG.debug(getRegionInfo().getEncodedName() + " : "
4357 + " Prepared empty flush with seqId:" + flush.getFlushSequenceNumber());
4358 }
4359 }
4360 status.abort("Flush prepare failed with " + prepareResult.result);
4361
4362 }
4363 return prepareResult;
4364 } else {
4365
4366 if (flush.getFlushSequenceNumber() == this.prepareFlushResult.flushOpSeqId) {
4367
4368 LOG.warn(getRegionInfo().getEncodedName() + " : "
4369 + "Received a flush prepare marker with the same seqId: " +
4370 + flush.getFlushSequenceNumber() + " before clearing the previous one with seqId: "
4371 + prepareFlushResult.flushOpSeqId + ". Ignoring");
4372
4373 } else if (flush.getFlushSequenceNumber() < this.prepareFlushResult.flushOpSeqId) {
4374
4375
4376 LOG.warn(getRegionInfo().getEncodedName() + " : "
4377 + "Received a flush prepare marker with a smaller seqId: " +
4378 + flush.getFlushSequenceNumber() + " before clearing the previous one with seqId: "
4379 + prepareFlushResult.flushOpSeqId + ". Ignoring");
4380
4381 } else {
4382
4383 LOG.warn(getRegionInfo().getEncodedName() + " : "
4384 + "Received a flush prepare marker with a larger seqId: " +
4385 + flush.getFlushSequenceNumber() + " before clearing the previous one with seqId: "
4386 + prepareFlushResult.flushOpSeqId + ". Ignoring");
4387
4388
4389
4390
4391
4392
4393
4394
4395
4396
4397
4398 }
4399 }
4400 } finally {
4401 status.cleanup();
4402 writestate.notifyAll();
4403 }
4404 }
4405 return null;
4406 }
4407
4408 @VisibleForTesting
4409 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="NN_NAKED_NOTIFY",
4410 justification="Intentional; post memstore flush")
4411 void replayWALFlushCommitMarker(FlushDescriptor flush) throws IOException {
4412 MonitoredTask status = TaskMonitor.get().createStatus("Committing flush " + this);
4413
4414
4415
4416
4417
4418 synchronized (writestate) {
4419 try {
4420 if (flush.getFlushSequenceNumber() < lastReplayedOpenRegionSeqId) {
4421 LOG.warn(getRegionInfo().getEncodedName() + " : "
4422 + "Skipping replaying flush event :" + TextFormat.shortDebugString(flush)
4423 + " because its sequence id is smaller than this regions lastReplayedOpenRegionSeqId "
4424 + " of " + lastReplayedOpenRegionSeqId);
4425 return;
4426 }
4427
4428 if (writestate.flushing) {
4429 PrepareFlushResult prepareFlushResult = this.prepareFlushResult;
4430 if (flush.getFlushSequenceNumber() == prepareFlushResult.flushOpSeqId) {
4431 if (LOG.isDebugEnabled()) {
4432 LOG.debug(getRegionInfo().getEncodedName() + " : "
4433 + "Received a flush commit marker with seqId:" + flush.getFlushSequenceNumber()
4434 + " and a previous prepared snapshot was found");
4435 }
4436
4437
4438 replayFlushInStores(flush, prepareFlushResult, true);
4439
4440
4441 this.addAndGetGlobalMemstoreSize(-prepareFlushResult.totalFlushableSize);
4442
4443 this.prepareFlushResult = null;
4444 writestate.flushing = false;
4445 } else if (flush.getFlushSequenceNumber() < prepareFlushResult.flushOpSeqId) {
4446
4447
4448
4449
4450 LOG.warn(getRegionInfo().getEncodedName() + " : "
4451 + "Received a flush commit marker with smaller seqId: "
4452 + flush.getFlushSequenceNumber() + " than what we have prepared with seqId: "
4453 + prepareFlushResult.flushOpSeqId + ". Picking up new file, but not dropping"
4454 +" prepared memstore snapshot");
4455 replayFlushInStores(flush, prepareFlushResult, false);
4456
4457
4458
4459 } else {
4460
4461
4462
4463
4464
4465 LOG.warn(getRegionInfo().getEncodedName() + " : "
4466 + "Received a flush commit marker with larger seqId: "
4467 + flush.getFlushSequenceNumber() + " than what we have prepared with seqId: " +
4468 prepareFlushResult.flushOpSeqId + ". Picking up new file and dropping prepared"
4469 +" memstore snapshot");
4470
4471 replayFlushInStores(flush, prepareFlushResult, true);
4472
4473
4474 this.addAndGetGlobalMemstoreSize(-prepareFlushResult.totalFlushableSize);
4475
4476
4477
4478 dropMemstoreContentsForSeqId(flush.getFlushSequenceNumber(), null);
4479
4480 this.prepareFlushResult = null;
4481 writestate.flushing = false;
4482 }
4483
4484
4485
4486
4487
4488 this.setReadsEnabled(true);
4489 } else {
4490 LOG.warn(getRegionInfo().getEncodedName() + " : "
4491 + "Received a flush commit marker with seqId:" + flush.getFlushSequenceNumber()
4492 + ", but no previous prepared snapshot was found");
4493
4494
4495 replayFlushInStores(flush, null, false);
4496
4497
4498
4499 dropMemstoreContentsForSeqId(flush.getFlushSequenceNumber(), null);
4500 }
4501
4502 status.markComplete("Flush commit successful");
4503
4504
4505 this.maxFlushedSeqId = flush.getFlushSequenceNumber();
4506
4507
4508 mvcc.advanceTo(flush.getFlushSequenceNumber());
4509
4510 } catch (FileNotFoundException ex) {
4511 LOG.warn(getRegionInfo().getEncodedName() + " : "
4512 + "At least one of the store files in flush: " + TextFormat.shortDebugString(flush)
4513 + " doesn't exist any more. Skip loading the file(s)", ex);
4514 }
4515 finally {
4516 status.cleanup();
4517 writestate.notifyAll();
4518 }
4519 }
4520
4521
4522
4523 synchronized (this) {
4524 notifyAll();
4525 }
4526 }
4527
4528
4529
4530
4531
4532
4533
4534
4535
4536 private void replayFlushInStores(FlushDescriptor flush, PrepareFlushResult prepareFlushResult,
4537 boolean dropMemstoreSnapshot)
4538 throws IOException {
4539 for (StoreFlushDescriptor storeFlush : flush.getStoreFlushesList()) {
4540 byte[] family = storeFlush.getFamilyName().toByteArray();
4541 Store store = getStore(family);
4542 if (store == null) {
4543 LOG.warn(getRegionInfo().getEncodedName() + " : "
4544 + "Received a flush commit marker from primary, but the family is not found."
4545 + "Ignoring StoreFlushDescriptor:" + storeFlush);
4546 continue;
4547 }
4548 List<String> flushFiles = storeFlush.getFlushOutputList();
4549 StoreFlushContext ctx = null;
4550 long startTime = EnvironmentEdgeManager.currentTime();
4551 if (prepareFlushResult == null || prepareFlushResult.storeFlushCtxs == null) {
4552 ctx = store.createFlushContext(flush.getFlushSequenceNumber());
4553 } else {
4554 ctx = prepareFlushResult.storeFlushCtxs.get(family);
4555 startTime = prepareFlushResult.startTime;
4556 }
4557
4558 if (ctx == null) {
4559 LOG.warn(getRegionInfo().getEncodedName() + " : "
4560 + "Unexpected: flush commit marker received from store "
4561 + Bytes.toString(family) + " but no associated flush context. Ignoring");
4562 continue;
4563 }
4564
4565 ctx.replayFlush(flushFiles, dropMemstoreSnapshot);
4566
4567
4568 this.lastStoreFlushTimeMap.put(store, startTime);
4569 }
4570 }
4571
4572
4573
4574
4575
4576
4577 private long dropMemstoreContentsForSeqId(long seqId, Store store) throws IOException {
4578 long totalFreedSize = 0;
4579 this.updatesLock.writeLock().lock();
4580 try {
4581
4582 long currentSeqId = mvcc.getReadPoint();
4583 if (seqId >= currentSeqId) {
4584
4585 LOG.info(getRegionInfo().getEncodedName() + " : "
4586 + "Dropping memstore contents as well since replayed flush seqId: "
4587 + seqId + " is greater than current seqId:" + currentSeqId);
4588
4589
4590 if (store == null) {
4591 for (Store s : stores.values()) {
4592 totalFreedSize += doDropStoreMemstoreContentsForSeqId(s, currentSeqId);
4593 }
4594 } else {
4595 totalFreedSize += doDropStoreMemstoreContentsForSeqId(store, currentSeqId);
4596 }
4597 } else {
4598 LOG.info(getRegionInfo().getEncodedName() + " : "
4599 + "Not dropping memstore contents since replayed flush seqId: "
4600 + seqId + " is smaller than current seqId:" + currentSeqId);
4601 }
4602 } finally {
4603 this.updatesLock.writeLock().unlock();
4604 }
4605 return totalFreedSize;
4606 }
4607
4608 private long doDropStoreMemstoreContentsForSeqId(Store s, long currentSeqId) throws IOException {
4609 long snapshotSize = s.getFlushableSize();
4610 this.addAndGetGlobalMemstoreSize(-snapshotSize);
4611 StoreFlushContext ctx = s.createFlushContext(currentSeqId);
4612 ctx.prepare();
4613 ctx.abort();
4614 return snapshotSize;
4615 }
4616
4617 private void replayWALFlushAbortMarker(FlushDescriptor flush) {
4618
4619
4620
4621 }
4622
4623 private void replayWALFlushCannotFlushMarker(FlushDescriptor flush, long replaySeqId) {
4624 synchronized (writestate) {
4625 if (this.lastReplayedOpenRegionSeqId > replaySeqId) {
4626 LOG.warn(getRegionInfo().getEncodedName() + " : "
4627 + "Skipping replaying flush event :" + TextFormat.shortDebugString(flush)
4628 + " because its sequence id " + replaySeqId + " is smaller than this regions "
4629 + "lastReplayedOpenRegionSeqId of " + lastReplayedOpenRegionSeqId);
4630 return;
4631 }
4632
4633
4634
4635
4636
4637
4638 this.setReadsEnabled(true);
4639 }
4640 }
4641
4642 @VisibleForTesting
4643 PrepareFlushResult getPrepareFlushResult() {
4644 return prepareFlushResult;
4645 }
4646
4647 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="NN_NAKED_NOTIFY",
4648 justification="Intentional; cleared the memstore")
4649 void replayWALRegionEventMarker(RegionEventDescriptor regionEvent) throws IOException {
4650 checkTargetRegion(regionEvent.getEncodedRegionName().toByteArray(),
4651 "RegionEvent marker from WAL ", regionEvent);
4652
4653 startRegionOperation(Operation.REPLAY_EVENT);
4654 try {
4655 if (ServerRegionReplicaUtil.isDefaultReplica(this.getRegionInfo())) {
4656 return;
4657 }
4658
4659 if (regionEvent.getEventType() == EventType.REGION_CLOSE) {
4660
4661 return;
4662 }
4663 if (regionEvent.getEventType() != EventType.REGION_OPEN) {
4664 LOG.warn(getRegionInfo().getEncodedName() + " : "
4665 + "Unknown region event received, ignoring :"
4666 + TextFormat.shortDebugString(regionEvent));
4667 return;
4668 }
4669
4670 if (LOG.isDebugEnabled()) {
4671 LOG.debug(getRegionInfo().getEncodedName() + " : "
4672 + "Replaying region open event marker " + TextFormat.shortDebugString(regionEvent));
4673 }
4674
4675
4676 synchronized (writestate) {
4677
4678
4679
4680
4681
4682
4683 if (this.lastReplayedOpenRegionSeqId <= regionEvent.getLogSequenceNumber()) {
4684 this.lastReplayedOpenRegionSeqId = regionEvent.getLogSequenceNumber();
4685 } else {
4686 LOG.warn(getRegionInfo().getEncodedName() + " : "
4687 + "Skipping replaying region event :" + TextFormat.shortDebugString(regionEvent)
4688 + " because its sequence id is smaller than this regions lastReplayedOpenRegionSeqId "
4689 + " of " + lastReplayedOpenRegionSeqId);
4690 return;
4691 }
4692
4693
4694
4695 for (StoreDescriptor storeDescriptor : regionEvent.getStoresList()) {
4696
4697 byte[] family = storeDescriptor.getFamilyName().toByteArray();
4698 Store store = getStore(family);
4699 if (store == null) {
4700 LOG.warn(getRegionInfo().getEncodedName() + " : "
4701 + "Received a region open marker from primary, but the family is not found. "
4702 + "Ignoring. StoreDescriptor:" + storeDescriptor);
4703 continue;
4704 }
4705
4706 long storeSeqId = store.getMaxSequenceId();
4707 List<String> storeFiles = storeDescriptor.getStoreFileList();
4708 try {
4709 store.refreshStoreFiles(storeFiles);
4710 } catch (FileNotFoundException ex) {
4711 LOG.warn(getRegionInfo().getEncodedName() + " : "
4712 + "At least one of the store files: " + storeFiles
4713 + " doesn't exist any more. Skip loading the file(s)", ex);
4714 continue;
4715 }
4716 if (store.getMaxSequenceId() != storeSeqId) {
4717
4718 lastStoreFlushTimeMap.put(store, EnvironmentEdgeManager.currentTime());
4719 }
4720
4721 if (writestate.flushing) {
4722
4723 if (this.prepareFlushResult.flushOpSeqId <= regionEvent.getLogSequenceNumber()) {
4724 StoreFlushContext ctx = this.prepareFlushResult.storeFlushCtxs == null ?
4725 null : this.prepareFlushResult.storeFlushCtxs.get(family);
4726 if (ctx != null) {
4727 long snapshotSize = store.getFlushableSize();
4728 ctx.abort();
4729 this.addAndGetGlobalMemstoreSize(-snapshotSize);
4730 this.prepareFlushResult.storeFlushCtxs.remove(family);
4731 }
4732 }
4733 }
4734
4735
4736 dropMemstoreContentsForSeqId(regionEvent.getLogSequenceNumber(), store);
4737 if (storeSeqId > this.maxFlushedSeqId) {
4738 this.maxFlushedSeqId = storeSeqId;
4739 }
4740 }
4741
4742
4743
4744 dropPrepareFlushIfPossible();
4745
4746
4747 mvcc.await();
4748
4749
4750
4751 this.setReadsEnabled(true);
4752
4753
4754
4755 synchronized (this) {
4756 notifyAll();
4757 }
4758 }
4759 logRegionFiles();
4760 } finally {
4761 closeRegionOperation(Operation.REPLAY_EVENT);
4762 }
4763 }
4764
4765 void replayWALBulkLoadEventMarker(WALProtos.BulkLoadDescriptor bulkLoadEvent) throws IOException {
4766 checkTargetRegion(bulkLoadEvent.getEncodedRegionName().toByteArray(),
4767 "BulkLoad marker from WAL ", bulkLoadEvent);
4768
4769 if (ServerRegionReplicaUtil.isDefaultReplica(this.getRegionInfo())) {
4770 return;
4771 }
4772
4773 if (LOG.isDebugEnabled()) {
4774 LOG.debug(getRegionInfo().getEncodedName() + " : "
4775 + "Replaying bulkload event marker " + TextFormat.shortDebugString(bulkLoadEvent));
4776 }
4777
4778 boolean multipleFamilies = false;
4779 byte[] family = null;
4780 for (StoreDescriptor storeDescriptor : bulkLoadEvent.getStoresList()) {
4781 byte[] fam = storeDescriptor.getFamilyName().toByteArray();
4782 if (family == null) {
4783 family = fam;
4784 } else if (!Bytes.equals(family, fam)) {
4785 multipleFamilies = true;
4786 break;
4787 }
4788 }
4789
4790 startBulkRegionOperation(multipleFamilies);
4791 try {
4792
4793 synchronized (writestate) {
4794
4795
4796
4797
4798
4799
4800 if (bulkLoadEvent.getBulkloadSeqNum() >= 0
4801 && this.lastReplayedOpenRegionSeqId >= bulkLoadEvent.getBulkloadSeqNum()) {
4802 LOG.warn(getRegionInfo().getEncodedName() + " : "
4803 + "Skipping replaying bulkload event :"
4804 + TextFormat.shortDebugString(bulkLoadEvent)
4805 + " because its sequence id is smaller than this region's lastReplayedOpenRegionSeqId"
4806 + " =" + lastReplayedOpenRegionSeqId);
4807
4808 return;
4809 }
4810
4811 for (StoreDescriptor storeDescriptor : bulkLoadEvent.getStoresList()) {
4812
4813 family = storeDescriptor.getFamilyName().toByteArray();
4814 Store store = getStore(family);
4815 if (store == null) {
4816 LOG.warn(getRegionInfo().getEncodedName() + " : "
4817 + "Received a bulk load marker from primary, but the family is not found. "
4818 + "Ignoring. StoreDescriptor:" + storeDescriptor);
4819 continue;
4820 }
4821
4822 List<String> storeFiles = storeDescriptor.getStoreFileList();
4823 for (String storeFile : storeFiles) {
4824 StoreFileInfo storeFileInfo = null;
4825 try {
4826 storeFileInfo = fs.getStoreFileInfo(Bytes.toString(family), storeFile);
4827 store.bulkLoadHFile(storeFileInfo);
4828 } catch(FileNotFoundException ex) {
4829 LOG.warn(getRegionInfo().getEncodedName() + " : "
4830 + ((storeFileInfo != null) ? storeFileInfo.toString() :
4831 (new Path(Bytes.toString(family), storeFile)).toString())
4832 + " doesn't exist any more. Skip loading the file");
4833 }
4834 }
4835 }
4836 }
4837 if (bulkLoadEvent.getBulkloadSeqNum() > 0) {
4838 mvcc.advanceTo(bulkLoadEvent.getBulkloadSeqNum());
4839 }
4840 } finally {
4841 closeBulkRegionOperation();
4842 }
4843 }
4844
4845
4846
4847
4848 private void dropPrepareFlushIfPossible() {
4849 if (writestate.flushing) {
4850 boolean canDrop = true;
4851 if (prepareFlushResult.storeFlushCtxs != null) {
4852 for (Entry<byte[], StoreFlushContext> entry
4853 : prepareFlushResult.storeFlushCtxs.entrySet()) {
4854 Store store = getStore(entry.getKey());
4855 if (store == null) {
4856 continue;
4857 }
4858 if (store.getSnapshotSize() > 0) {
4859 canDrop = false;
4860 break;
4861 }
4862 }
4863 }
4864
4865
4866
4867 if (canDrop) {
4868 writestate.flushing = false;
4869 this.prepareFlushResult = null;
4870 }
4871 }
4872 }
4873
4874 @Override
4875 public boolean refreshStoreFiles() throws IOException {
4876 return refreshStoreFiles(false);
4877 }
4878
4879 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="NN_NAKED_NOTIFY",
4880 justification="Notify is about post replay. Intentional")
4881 protected boolean refreshStoreFiles(boolean force) throws IOException {
4882 if (!force && ServerRegionReplicaUtil.isDefaultReplica(this.getRegionInfo())) {
4883 return false;
4884 }
4885
4886 if (LOG.isDebugEnabled()) {
4887 LOG.debug(getRegionInfo().getEncodedName() + " : "
4888 + "Refreshing store files to see whether we can free up memstore");
4889 }
4890
4891 long totalFreedSize = 0;
4892
4893 long smallestSeqIdInStores = Long.MAX_VALUE;
4894
4895 startRegionOperation();
4896 try {
4897 synchronized (writestate) {
4898 for (Store store : getStores()) {
4899
4900
4901 long maxSeqIdBefore = store.getMaxSequenceId();
4902
4903
4904 store.refreshStoreFiles();
4905
4906 long storeSeqId = store.getMaxSequenceId();
4907 if (storeSeqId < smallestSeqIdInStores) {
4908 smallestSeqIdInStores = storeSeqId;
4909 }
4910
4911
4912 if (storeSeqId > maxSeqIdBefore) {
4913
4914 if (writestate.flushing) {
4915
4916 if (this.prepareFlushResult.flushOpSeqId <= storeSeqId) {
4917 StoreFlushContext ctx = this.prepareFlushResult.storeFlushCtxs == null ?
4918 null : this.prepareFlushResult.storeFlushCtxs.get(store.getFamily().getName());
4919 if (ctx != null) {
4920 long snapshotSize = store.getFlushableSize();
4921 ctx.abort();
4922 this.addAndGetGlobalMemstoreSize(-snapshotSize);
4923 this.prepareFlushResult.storeFlushCtxs.remove(store.getFamily().getName());
4924 totalFreedSize += snapshotSize;
4925 }
4926 }
4927 }
4928
4929
4930 totalFreedSize += dropMemstoreContentsForSeqId(storeSeqId, store);
4931 }
4932 }
4933
4934
4935
4936 dropPrepareFlushIfPossible();
4937
4938
4939
4940 for (Store s : getStores()) {
4941 mvcc.advanceTo(s.getMaxMemstoreTS());
4942 }
4943
4944
4945
4946
4947
4948
4949 if (this.lastReplayedOpenRegionSeqId < smallestSeqIdInStores) {
4950 this.lastReplayedOpenRegionSeqId = smallestSeqIdInStores;
4951 }
4952 }
4953
4954
4955 synchronized (this) {
4956 notifyAll();
4957 }
4958 return totalFreedSize > 0;
4959 } finally {
4960 closeRegionOperation();
4961 }
4962 }
4963
4964 private void logRegionFiles() {
4965 if (LOG.isTraceEnabled()) {
4966 LOG.trace(getRegionInfo().getEncodedName() + " : Store files for region: ");
4967 for (Store s : stores.values()) {
4968 Collection<StoreFile> storeFiles = s.getStorefiles();
4969 if (storeFiles == null) continue;
4970 for (StoreFile sf : storeFiles) {
4971 LOG.trace(getRegionInfo().getEncodedName() + " : " + sf);
4972 }
4973 }
4974 }
4975 }
4976
4977
4978
4979
4980 private void checkTargetRegion(byte[] encodedRegionName, String exceptionMsg, Object payload)
4981 throws WrongRegionException {
4982 if (Bytes.equals(this.getRegionInfo().getEncodedNameAsBytes(), encodedRegionName)) {
4983 return;
4984 }
4985
4986 if (!RegionReplicaUtil.isDefaultReplica(this.getRegionInfo()) &&
4987 Bytes.equals(encodedRegionName,
4988 this.fs.getRegionInfoForFS().getEncodedNameAsBytes())) {
4989 return;
4990 }
4991
4992 throw new WrongRegionException(exceptionMsg + payload
4993 + " targetted for region " + Bytes.toStringBinary(encodedRegionName)
4994 + " does not match this region: " + this.getRegionInfo());
4995 }
4996
4997
4998
4999
5000
5001
5002
5003 protected boolean restoreEdit(final Store s, final Cell cell) {
5004 long kvSize = s.add(cell);
5005 if (this.rsAccounting != null) {
5006 rsAccounting.addAndGetRegionReplayEditsSize(getRegionInfo().getRegionName(), kvSize);
5007 }
5008 return isFlushSize(this.addAndGetGlobalMemstoreSize(kvSize));
5009 }
5010
5011
5012
5013
5014
5015
5016
5017 private static boolean isZeroLengthThenDelete(final FileSystem fs, final Path p)
5018 throws IOException {
5019 FileStatus stat = fs.getFileStatus(p);
5020 if (stat.getLen() > 0) return false;
5021 LOG.warn("File " + p + " is zero-length, deleting.");
5022 fs.delete(p, false);
5023 return true;
5024 }
5025
5026 protected HStore instantiateHStore(final HColumnDescriptor family) throws IOException {
5027 if (family.isMobEnabled()) {
5028 if (HFile.getFormatVersion(this.conf) < HFile.MIN_FORMAT_VERSION_WITH_TAGS) {
5029 throw new IOException("A minimum HFile version of "
5030 + HFile.MIN_FORMAT_VERSION_WITH_TAGS
5031 + " is required for MOB feature. Consider setting " + HFile.FORMAT_VERSION_KEY
5032 + " accordingly.");
5033 }
5034 return new HMobStore(this, family, this.conf);
5035 }
5036 return new HStore(this, family, this.conf);
5037 }
5038
5039 @Override
5040 public Store getStore(final byte[] column) {
5041 return this.stores.get(column);
5042 }
5043
5044
5045
5046
5047
5048 private Store getStore(Cell cell) {
5049 for (Map.Entry<byte[], Store> famStore : stores.entrySet()) {
5050 if (Bytes.equals(
5051 cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength(),
5052 famStore.getKey(), 0, famStore.getKey().length)) {
5053 return famStore.getValue();
5054 }
5055 }
5056
5057 return null;
5058 }
5059
5060 @Override
5061 public List<Store> getStores() {
5062 List<Store> list = new ArrayList<Store>(stores.size());
5063 list.addAll(stores.values());
5064 return list;
5065 }
5066
5067 @Override
5068 public List<String> getStoreFileList(final byte [][] columns)
5069 throws IllegalArgumentException {
5070 List<String> storeFileNames = new ArrayList<String>();
5071 synchronized(closeLock) {
5072 for(byte[] column : columns) {
5073 Store store = this.stores.get(column);
5074 if (store == null) {
5075 throw new IllegalArgumentException("No column family : " +
5076 new String(column) + " available");
5077 }
5078 Collection<StoreFile> storeFiles = store.getStorefiles();
5079 if (storeFiles == null) continue;
5080 for (StoreFile storeFile: storeFiles) {
5081 storeFileNames.add(storeFile.getPath().toString());
5082 }
5083
5084 logRegionFiles();
5085 }
5086 }
5087 return storeFileNames;
5088 }
5089
5090
5091
5092
5093
5094
5095 void checkRow(final byte [] row, String op) throws IOException {
5096 if (!rowIsInRange(getRegionInfo(), row)) {
5097 throw new WrongRegionException("Requested row out of range for " +
5098 op + " on HRegion " + this + ", startKey='" +
5099 Bytes.toStringBinary(getRegionInfo().getStartKey()) + "', getEndKey()='" +
5100 Bytes.toStringBinary(getRegionInfo().getEndKey()) + "', row='" +
5101 Bytes.toStringBinary(row) + "'");
5102 }
5103 }
5104
5105
5106
5107
5108
5109
5110
5111
5112 public RowLock getRowLock(byte[] row) throws IOException {
5113 return getRowLock(row, false);
5114 }
5115
5116
5117
5118
5119
5120
5121
5122
5123
5124
5125
5126 public RowLock getRowLock(byte[] row, boolean readLock) throws IOException {
5127
5128 checkRow(row, "row lock");
5129
5130 HashedBytes rowKey = new HashedBytes(row);
5131
5132 RowLockContext rowLockContext = null;
5133 RowLockImpl result = null;
5134 TraceScope traceScope = null;
5135
5136
5137 if (Trace.isTracing()) {
5138 traceScope = Trace.startSpan("HRegion.getRowLock");
5139 traceScope.getSpan().addTimelineAnnotation("Getting a " + (readLock?"readLock":"writeLock"));
5140 }
5141
5142 try {
5143
5144
5145 while (result == null) {
5146
5147
5148
5149 rowLockContext = new RowLockContext(rowKey);
5150 RowLockContext existingContext = lockedRows.putIfAbsent(rowKey, rowLockContext);
5151
5152
5153 if (existingContext != null) {
5154 rowLockContext = existingContext;
5155 }
5156
5157
5158
5159
5160 if (readLock) {
5161 result = rowLockContext.newReadLock();
5162 } else {
5163 result = rowLockContext.newWriteLock();
5164 }
5165 }
5166 if (!result.getLock().tryLock(this.rowLockWaitDuration, TimeUnit.MILLISECONDS)) {
5167 if (traceScope != null) {
5168 traceScope.getSpan().addTimelineAnnotation("Failed to get row lock");
5169 }
5170 result = null;
5171
5172 rowLockContext.cleanUp();
5173 throw new IOException("Timed out waiting for lock for row: " + rowKey);
5174 }
5175 return result;
5176 } catch (InterruptedException ie) {
5177 LOG.warn("Thread interrupted waiting for lock on row: " + rowKey);
5178 InterruptedIOException iie = new InterruptedIOException();
5179 iie.initCause(ie);
5180 if (traceScope != null) {
5181 traceScope.getSpan().addTimelineAnnotation("Interrupted exception getting row lock");
5182 }
5183 Thread.currentThread().interrupt();
5184 throw iie;
5185 } finally {
5186 if (traceScope != null) {
5187 traceScope.close();
5188 }
5189 }
5190 }
5191
5192 @Override
5193 public void releaseRowLocks(List<RowLock> rowLocks) {
5194 if (rowLocks != null) {
5195 for (RowLock rowLock : rowLocks) {
5196 rowLock.release();
5197 }
5198 rowLocks.clear();
5199 }
5200 }
5201
5202 @VisibleForTesting
5203 class RowLockContext {
5204 private final HashedBytes row;
5205 final ReadWriteLock readWriteLock = new ReentrantReadWriteLock(true);
5206 final AtomicBoolean usable = new AtomicBoolean(true);
5207 final AtomicInteger count = new AtomicInteger(0);
5208 final Object lock = new Object();
5209
5210 RowLockContext(HashedBytes row) {
5211 this.row = row;
5212 }
5213
5214 RowLockImpl newWriteLock() {
5215 Lock l = readWriteLock.writeLock();
5216 return getRowLock(l);
5217 }
5218 RowLockImpl newReadLock() {
5219 Lock l = readWriteLock.readLock();
5220 return getRowLock(l);
5221 }
5222
5223 private RowLockImpl getRowLock(Lock l) {
5224 count.incrementAndGet();
5225 synchronized (lock) {
5226 if (usable.get()) {
5227 return new RowLockImpl(this, l);
5228 } else {
5229 return null;
5230 }
5231 }
5232 }
5233
5234 void cleanUp() {
5235 long c = count.decrementAndGet();
5236 if (c <= 0) {
5237 synchronized (lock) {
5238 if (count.get() <= 0 ){
5239 usable.set(false);
5240 RowLockContext removed = lockedRows.remove(row);
5241 assert removed == this: "we should never remove a different context";
5242 }
5243 }
5244 }
5245 }
5246
5247 @Override
5248 public String toString() {
5249 return "RowLockContext{" +
5250 "row=" + row +
5251 ", readWriteLock=" + readWriteLock +
5252 ", count=" + count +
5253 '}';
5254 }
5255 }
5256
5257
5258
5259
5260 public static class RowLockImpl implements RowLock {
5261 private final RowLockContext context;
5262 private final Lock lock;
5263
5264 public RowLockImpl(RowLockContext context, Lock lock) {
5265 this.context = context;
5266 this.lock = lock;
5267 }
5268
5269 public Lock getLock() {
5270 return lock;
5271 }
5272
5273 @VisibleForTesting
5274 public RowLockContext getContext() {
5275 return context;
5276 }
5277
5278 @Override
5279 public void release() {
5280 lock.unlock();
5281 context.cleanUp();
5282 }
5283
5284 @Override
5285 public String toString() {
5286 return "RowLockImpl{" +
5287 "context=" + context +
5288 ", lock=" + lock +
5289 '}';
5290 }
5291 }
5292
5293
5294
5295
5296
5297
5298
5299 private static boolean hasMultipleColumnFamilies(Collection<Pair<byte[], String>> familyPaths) {
5300 boolean multipleFamilies = false;
5301 byte[] family = null;
5302 for (Pair<byte[], String> pair : familyPaths) {
5303 byte[] fam = pair.getFirst();
5304 if (family == null) {
5305 family = fam;
5306 } else if (!Bytes.equals(family, fam)) {
5307 multipleFamilies = true;
5308 break;
5309 }
5310 }
5311 return multipleFamilies;
5312 }
5313
5314 @Override
5315 public boolean bulkLoadHFiles(Collection<Pair<byte[], String>> familyPaths, boolean assignSeqId,
5316 BulkLoadListener bulkLoadListener) throws IOException {
5317 long seqId = -1;
5318 Map<byte[], List<Path>> storeFiles = new TreeMap<byte[], List<Path>>(Bytes.BYTES_COMPARATOR);
5319 Preconditions.checkNotNull(familyPaths);
5320
5321 startBulkRegionOperation(hasMultipleColumnFamilies(familyPaths));
5322 try {
5323 this.writeRequestsCount.increment();
5324
5325
5326
5327
5328 List<IOException> ioes = new ArrayList<IOException>();
5329 List<Pair<byte[], String>> failures = new ArrayList<Pair<byte[], String>>();
5330 for (Pair<byte[], String> p : familyPaths) {
5331 byte[] familyName = p.getFirst();
5332 String path = p.getSecond();
5333
5334 Store store = getStore(familyName);
5335 if (store == null) {
5336 IOException ioe = new org.apache.hadoop.hbase.DoNotRetryIOException(
5337 "No such column family " + Bytes.toStringBinary(familyName));
5338 ioes.add(ioe);
5339 } else {
5340 try {
5341 store.assertBulkLoadHFileOk(new Path(path));
5342 } catch (WrongRegionException wre) {
5343
5344 failures.add(p);
5345 } catch (IOException ioe) {
5346
5347 ioes.add(ioe);
5348 }
5349 }
5350 }
5351
5352
5353 if (ioes.size() != 0) {
5354 IOException e = MultipleIOException.createIOException(ioes);
5355 LOG.error("There were one or more IO errors when checking if the bulk load is ok.", e);
5356 throw e;
5357 }
5358
5359
5360 if (failures.size() != 0) {
5361 StringBuilder list = new StringBuilder();
5362 for (Pair<byte[], String> p : failures) {
5363 list.append("\n").append(Bytes.toString(p.getFirst())).append(" : ")
5364 .append(p.getSecond());
5365 }
5366
5367 LOG.warn("There was a recoverable bulk load failure likely due to a" +
5368 " split. These (family, HFile) pairs were not loaded: " + list);
5369 return false;
5370 }
5371
5372
5373
5374
5375
5376
5377 if (assignSeqId) {
5378 FlushResult fs = flushcache(true, false);
5379 if (fs.isFlushSucceeded()) {
5380 seqId = ((FlushResultImpl)fs).flushSequenceId;
5381 } else if (fs.getResult() == FlushResult.Result.CANNOT_FLUSH_MEMSTORE_EMPTY) {
5382 seqId = ((FlushResultImpl)fs).flushSequenceId;
5383 } else {
5384 throw new IOException("Could not bulk load with an assigned sequential ID because the "+
5385 "flush didn't run. Reason for not flushing: " + ((FlushResultImpl)fs).failureReason);
5386 }
5387 }
5388
5389 for (Pair<byte[], String> p : familyPaths) {
5390 byte[] familyName = p.getFirst();
5391 String path = p.getSecond();
5392 Store store = getStore(familyName);
5393 try {
5394 String finalPath = path;
5395 if (bulkLoadListener != null) {
5396 finalPath = bulkLoadListener.prepareBulkLoad(familyName, path);
5397 }
5398 Path commitedStoreFile = store.bulkLoadHFile(finalPath, seqId);
5399
5400 if(storeFiles.containsKey(familyName)) {
5401 storeFiles.get(familyName).add(commitedStoreFile);
5402 } else {
5403 List<Path> storeFileNames = new ArrayList<Path>();
5404 storeFileNames.add(commitedStoreFile);
5405 storeFiles.put(familyName, storeFileNames);
5406 }
5407 if (bulkLoadListener != null) {
5408 bulkLoadListener.doneBulkLoad(familyName, path);
5409 }
5410 } catch (IOException ioe) {
5411
5412
5413
5414
5415 LOG.error("There was a partial failure due to IO when attempting to" +
5416 " load " + Bytes.toString(p.getFirst()) + " : " + p.getSecond(), ioe);
5417 if (bulkLoadListener != null) {
5418 try {
5419 bulkLoadListener.failedBulkLoad(familyName, path);
5420 } catch (Exception ex) {
5421 LOG.error("Error while calling failedBulkLoad for family " +
5422 Bytes.toString(familyName) + " with path " + path, ex);
5423 }
5424 }
5425 throw ioe;
5426 }
5427 }
5428
5429 return true;
5430 } finally {
5431 if (wal != null && !storeFiles.isEmpty()) {
5432
5433 try {
5434 WALProtos.BulkLoadDescriptor loadDescriptor = ProtobufUtil.toBulkLoadDescriptor(
5435 this.getRegionInfo().getTable(),
5436 ByteStringer.wrap(this.getRegionInfo().getEncodedNameAsBytes()), storeFiles, seqId);
5437 WALUtil.writeBulkLoadMarkerAndSync(wal, this.htableDescriptor, getRegionInfo(),
5438 loadDescriptor, mvcc);
5439 } catch (IOException ioe) {
5440 if (this.rsServices != null) {
5441
5442
5443 this.rsServices.abort("Failed to write bulk load event into WAL.", ioe);
5444 }
5445 }
5446 }
5447
5448 closeBulkRegionOperation();
5449 }
5450 }
5451
5452 @Override
5453 public boolean equals(Object o) {
5454 return o instanceof HRegion && Bytes.equals(getRegionInfo().getRegionName(),
5455 ((HRegion) o).getRegionInfo().getRegionName());
5456 }
5457
5458 @Override
5459 public int hashCode() {
5460 return Bytes.hashCode(getRegionInfo().getRegionName());
5461 }
5462
5463 @Override
5464 public String toString() {
5465 return getRegionInfo().getRegionNameAsString();
5466 }
5467
5468
5469
5470
5471 class RegionScannerImpl implements RegionScanner {
5472
5473 KeyValueHeap storeHeap = null;
5474
5475
5476 KeyValueHeap joinedHeap = null;
5477
5478
5479
5480 protected Cell joinedContinuationRow = null;
5481 private boolean filterClosed = false;
5482
5483 protected final int isScan;
5484 protected final byte[] stopRow;
5485 protected final HRegion region;
5486
5487 private final long readPt;
5488 private final long maxResultSize;
5489 private final ScannerContext defaultScannerContext;
5490 private final FilterWrapper filter;
5491
5492 @Override
5493 public HRegionInfo getRegionInfo() {
5494 return region.getRegionInfo();
5495 }
5496
5497 RegionScannerImpl(Scan scan, List<KeyValueScanner> additionalScanners, HRegion region)
5498 throws IOException {
5499 this.region = region;
5500 this.maxResultSize = scan.getMaxResultSize();
5501 if (scan.hasFilter()) {
5502 this.filter = new FilterWrapper(scan.getFilter());
5503 } else {
5504 this.filter = null;
5505 }
5506
5507
5508
5509
5510
5511
5512 defaultScannerContext = ScannerContext.newBuilder().setBatchLimit(scan.getBatch()).build();
5513
5514 if (Bytes.equals(scan.getStopRow(), HConstants.EMPTY_END_ROW) && !scan.isGetScan()) {
5515 this.stopRow = null;
5516 } else {
5517 this.stopRow = scan.getStopRow();
5518 }
5519
5520
5521 this.isScan = scan.isGetScan() ? -1 : 0;
5522
5523
5524
5525 IsolationLevel isolationLevel = scan.getIsolationLevel();
5526 synchronized(scannerReadPoints) {
5527 this.readPt = getReadpoint(isolationLevel);
5528 scannerReadPoints.put(this, this.readPt);
5529 }
5530
5531
5532
5533 List<KeyValueScanner> scanners = new ArrayList<KeyValueScanner>();
5534 List<KeyValueScanner> joinedScanners = new ArrayList<KeyValueScanner>();
5535 if (additionalScanners != null) {
5536 scanners.addAll(additionalScanners);
5537 }
5538
5539 for (Map.Entry<byte[], NavigableSet<byte[]>> entry : scan.getFamilyMap().entrySet()) {
5540 Store store = stores.get(entry.getKey());
5541 KeyValueScanner scanner;
5542 try {
5543 scanner = store.getScanner(scan, entry.getValue(), this.readPt);
5544 } catch (FileNotFoundException e) {
5545 throw handleFileNotFound(e);
5546 }
5547 if (this.filter == null || !scan.doLoadColumnFamiliesOnDemand()
5548 || this.filter.isFamilyEssential(entry.getKey())) {
5549 scanners.add(scanner);
5550 } else {
5551 joinedScanners.add(scanner);
5552 }
5553 }
5554 initializeKVHeap(scanners, joinedScanners, region);
5555 }
5556
5557 protected void initializeKVHeap(List<KeyValueScanner> scanners,
5558 List<KeyValueScanner> joinedScanners, HRegion region)
5559 throws IOException {
5560 this.storeHeap = new KeyValueHeap(scanners, region.comparator);
5561 if (!joinedScanners.isEmpty()) {
5562 this.joinedHeap = new KeyValueHeap(joinedScanners, region.comparator);
5563 }
5564 }
5565
5566 @Override
5567 public long getMaxResultSize() {
5568 return maxResultSize;
5569 }
5570
5571 @Override
5572 public long getMvccReadPoint() {
5573 return this.readPt;
5574 }
5575
5576 @Override
5577 public int getBatch() {
5578 return this.defaultScannerContext.getBatchLimit();
5579 }
5580
5581
5582
5583
5584
5585
5586 protected void resetFilters() throws IOException {
5587 if (filter != null) {
5588 filter.reset();
5589 }
5590 }
5591
5592 @Override
5593 public boolean next(List<Cell> outResults)
5594 throws IOException {
5595
5596 return next(outResults, defaultScannerContext);
5597 }
5598
5599 @Override
5600 public synchronized boolean next(List<Cell> outResults, ScannerContext scannerContext)
5601 throws IOException {
5602 if (this.filterClosed) {
5603 throw new UnknownScannerException("Scanner was closed (timed out?) " +
5604 "after we renewed it. Could be caused by a very slow scanner " +
5605 "or a lengthy garbage collection");
5606 }
5607 startRegionOperation(Operation.SCAN);
5608 readRequestsCount.increment();
5609 try {
5610 return nextRaw(outResults, scannerContext);
5611 } finally {
5612 closeRegionOperation(Operation.SCAN);
5613 }
5614 }
5615
5616 @Override
5617 public boolean nextRaw(List<Cell> outResults) throws IOException {
5618
5619 return nextRaw(outResults, defaultScannerContext);
5620 }
5621
5622 @Override
5623 public boolean nextRaw(List<Cell> outResults, ScannerContext scannerContext)
5624 throws IOException {
5625 if (storeHeap == null) {
5626
5627 throw new UnknownScannerException("Scanner was closed");
5628 }
5629 boolean moreValues;
5630 if (outResults.isEmpty()) {
5631
5632
5633 moreValues = nextInternal(outResults, scannerContext);
5634 } else {
5635 List<Cell> tmpList = new ArrayList<Cell>();
5636 moreValues = nextInternal(tmpList, scannerContext);
5637 outResults.addAll(tmpList);
5638 }
5639
5640
5641
5642
5643 if (!scannerContext.partialResultFormed()) resetFilters();
5644
5645 if (isFilterDoneInternal()) {
5646 moreValues = false;
5647 }
5648 return moreValues;
5649 }
5650
5651
5652
5653
5654 private boolean populateFromJoinedHeap(List<Cell> results, ScannerContext scannerContext)
5655 throws IOException {
5656 assert joinedContinuationRow != null;
5657 boolean moreValues =
5658 populateResult(results, this.joinedHeap, scannerContext,
5659 joinedContinuationRow.getRowArray(), joinedContinuationRow.getRowOffset(),
5660 joinedContinuationRow.getRowLength());
5661
5662 if (!scannerContext.checkAnyLimitReached(LimitScope.BETWEEN_CELLS)) {
5663
5664 joinedContinuationRow = null;
5665 }
5666
5667
5668 Collections.sort(results, comparator);
5669 return moreValues;
5670 }
5671
5672
5673
5674
5675
5676
5677
5678
5679
5680
5681
5682 private boolean populateResult(List<Cell> results, KeyValueHeap heap,
5683 ScannerContext scannerContext, byte[] currentRow, int offset, short length)
5684 throws IOException {
5685 Cell nextKv;
5686 boolean moreCellsInRow = false;
5687 boolean tmpKeepProgress = scannerContext.getKeepProgress();
5688
5689 LimitScope limitScope = LimitScope.BETWEEN_CELLS;
5690 try {
5691 do {
5692
5693
5694
5695 scannerContext.setKeepProgress(true);
5696 heap.next(results, scannerContext);
5697 scannerContext.setKeepProgress(tmpKeepProgress);
5698
5699 nextKv = heap.peek();
5700 moreCellsInRow = moreCellsInRow(nextKv, currentRow, offset, length);
5701 if (!moreCellsInRow) incrementCountOfRowsScannedMetric(scannerContext);
5702
5703 if (scannerContext.checkBatchLimit(limitScope)) {
5704 return scannerContext.setScannerState(NextState.BATCH_LIMIT_REACHED).hasMoreValues();
5705 } else if (scannerContext.checkSizeLimit(limitScope)) {
5706 ScannerContext.NextState state =
5707 moreCellsInRow? NextState.SIZE_LIMIT_REACHED_MID_ROW: NextState.SIZE_LIMIT_REACHED;
5708 return scannerContext.setScannerState(state).hasMoreValues();
5709 } else if (scannerContext.checkTimeLimit(limitScope)) {
5710 ScannerContext.NextState state =
5711 moreCellsInRow? NextState.TIME_LIMIT_REACHED_MID_ROW: NextState.TIME_LIMIT_REACHED;
5712 return scannerContext.setScannerState(state).hasMoreValues();
5713 }
5714 } while (moreCellsInRow);
5715 } catch (FileNotFoundException e) {
5716 throw handleFileNotFound(e);
5717 }
5718 return nextKv != null;
5719 }
5720
5721
5722
5723
5724
5725
5726
5727
5728
5729
5730
5731 private boolean moreCellsInRow(final Cell nextKv, byte[] currentRow, int offset,
5732 short length) {
5733 return nextKv != null && CellUtil.matchingRow(nextKv, currentRow, offset, length);
5734 }
5735
5736
5737
5738
5739 @Override
5740 public synchronized boolean isFilterDone() throws IOException {
5741 return isFilterDoneInternal();
5742 }
5743
5744 private boolean isFilterDoneInternal() throws IOException {
5745 return this.filter != null && this.filter.filterAllRemaining();
5746 }
5747
5748 private boolean nextInternal(List<Cell> results, ScannerContext scannerContext)
5749 throws IOException {
5750 if (!results.isEmpty()) {
5751 throw new IllegalArgumentException("First parameter should be an empty list");
5752 }
5753 if (scannerContext == null) {
5754 throw new IllegalArgumentException("Scanner context cannot be null");
5755 }
5756 RpcCallContext rpcCall = RpcServer.getCurrentCall();
5757
5758
5759
5760
5761 int initialBatchProgress = scannerContext.getBatchProgress();
5762 long initialSizeProgress = scannerContext.getSizeProgress();
5763 long initialTimeProgress = scannerContext.getTimeProgress();
5764
5765
5766
5767
5768
5769
5770 while (true) {
5771
5772
5773 if (scannerContext.getKeepProgress()) {
5774
5775 scannerContext.setProgress(initialBatchProgress, initialSizeProgress,
5776 initialTimeProgress);
5777 } else {
5778 scannerContext.clearProgress();
5779 }
5780
5781 if (rpcCall != null) {
5782
5783
5784
5785
5786 long afterTime = rpcCall.disconnectSince();
5787 if (afterTime >= 0) {
5788 throw new CallerDisconnectedException(
5789 "Aborting on region " + getRegionInfo().getRegionNameAsString() + ", call " +
5790 this + " after " + afterTime + " ms, since " +
5791 "caller disconnected");
5792 }
5793 }
5794
5795
5796 Cell current = this.storeHeap.peek();
5797
5798 byte[] currentRow = null;
5799 int offset = 0;
5800 short length = 0;
5801 if (current != null) {
5802 currentRow = current.getRowArray();
5803 offset = current.getRowOffset();
5804 length = current.getRowLength();
5805 }
5806
5807 boolean stopRow = isStopRow(currentRow, offset, length);
5808
5809
5810
5811
5812 boolean hasFilterRow = this.filter != null && this.filter.hasFilterRow();
5813
5814
5815
5816
5817
5818 if (hasFilterRow) {
5819 if (LOG.isTraceEnabled()) {
5820 LOG.trace("filter#hasFilterRow is true which prevents partial results from being "
5821 + " formed. Changing scope of limits that may create partials");
5822 }
5823 scannerContext.setSizeLimitScope(LimitScope.BETWEEN_ROWS);
5824 scannerContext.setTimeLimitScope(LimitScope.BETWEEN_ROWS);
5825 }
5826
5827
5828
5829 if (joinedContinuationRow == null) {
5830
5831 if (stopRow) {
5832 if (hasFilterRow) {
5833 filter.filterRowCells(results);
5834 }
5835 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
5836 }
5837
5838
5839
5840 if (filterRowKey(currentRow, offset, length)) {
5841 incrementCountOfRowsFilteredMetric(scannerContext);
5842
5843
5844
5845 incrementCountOfRowsScannedMetric(scannerContext);
5846 boolean moreRows = nextRow(scannerContext, currentRow, offset, length);
5847 if (!moreRows) {
5848 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
5849 }
5850 results.clear();
5851 continue;
5852 }
5853
5854
5855 populateResult(results, this.storeHeap, scannerContext, currentRow, offset, length);
5856
5857 if (scannerContext.checkAnyLimitReached(LimitScope.BETWEEN_CELLS)) {
5858 if (hasFilterRow) {
5859 throw new IncompatibleFilterException(
5860 "Filter whose hasFilterRow() returns true is incompatible with scans that must "
5861 + " stop mid-row because of a limit. ScannerContext:" + scannerContext);
5862 }
5863 return true;
5864 }
5865
5866 Cell nextKv = this.storeHeap.peek();
5867 stopRow = nextKv == null ||
5868 isStopRow(nextKv.getRowArray(), nextKv.getRowOffset(), nextKv.getRowLength());
5869
5870 final boolean isEmptyRow = results.isEmpty();
5871
5872
5873
5874 FilterWrapper.FilterRowRetCode ret = FilterWrapper.FilterRowRetCode.NOT_CALLED;
5875 if (hasFilterRow) {
5876 ret = filter.filterRowCellsWithRet(results);
5877
5878
5879
5880
5881 long timeProgress = scannerContext.getTimeProgress();
5882 if (scannerContext.getKeepProgress()) {
5883 scannerContext.setProgress(initialBatchProgress, initialSizeProgress,
5884 initialTimeProgress);
5885 } else {
5886 scannerContext.clearProgress();
5887 }
5888 scannerContext.setTimeProgress(timeProgress);
5889 scannerContext.incrementBatchProgress(results.size());
5890 for (Cell cell : results) {
5891 scannerContext.incrementSizeProgress(CellUtil.estimatedHeapSizeOfWithoutTags(cell));
5892 }
5893 }
5894
5895 if (isEmptyRow || ret == FilterWrapper.FilterRowRetCode.EXCLUDE || filterRow()) {
5896 incrementCountOfRowsFilteredMetric(scannerContext);
5897 results.clear();
5898 boolean moreRows = nextRow(scannerContext, currentRow, offset, length);
5899 if (!moreRows) {
5900 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
5901 }
5902
5903
5904
5905 if (!stopRow) continue;
5906 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
5907 }
5908
5909
5910
5911
5912
5913 if (this.joinedHeap != null) {
5914 boolean mayHaveData = joinedHeapMayHaveData(currentRow, offset, length);
5915 if (mayHaveData) {
5916 joinedContinuationRow = current;
5917 populateFromJoinedHeap(results, scannerContext);
5918
5919 if (scannerContext.checkAnyLimitReached(LimitScope.BETWEEN_CELLS)) {
5920 return true;
5921 }
5922 }
5923 }
5924 } else {
5925
5926 populateFromJoinedHeap(results, scannerContext);
5927 if (scannerContext.checkAnyLimitReached(LimitScope.BETWEEN_CELLS)) {
5928 return true;
5929 }
5930 }
5931
5932
5933 if (joinedContinuationRow != null) {
5934 return scannerContext.setScannerState(NextState.MORE_VALUES).hasMoreValues();
5935 }
5936
5937
5938
5939
5940 if (results.isEmpty()) {
5941 incrementCountOfRowsFilteredMetric(scannerContext);
5942 boolean moreRows = nextRow(scannerContext, currentRow, offset, length);
5943 if (!moreRows) {
5944 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
5945 }
5946 if (!stopRow) continue;
5947 }
5948
5949 if (stopRow) {
5950 return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
5951 } else {
5952 return scannerContext.setScannerState(NextState.MORE_VALUES).hasMoreValues();
5953 }
5954 }
5955 }
5956
5957 protected void incrementCountOfRowsFilteredMetric(ScannerContext scannerContext) {
5958 if (scannerContext == null || !scannerContext.isTrackingMetrics()) return;
5959
5960 scannerContext.getMetrics().countOfRowsFiltered.incrementAndGet();
5961 }
5962
5963 protected void incrementCountOfRowsScannedMetric(ScannerContext scannerContext) {
5964 if (scannerContext == null || !scannerContext.isTrackingMetrics()) return;
5965
5966 scannerContext.getMetrics().countOfRowsScanned.incrementAndGet();
5967 }
5968
5969
5970
5971
5972
5973
5974
5975
5976 private boolean joinedHeapMayHaveData(byte[] currentRow, int offset, short length)
5977 throws IOException {
5978 Cell nextJoinedKv = joinedHeap.peek();
5979 boolean matchCurrentRow =
5980 nextJoinedKv != null && CellUtil.matchingRow(nextJoinedKv, currentRow, offset, length);
5981 boolean matchAfterSeek = false;
5982
5983
5984
5985 if (!matchCurrentRow) {
5986 Cell firstOnCurrentRow = KeyValueUtil.createFirstOnRow(currentRow, offset, length);
5987 boolean seekSuccessful = this.joinedHeap.requestSeek(firstOnCurrentRow, true, true);
5988 matchAfterSeek =
5989 seekSuccessful && joinedHeap.peek() != null
5990 && CellUtil.matchingRow(joinedHeap.peek(), currentRow, offset, length);
5991 }
5992
5993 return matchCurrentRow || matchAfterSeek;
5994 }
5995
5996
5997
5998
5999
6000
6001
6002
6003 private boolean filterRow() throws IOException {
6004
6005
6006 return filter != null && (!filter.hasFilterRow())
6007 && filter.filterRow();
6008 }
6009
6010 private boolean filterRowKey(byte[] row, int offset, short length) throws IOException {
6011 return filter != null
6012 && filter.filterRowKey(row, offset, length);
6013 }
6014
6015 protected boolean nextRow(ScannerContext scannerContext, byte[] currentRow, int offset,
6016 short length) throws IOException {
6017 assert this.joinedContinuationRow == null:
6018 "Trying to go to next row during joinedHeap read.";
6019 Cell next;
6020 while ((next = this.storeHeap.peek()) != null &&
6021 CellUtil.matchingRow(next, currentRow, offset, length)) {
6022 this.storeHeap.next(MOCKED_LIST);
6023 }
6024 resetFilters();
6025
6026
6027 return this.region.getCoprocessorHost() == null
6028 || this.region.getCoprocessorHost()
6029 .postScannerFilterRow(this, currentRow, offset, length);
6030 }
6031
6032 protected boolean isStopRow(byte[] currentRow, int offset, short length) {
6033 return currentRow == null ||
6034 (stopRow != null &&
6035 comparator.compareRows(stopRow, 0, stopRow.length,
6036 currentRow, offset, length) <= isScan);
6037 }
6038
6039 @Override
6040 public synchronized void close() {
6041 if (storeHeap != null) {
6042 storeHeap.close();
6043 storeHeap = null;
6044 }
6045 if (joinedHeap != null) {
6046 joinedHeap.close();
6047 joinedHeap = null;
6048 }
6049
6050 scannerReadPoints.remove(this);
6051 this.filterClosed = true;
6052 }
6053
6054 KeyValueHeap getStoreHeapForTesting() {
6055 return storeHeap;
6056 }
6057
6058 @Override
6059 public synchronized boolean reseek(byte[] row) throws IOException {
6060 if (row == null) {
6061 throw new IllegalArgumentException("Row cannot be null.");
6062 }
6063 boolean result = false;
6064 startRegionOperation();
6065 KeyValue kv = KeyValueUtil.createFirstOnRow(row);
6066 try {
6067
6068 result = this.storeHeap.requestSeek(kv, true, true);
6069 if (this.joinedHeap != null) {
6070 result = this.joinedHeap.requestSeek(kv, true, true) || result;
6071 }
6072 } catch (FileNotFoundException e) {
6073 throw handleFileNotFound(e);
6074 } finally {
6075 closeRegionOperation();
6076 }
6077 return result;
6078 }
6079
6080 private IOException handleFileNotFound(FileNotFoundException fnfe) throws IOException {
6081
6082
6083 try {
6084 region.refreshStoreFiles(true);
6085 return new IOException("unable to read store file");
6086 } catch (IOException e) {
6087 String msg = "a store file got lost: " + fnfe.getMessage();
6088 LOG.error(msg);
6089 LOG.error("unable to refresh store files", e);
6090 abortRegionServer(msg);
6091 return new NotServingRegionException(getRegionInfo().getRegionNameAsString() + " closing");
6092 }
6093 }
6094
6095 private void abortRegionServer(String msg) throws IOException {
6096 if (rsServices instanceof HRegionServer) {
6097 ((HRegionServer)rsServices).abort(msg);
6098 }
6099 throw new UnsupportedOperationException("not able to abort RS after: " + msg);
6100 }
6101 }
6102
6103
6104
6105
6106
6107
6108
6109
6110
6111
6112
6113
6114
6115
6116
6117
6118
6119
6120
6121
6122 static HRegion newHRegion(Path tableDir, WAL wal, FileSystem fs,
6123 Configuration conf, HRegionInfo regionInfo, final HTableDescriptor htd,
6124 RegionServerServices rsServices) {
6125 try {
6126 @SuppressWarnings("unchecked")
6127 Class<? extends HRegion> regionClass =
6128 (Class<? extends HRegion>) conf.getClass(HConstants.REGION_IMPL, HRegion.class);
6129
6130 Constructor<? extends HRegion> c =
6131 regionClass.getConstructor(Path.class, WAL.class, FileSystem.class,
6132 Configuration.class, HRegionInfo.class, HTableDescriptor.class,
6133 RegionServerServices.class);
6134
6135 return c.newInstance(tableDir, wal, fs, conf, regionInfo, htd, rsServices);
6136 } catch (Throwable e) {
6137
6138 throw new IllegalStateException("Could not instantiate a region instance.", e);
6139 }
6140 }
6141
6142
6143
6144
6145
6146
6147
6148
6149
6150
6151
6152
6153
6154
6155
6156
6157
6158 public static HRegion createHRegion(final HRegionInfo info, final Path rootDir,
6159 final Configuration conf, final HTableDescriptor hTableDescriptor)
6160 throws IOException {
6161 return createHRegion(info, rootDir, conf, hTableDescriptor, null);
6162 }
6163
6164
6165
6166
6167
6168
6169
6170
6171
6172
6173
6174 public static void closeHRegion(final HRegion r) throws IOException {
6175 if (r == null) return;
6176 r.close();
6177 if (r.getWAL() == null) return;
6178 r.getWAL().close();
6179 }
6180
6181
6182
6183
6184
6185
6186
6187
6188
6189
6190
6191
6192
6193
6194 public static HRegion createHRegion(final HRegionInfo info, final Path rootDir,
6195 final Configuration conf,
6196 final HTableDescriptor hTableDescriptor,
6197 final WAL wal,
6198 final boolean initialize)
6199 throws IOException {
6200 return createHRegion(info, rootDir, conf, hTableDescriptor,
6201 wal, initialize, false);
6202 }
6203
6204
6205
6206
6207
6208
6209
6210
6211
6212
6213
6214
6215
6216
6217
6218 public static HRegion createHRegion(final HRegionInfo info, final Path rootDir,
6219 final Configuration conf,
6220 final HTableDescriptor hTableDescriptor,
6221 final WAL wal,
6222 final boolean initialize, final boolean ignoreWAL)
6223 throws IOException {
6224 Path tableDir = FSUtils.getTableDir(rootDir, info.getTable());
6225 return createHRegion(info, rootDir, tableDir, conf, hTableDescriptor, wal, initialize,
6226 ignoreWAL);
6227 }
6228
6229
6230
6231
6232
6233
6234
6235
6236
6237
6238
6239
6240
6241
6242
6243
6244 public static HRegion createHRegion(final HRegionInfo info, final Path rootDir,
6245 final Path tableDir, final Configuration conf, final HTableDescriptor hTableDescriptor,
6246 final WAL wal, final boolean initialize, final boolean ignoreWAL)
6247 throws IOException {
6248 LOG.info("creating HRegion " + info.getTable().getNameAsString()
6249 + " HTD == " + hTableDescriptor + " RootDir = " + rootDir +
6250 " Table name == " + info.getTable().getNameAsString());
6251 FileSystem fs = FileSystem.get(conf);
6252 HRegionFileSystem.createRegionOnFileSystem(conf, fs, tableDir, info);
6253 WAL effectiveWAL = wal;
6254 if (wal == null && !ignoreWAL) {
6255
6256
6257
6258 Configuration confForWAL = new Configuration(conf);
6259 confForWAL.set(HConstants.HBASE_DIR, rootDir.toString());
6260 effectiveWAL = (new WALFactory(confForWAL,
6261 Collections.<WALActionsListener>singletonList(new MetricsWAL()),
6262 "hregion-" + RandomStringUtils.randomNumeric(8))).
6263 getWAL(info.getEncodedNameAsBytes());
6264 }
6265 HRegion region = HRegion.newHRegion(tableDir,
6266 effectiveWAL, fs, conf, info, hTableDescriptor, null);
6267 if (initialize) region.initialize(null);
6268 return region;
6269 }
6270
6271 public static HRegion createHRegion(final HRegionInfo info, final Path rootDir,
6272 final Configuration conf,
6273 final HTableDescriptor hTableDescriptor,
6274 final WAL wal)
6275 throws IOException {
6276 return createHRegion(info, rootDir, conf, hTableDescriptor, wal, true);
6277 }
6278
6279
6280
6281
6282
6283
6284
6285
6286
6287
6288
6289
6290
6291 public static HRegion openHRegion(final HRegionInfo info,
6292 final HTableDescriptor htd, final WAL wal,
6293 final Configuration conf)
6294 throws IOException {
6295 return openHRegion(info, htd, wal, conf, null, null);
6296 }
6297
6298
6299
6300
6301
6302
6303
6304
6305
6306
6307
6308
6309
6310
6311
6312
6313 public static HRegion openHRegion(final HRegionInfo info,
6314 final HTableDescriptor htd, final WAL wal, final Configuration conf,
6315 final RegionServerServices rsServices,
6316 final CancelableProgressable reporter)
6317 throws IOException {
6318 return openHRegion(FSUtils.getRootDir(conf), info, htd, wal, conf, rsServices, reporter);
6319 }
6320
6321
6322
6323
6324
6325
6326
6327
6328
6329
6330
6331
6332
6333
6334 public static HRegion openHRegion(Path rootDir, final HRegionInfo info,
6335 final HTableDescriptor htd, final WAL wal, final Configuration conf)
6336 throws IOException {
6337 return openHRegion(rootDir, info, htd, wal, conf, null, null);
6338 }
6339
6340
6341
6342
6343
6344
6345
6346
6347
6348
6349
6350
6351
6352
6353
6354
6355 public static HRegion openHRegion(final Path rootDir, final HRegionInfo info,
6356 final HTableDescriptor htd, final WAL wal, final Configuration conf,
6357 final RegionServerServices rsServices,
6358 final CancelableProgressable reporter)
6359 throws IOException {
6360 FileSystem fs = null;
6361 if (rsServices != null) {
6362 fs = rsServices.getFileSystem();
6363 }
6364 if (fs == null) {
6365 fs = FileSystem.get(conf);
6366 }
6367 return openHRegion(conf, fs, rootDir, info, htd, wal, rsServices, reporter);
6368 }
6369
6370
6371
6372
6373
6374
6375
6376
6377
6378
6379
6380
6381
6382
6383
6384 public static HRegion openHRegion(final Configuration conf, final FileSystem fs,
6385 final Path rootDir, final HRegionInfo info, final HTableDescriptor htd, final WAL wal)
6386 throws IOException {
6387 return openHRegion(conf, fs, rootDir, info, htd, wal, null, null);
6388 }
6389
6390
6391
6392
6393
6394
6395
6396
6397
6398
6399
6400
6401
6402
6403
6404
6405
6406 public static HRegion openHRegion(final Configuration conf, final FileSystem fs,
6407 final Path rootDir, final HRegionInfo info, final HTableDescriptor htd, final WAL wal,
6408 final RegionServerServices rsServices, final CancelableProgressable reporter)
6409 throws IOException {
6410 Path tableDir = FSUtils.getTableDir(rootDir, info.getTable());
6411 return openHRegion(conf, fs, rootDir, tableDir, info, htd, wal, rsServices, reporter);
6412 }
6413
6414
6415
6416
6417
6418
6419
6420
6421
6422
6423
6424
6425
6426
6427
6428
6429
6430 public static HRegion openHRegion(final Configuration conf, final FileSystem fs,
6431 final Path rootDir, final Path tableDir, final HRegionInfo info, final HTableDescriptor htd,
6432 final WAL wal, final RegionServerServices rsServices,
6433 final CancelableProgressable reporter)
6434 throws IOException {
6435 if (info == null) throw new NullPointerException("Passed region info is null");
6436 if (LOG.isDebugEnabled()) {
6437 LOG.debug("Opening region: " + info);
6438 }
6439 HRegion r = HRegion.newHRegion(tableDir, wal, fs, conf, info, htd, rsServices);
6440 return r.openHRegion(reporter);
6441 }
6442
6443
6444
6445
6446
6447
6448
6449
6450
6451 public static HRegion openHRegion(final HRegion other, final CancelableProgressable reporter)
6452 throws IOException {
6453 HRegionFileSystem regionFs = other.getRegionFileSystem();
6454 HRegion r = newHRegion(regionFs.getTableDir(), other.getWAL(), regionFs.getFileSystem(),
6455 other.baseConf, other.getRegionInfo(), other.getTableDesc(), null);
6456 return r.openHRegion(reporter);
6457 }
6458
6459 public static Region openHRegion(final Region other, final CancelableProgressable reporter)
6460 throws IOException {
6461 return openHRegion((HRegion)other, reporter);
6462 }
6463
6464
6465
6466
6467
6468
6469
6470 protected HRegion openHRegion(final CancelableProgressable reporter)
6471 throws IOException {
6472
6473 checkCompressionCodecs();
6474
6475
6476 checkEncryption();
6477
6478 checkClassLoading();
6479 this.openSeqNum = initialize(reporter);
6480 this.mvcc.advanceTo(openSeqNum);
6481 if (wal != null && getRegionServerServices() != null && !writestate.readOnly
6482 && !recovering) {
6483
6484
6485
6486 writeRegionOpenMarker(wal, openSeqNum);
6487 }
6488 return this;
6489 }
6490
6491 public static void warmupHRegion(final HRegionInfo info,
6492 final HTableDescriptor htd, final WAL wal, final Configuration conf,
6493 final RegionServerServices rsServices,
6494 final CancelableProgressable reporter)
6495 throws IOException {
6496
6497 if (info == null) throw new NullPointerException("Passed region info is null");
6498
6499 if (LOG.isDebugEnabled()) {
6500 LOG.debug("HRegion.Warming up region: " + info);
6501 }
6502
6503 Path rootDir = FSUtils.getRootDir(conf);
6504 Path tableDir = FSUtils.getTableDir(rootDir, info.getTable());
6505
6506 FileSystem fs = null;
6507 if (rsServices != null) {
6508 fs = rsServices.getFileSystem();
6509 }
6510 if (fs == null) {
6511 fs = FileSystem.get(conf);
6512 }
6513
6514 HRegion r = HRegion.newHRegion(tableDir, wal, fs, conf, info, htd, rsServices);
6515 r.initializeWarmup(reporter);
6516 r.close();
6517 }
6518
6519
6520 private void checkCompressionCodecs() throws IOException {
6521 for (HColumnDescriptor fam: this.htableDescriptor.getColumnFamilies()) {
6522 CompressionTest.testCompression(fam.getCompression());
6523 CompressionTest.testCompression(fam.getCompactionCompression());
6524 }
6525 }
6526
6527 private void checkEncryption() throws IOException {
6528 for (HColumnDescriptor fam: this.htableDescriptor.getColumnFamilies()) {
6529 EncryptionTest.testEncryption(conf, fam.getEncryptionType(), fam.getEncryptionKey());
6530 }
6531 }
6532
6533 private void checkClassLoading() throws IOException {
6534 RegionSplitPolicy.getSplitPolicyClass(this.htableDescriptor, conf);
6535 RegionCoprocessorHost.testTableCoprocessorAttrs(conf, this.htableDescriptor);
6536 }
6537
6538
6539
6540
6541
6542
6543 HRegion createDaughterRegionFromSplits(final HRegionInfo hri) throws IOException {
6544
6545 fs.commitDaughterRegion(hri);
6546
6547
6548 HRegion r = HRegion.newHRegion(this.fs.getTableDir(), this.getWAL(), fs.getFileSystem(),
6549 this.getBaseConf(), hri, this.getTableDesc(), rsServices);
6550 r.readRequestsCount.set(this.getReadRequestsCount() / 2);
6551 r.writeRequestsCount.set(this.getWriteRequestsCount() / 2);
6552 return r;
6553 }
6554
6555
6556
6557
6558
6559
6560
6561 HRegion createMergedRegionFromMerges(final HRegionInfo mergedRegionInfo,
6562 final HRegion region_b) throws IOException {
6563 HRegion r = HRegion.newHRegion(this.fs.getTableDir(), this.getWAL(),
6564 fs.getFileSystem(), this.getBaseConf(), mergedRegionInfo,
6565 this.getTableDesc(), this.rsServices);
6566 r.readRequestsCount.set(this.getReadRequestsCount()
6567 + region_b.getReadRequestsCount());
6568 r.writeRequestsCount.set(this.getWriteRequestsCount()
6569
6570 + region_b.getWriteRequestsCount());
6571 this.fs.commitMergedRegion(mergedRegionInfo);
6572 return r;
6573 }
6574
6575
6576
6577
6578
6579
6580
6581
6582
6583
6584
6585
6586 public static void addRegionToMETA(final HRegion meta, final HRegion r) throws IOException {
6587 meta.checkResources();
6588
6589 byte[] row = r.getRegionInfo().getRegionName();
6590 final long now = EnvironmentEdgeManager.currentTime();
6591 final List<Cell> cells = new ArrayList<Cell>(2);
6592 cells.add(new KeyValue(row, HConstants.CATALOG_FAMILY,
6593 HConstants.REGIONINFO_QUALIFIER, now,
6594 r.getRegionInfo().toByteArray()));
6595
6596 cells.add(new KeyValue(row, HConstants.CATALOG_FAMILY,
6597 HConstants.META_VERSION_QUALIFIER, now,
6598 Bytes.toBytes(HConstants.META_VERSION)));
6599 meta.put(row, HConstants.CATALOG_FAMILY, cells);
6600 }
6601
6602
6603
6604
6605
6606
6607
6608
6609
6610 @Deprecated
6611 public static Path getRegionDir(final Path tabledir, final String name) {
6612 return new Path(tabledir, name);
6613 }
6614
6615
6616
6617
6618
6619
6620
6621
6622
6623 @Deprecated
6624 @VisibleForTesting
6625 public static Path getRegionDir(final Path rootdir, final HRegionInfo info) {
6626 return new Path(
6627 FSUtils.getTableDir(rootdir, info.getTable()), info.getEncodedName());
6628 }
6629
6630
6631
6632
6633
6634
6635
6636
6637
6638 public static boolean rowIsInRange(HRegionInfo info, final byte [] row) {
6639 return ((info.getStartKey().length == 0) ||
6640 (Bytes.compareTo(info.getStartKey(), row) <= 0)) &&
6641 ((info.getEndKey().length == 0) ||
6642 (Bytes.compareTo(info.getEndKey(), row) > 0));
6643 }
6644
6645 public static boolean rowIsInRange(HRegionInfo info, final byte [] row, final int offset,
6646 final short length) {
6647 return ((info.getStartKey().length == 0) ||
6648 (Bytes.compareTo(info.getStartKey(), 0, info.getStartKey().length,
6649 row, offset, length) <= 0)) &&
6650 ((info.getEndKey().length == 0) ||
6651 (Bytes.compareTo(info.getEndKey(), 0, info.getEndKey().length, row, offset, length) > 0));
6652 }
6653
6654
6655
6656
6657
6658
6659
6660 public static HRegion mergeAdjacent(final HRegion srcA, final HRegion srcB)
6661 throws IOException {
6662 HRegion a = srcA;
6663 HRegion b = srcB;
6664
6665
6666
6667 if (srcA.getRegionInfo().getStartKey() == null) {
6668 if (srcB.getRegionInfo().getStartKey() == null) {
6669 throw new IOException("Cannot merge two regions with null start key");
6670 }
6671
6672 } else if ((srcB.getRegionInfo().getStartKey() == null) ||
6673 (Bytes.compareTo(srcA.getRegionInfo().getStartKey(),
6674 srcB.getRegionInfo().getStartKey()) > 0)) {
6675 a = srcB;
6676 b = srcA;
6677 }
6678
6679 if (!(Bytes.compareTo(a.getRegionInfo().getEndKey(),
6680 b.getRegionInfo().getStartKey()) == 0)) {
6681 throw new IOException("Cannot merge non-adjacent regions");
6682 }
6683 return merge(a, b);
6684 }
6685
6686
6687
6688
6689
6690
6691
6692
6693
6694 public static HRegion merge(final HRegion a, final HRegion b) throws IOException {
6695 if (!a.getRegionInfo().getTable().equals(b.getRegionInfo().getTable())) {
6696 throw new IOException("Regions do not belong to the same table");
6697 }
6698
6699 FileSystem fs = a.getRegionFileSystem().getFileSystem();
6700
6701 a.flush(true);
6702 b.flush(true);
6703
6704
6705 a.compact(true);
6706 if (LOG.isDebugEnabled()) {
6707 LOG.debug("Files for region: " + a);
6708 a.getRegionFileSystem().logFileSystemState(LOG);
6709 }
6710 b.compact(true);
6711 if (LOG.isDebugEnabled()) {
6712 LOG.debug("Files for region: " + b);
6713 b.getRegionFileSystem().logFileSystemState(LOG);
6714 }
6715
6716 RegionMergeTransactionImpl rmt = new RegionMergeTransactionImpl(a, b, true);
6717 if (!rmt.prepare(null)) {
6718 throw new IOException("Unable to merge regions " + a + " and " + b);
6719 }
6720 HRegionInfo mergedRegionInfo = rmt.getMergedRegionInfo();
6721 LOG.info("starting merge of regions: " + a + " and " + b
6722 + " into new region " + mergedRegionInfo.getRegionNameAsString()
6723 + " with start key <"
6724 + Bytes.toStringBinary(mergedRegionInfo.getStartKey())
6725 + "> and end key <"
6726 + Bytes.toStringBinary(mergedRegionInfo.getEndKey()) + ">");
6727 HRegion dstRegion;
6728 try {
6729 dstRegion = (HRegion)rmt.execute(null, null);
6730 } catch (IOException ioe) {
6731 rmt.rollback(null, null);
6732 throw new IOException("Failed merging region " + a + " and " + b
6733 + ", and successfully rolled back");
6734 }
6735 dstRegion.compact(true);
6736
6737 if (LOG.isDebugEnabled()) {
6738 LOG.debug("Files for new region");
6739 dstRegion.getRegionFileSystem().logFileSystemState(LOG);
6740 }
6741
6742 if (dstRegion.getRegionFileSystem().hasReferences(dstRegion.getTableDesc())) {
6743 throw new IOException("Merged region " + dstRegion
6744 + " still has references after the compaction, is compaction canceled?");
6745 }
6746
6747
6748 HFileArchiver.archiveRegion(a.getBaseConf(), fs, a.getRegionInfo());
6749
6750 HFileArchiver.archiveRegion(b.getBaseConf(), fs, b.getRegionInfo());
6751
6752 LOG.info("merge completed. New region is " + dstRegion);
6753 return dstRegion;
6754 }
6755
6756 @Override
6757 public Result get(final Get get) throws IOException {
6758 checkRow(get.getRow(), "Get");
6759
6760 if (get.hasFamilies()) {
6761 for (byte [] family: get.familySet()) {
6762 checkFamily(family);
6763 }
6764 } else {
6765 for (byte[] family: this.htableDescriptor.getFamiliesKeys()) {
6766 get.addFamily(family);
6767 }
6768 }
6769 List<Cell> results = get(get, true);
6770 boolean stale = this.getRegionInfo().getReplicaId() != 0;
6771 return Result.create(results, get.isCheckExistenceOnly() ? !results.isEmpty() : null, stale);
6772 }
6773
6774 @Override
6775 public List<Cell> get(Get get, boolean withCoprocessor) throws IOException {
6776
6777 List<Cell> results = new ArrayList<Cell>();
6778
6779
6780 if (withCoprocessor && (coprocessorHost != null)) {
6781 if (coprocessorHost.preGet(get, results)) {
6782 return results;
6783 }
6784 }
6785
6786 Scan scan = new Scan(get);
6787
6788 RegionScanner scanner = null;
6789 try {
6790 scanner = getScanner(scan);
6791 scanner.next(results);
6792 } finally {
6793 if (scanner != null)
6794 scanner.close();
6795 }
6796
6797
6798 if (withCoprocessor && (coprocessorHost != null)) {
6799 coprocessorHost.postGet(get, results);
6800 }
6801
6802
6803 if (this.metricsRegion != null) {
6804 long totalSize = 0L;
6805 for (Cell cell : results) {
6806 totalSize += CellUtil.estimatedSerializedSizeOf(cell);
6807 }
6808 this.metricsRegion.updateGet(totalSize);
6809 }
6810
6811 return results;
6812 }
6813
6814 @Override
6815 public void mutateRow(RowMutations rm) throws IOException {
6816
6817 mutateRowsWithLocks(rm.getMutations(), Collections.singleton(rm.getRow()));
6818 }
6819
6820
6821
6822
6823
6824 public void mutateRowsWithLocks(Collection<Mutation> mutations,
6825 Collection<byte[]> rowsToLock) throws IOException {
6826 mutateRowsWithLocks(mutations, rowsToLock, HConstants.NO_NONCE, HConstants.NO_NONCE);
6827 }
6828
6829
6830
6831
6832
6833
6834
6835
6836
6837
6838
6839
6840
6841 @Override
6842 public void mutateRowsWithLocks(Collection<Mutation> mutations,
6843 Collection<byte[]> rowsToLock, long nonceGroup, long nonce) throws IOException {
6844 MultiRowMutationProcessor proc = new MultiRowMutationProcessor(mutations, rowsToLock);
6845 processRowsWithLocks(proc, -1, nonceGroup, nonce);
6846 }
6847
6848
6849
6850
6851 public ClientProtos.RegionLoadStats getRegionStats() {
6852 if (!regionStatsEnabled) {
6853 return null;
6854 }
6855 ClientProtos.RegionLoadStats.Builder stats = ClientProtos.RegionLoadStats.newBuilder();
6856 stats.setMemstoreLoad((int) (Math.min(100, (this.memstoreSize.get() * 100) / this
6857 .memstoreFlushSize)));
6858 stats.setHeapOccupancy((int)rsServices.getHeapMemoryManager().getHeapOccupancyPercent()*100);
6859 stats.setCompactionPressure((int)rsServices.getCompactionPressure()*100 > 100 ? 100 :
6860 (int)rsServices.getCompactionPressure()*100);
6861 return stats.build();
6862 }
6863
6864 @Override
6865 public void processRowsWithLocks(RowProcessor<?,?> processor) throws IOException {
6866 processRowsWithLocks(processor, rowProcessorTimeout, HConstants.NO_NONCE,
6867 HConstants.NO_NONCE);
6868 }
6869
6870 @Override
6871 public void processRowsWithLocks(RowProcessor<?,?> processor, long nonceGroup, long nonce)
6872 throws IOException {
6873 processRowsWithLocks(processor, rowProcessorTimeout, nonceGroup, nonce);
6874 }
6875
6876 @Override
6877 public void processRowsWithLocks(RowProcessor<?,?> processor, long timeout,
6878 long nonceGroup, long nonce) throws IOException {
6879
6880 for (byte[] row : processor.getRowsToLock()) {
6881 checkRow(row, "processRowsWithLocks");
6882 }
6883 if (!processor.readOnly()) {
6884 checkReadOnly();
6885 }
6886 checkResources();
6887
6888 startRegionOperation();
6889 WALEdit walEdit = new WALEdit();
6890
6891
6892 try {
6893 processor.preProcess(this, walEdit);
6894 } catch (IOException e) {
6895 closeRegionOperation();
6896 throw e;
6897 }
6898
6899 if (processor.readOnly()) {
6900 try {
6901 long now = EnvironmentEdgeManager.currentTime();
6902 doProcessRowWithTimeout(
6903 processor, now, this, null, null, timeout);
6904 processor.postProcess(this, walEdit, true);
6905 } finally {
6906 closeRegionOperation();
6907 }
6908 return;
6909 }
6910
6911 MultiVersionConcurrencyControl.WriteEntry writeEntry = null;
6912 boolean locked;
6913 boolean walSyncSuccessful = false;
6914 List<RowLock> acquiredRowLocks;
6915 long addedSize = 0;
6916 List<Mutation> mutations = new ArrayList<Mutation>();
6917 Collection<byte[]> rowsToLock = processor.getRowsToLock();
6918 long mvccNum = 0;
6919 WALKey walKey = null;
6920 try {
6921
6922 acquiredRowLocks = new ArrayList<RowLock>(rowsToLock.size());
6923 for (byte[] row : rowsToLock) {
6924
6925
6926 acquiredRowLocks.add(getRowLock(row));
6927 }
6928
6929 lock(this.updatesLock.readLock(), acquiredRowLocks.size() == 0 ? 1 : acquiredRowLocks.size());
6930 locked = true;
6931
6932 long now = EnvironmentEdgeManager.currentTime();
6933 try {
6934
6935
6936 doProcessRowWithTimeout(
6937 processor, now, this, mutations, walEdit, timeout);
6938
6939 if (!mutations.isEmpty()) {
6940
6941
6942 processor.preBatchMutate(this, walEdit);
6943
6944 long txid = 0;
6945
6946 if (!walEdit.isEmpty()) {
6947
6948 walKey = new HLogKey(this.getRegionInfo().getEncodedNameAsBytes(),
6949 this.htableDescriptor.getTableName(), WALKey.NO_SEQUENCE_ID, now,
6950 processor.getClusterIds(), nonceGroup, nonce, mvcc);
6951 txid = this.wal.append(this.htableDescriptor, this.getRegionInfo(),
6952 walKey, walEdit, false);
6953 }
6954 if(walKey == null){
6955
6956
6957 walKey = this.appendEmptyEdit(this.wal);
6958 }
6959
6960
6961 writeEntry = walKey.getWriteEntry();
6962 mvccNum = walKey.getSequenceId();
6963
6964
6965
6966
6967 for (Mutation m : mutations) {
6968
6969 rewriteCellTags(m.getFamilyCellMap(), m);
6970
6971 for (CellScanner cellScanner = m.cellScanner(); cellScanner.advance();) {
6972 Cell cell = cellScanner.current();
6973 CellUtil.setSequenceId(cell, mvccNum);
6974 Store store = getStore(cell);
6975 if (store == null) {
6976 checkFamily(CellUtil.cloneFamily(cell));
6977
6978 }
6979 addedSize += store.add(cell);
6980 }
6981 }
6982
6983
6984 if (locked) {
6985 this.updatesLock.readLock().unlock();
6986 locked = false;
6987 }
6988
6989
6990 releaseRowLocks(acquiredRowLocks);
6991
6992
6993 if (txid != 0) {
6994 syncOrDefer(txid, getEffectiveDurability(processor.useDurability()));
6995 }
6996 walSyncSuccessful = true;
6997
6998 processor.postBatchMutate(this);
6999 }
7000 } finally {
7001
7002
7003
7004 if (!mutations.isEmpty() && !walSyncSuccessful) {
7005 LOG.warn("Wal sync failed. Roll back " + mutations.size() +
7006 " memstore keyvalues for row(s):" + StringUtils.byteToHexString(
7007 processor.getRowsToLock().iterator().next()) + "...");
7008 for (Mutation m : mutations) {
7009 for (CellScanner cellScanner = m.cellScanner(); cellScanner.advance();) {
7010 Cell cell = cellScanner.current();
7011 getStore(cell).rollback(cell);
7012 }
7013 }
7014 if (writeEntry != null) {
7015 mvcc.complete(writeEntry);
7016 writeEntry = null;
7017 }
7018 }
7019
7020 if (writeEntry != null) {
7021 mvcc.completeAndWait(writeEntry);
7022 }
7023 if (locked) {
7024 this.updatesLock.readLock().unlock();
7025 }
7026
7027 releaseRowLocks(acquiredRowLocks);
7028 }
7029
7030
7031 processor.postProcess(this, walEdit, walSyncSuccessful);
7032
7033 } finally {
7034 closeRegionOperation();
7035 if (!mutations.isEmpty() &&
7036 isFlushSize(this.addAndGetGlobalMemstoreSize(addedSize))) {
7037 requestFlush();
7038 }
7039 }
7040 }
7041
7042 private void doProcessRowWithTimeout(final RowProcessor<?,?> processor,
7043 final long now,
7044 final HRegion region,
7045 final List<Mutation> mutations,
7046 final WALEdit walEdit,
7047 final long timeout) throws IOException {
7048
7049 if (timeout < 0) {
7050 try {
7051 processor.process(now, region, mutations, walEdit);
7052 } catch (IOException e) {
7053 LOG.warn("RowProcessor:" + processor.getClass().getName() +
7054 " throws Exception on row(s):" +
7055 Bytes.toStringBinary(
7056 processor.getRowsToLock().iterator().next()) + "...", e);
7057 throw e;
7058 }
7059 return;
7060 }
7061
7062
7063 FutureTask<Void> task =
7064 new FutureTask<Void>(new Callable<Void>() {
7065 @Override
7066 public Void call() throws IOException {
7067 try {
7068 processor.process(now, region, mutations, walEdit);
7069 return null;
7070 } catch (IOException e) {
7071 LOG.warn("RowProcessor:" + processor.getClass().getName() +
7072 " throws Exception on row(s):" +
7073 Bytes.toStringBinary(
7074 processor.getRowsToLock().iterator().next()) + "...", e);
7075 throw e;
7076 }
7077 }
7078 });
7079 rowProcessorExecutor.execute(task);
7080 try {
7081 task.get(timeout, TimeUnit.MILLISECONDS);
7082 } catch (TimeoutException te) {
7083 LOG.error("RowProcessor timeout:" + timeout + " ms on row(s):" +
7084 Bytes.toStringBinary(processor.getRowsToLock().iterator().next()) +
7085 "...");
7086 throw new IOException(te);
7087 } catch (Exception e) {
7088 throw new IOException(e);
7089 }
7090 }
7091
7092
7093
7094
7095
7096
7097 private static List<Tag> carryForwardTags(final Cell cell, final List<Tag> tags) {
7098 if (cell.getTagsLength() <= 0) return tags;
7099 List<Tag> newTags = tags == null? new ArrayList<Tag>():
7100 Iterator<Tag> i =
7101 CellUtil.tagsIterator(cell.getTagsArray(), cell.getTagsOffset(), cell.getTagsLength());
7102 while (i.hasNext()) newTags.add(i.next());
7103 return newTags;
7104 }
7105
7106
7107
7108
7109
7110
7111
7112
7113
7114
7115 private List<Cell> doGet(final Store store, final byte [] row,
7116 final Map.Entry<byte[], List<Cell>> family, final TimeRange tr)
7117 throws IOException {
7118
7119
7120
7121
7122 Collections.sort(family.getValue(), store.getComparator());
7123
7124 Get get = new Get(row);
7125 for (Cell cell : family.getValue()) {
7126 get.addColumn(family.getKey(), CellUtil.cloneQualifier(cell));
7127 }
7128 if (tr != null) get.setTimeRange(tr.getMin(), tr.getMax());
7129 return get(get, false);
7130 }
7131
7132 public Result append(Append append) throws IOException {
7133 return append(append, HConstants.NO_NONCE, HConstants.NO_NONCE);
7134 }
7135
7136
7137
7138
7139
7140 @Override
7141 public Result append(Append mutate, long nonceGroup, long nonce) throws IOException {
7142 Operation op = Operation.APPEND;
7143 byte[] row = mutate.getRow();
7144 checkRow(row, op.toString());
7145 boolean flush = false;
7146 Durability durability = getEffectiveDurability(mutate.getDurability());
7147 boolean writeToWAL = durability != Durability.SKIP_WAL;
7148 WALEdit walEdits = null;
7149 List<Cell> allKVs = new ArrayList<Cell>(mutate.size());
7150 Map<Store, List<Cell>> tempMemstore = new HashMap<Store, List<Cell>>();
7151 long size = 0;
7152 long txid = 0;
7153 checkReadOnly();
7154 checkResources();
7155
7156 startRegionOperation(op);
7157 this.writeRequestsCount.increment();
7158 RowLock rowLock = null;
7159 WALKey walKey = null;
7160 boolean doRollBackMemstore = false;
7161 try {
7162 rowLock = getRowLock(row);
7163 assert rowLock != null;
7164 try {
7165 lock(this.updatesLock.readLock());
7166 try {
7167
7168
7169 mvcc.await();
7170 if (this.coprocessorHost != null) {
7171 Result r = this.coprocessorHost.preAppendAfterRowLock(mutate);
7172 if (r!= null) {
7173 return r;
7174 }
7175 }
7176 long now = EnvironmentEdgeManager.currentTime();
7177
7178 for (Map.Entry<byte[], List<Cell>> family : mutate.getFamilyCellMap().entrySet()) {
7179 Store store = stores.get(family.getKey());
7180 List<Cell> kvs = new ArrayList<Cell>(family.getValue().size());
7181
7182 List<Cell> results = doGet(store, row, family, null);
7183
7184
7185
7186
7187
7188
7189
7190 int idx = 0;
7191 for (Cell cell : family.getValue()) {
7192 Cell newCell;
7193 Cell oldCell = null;
7194 if (idx < results.size()
7195 && CellUtil.matchingQualifier(results.get(idx), cell)) {
7196 oldCell = results.get(idx);
7197 long ts = Math.max(now, oldCell.getTimestamp());
7198
7199
7200
7201 List<Tag> tags = Tag.carryForwardTags(null, oldCell);
7202 tags = Tag.carryForwardTags(tags, cell);
7203 tags = carryForwardTTLTag(tags, mutate);
7204
7205 newCell = getNewCell(row, ts, cell, oldCell, Tag.fromList(tags));
7206
7207 idx++;
7208 } else {
7209
7210 CellUtil.updateLatestStamp(cell, now);
7211
7212
7213 newCell = getNewCell(mutate, cell);
7214 }
7215
7216
7217 if (coprocessorHost != null) {
7218 newCell = coprocessorHost.postMutationBeforeWAL(RegionObserver.MutationType.APPEND,
7219 mutate, oldCell, newCell);
7220 }
7221 kvs.add(newCell);
7222
7223
7224 if (writeToWAL) {
7225 if (walEdits == null) {
7226 walEdits = new WALEdit();
7227 }
7228 walEdits.add(newCell);
7229 }
7230 }
7231
7232
7233 tempMemstore.put(store, kvs);
7234 }
7235
7236
7237 if (walEdits != null && !walEdits.isEmpty()) {
7238 if (writeToWAL) {
7239
7240
7241
7242
7243 walKey = new HLogKey(
7244 getRegionInfo().getEncodedNameAsBytes(),
7245 this.htableDescriptor.getTableName(),
7246 WALKey.NO_SEQUENCE_ID,
7247 nonceGroup,
7248 nonce,
7249 mvcc);
7250 txid =
7251 this.wal.append(this.htableDescriptor, getRegionInfo(), walKey, walEdits, true);
7252 } else {
7253 recordMutationWithoutWal(mutate.getFamilyCellMap());
7254 }
7255 }
7256 if (walKey == null) {
7257
7258 walKey = this.appendEmptyEdit(this.wal);
7259 }
7260
7261
7262 walKey.getWriteEntry();
7263
7264
7265 if (!tempMemstore.isEmpty()) {
7266 for (Map.Entry<Store, List<Cell>> entry : tempMemstore.entrySet()) {
7267 Store store = entry.getKey();
7268 if (store.getFamily().getMaxVersions() == 1) {
7269
7270
7271 size += store.upsert(entry.getValue(), getSmallestReadPoint());
7272 } else {
7273
7274 for (Cell cell: entry.getValue()) {
7275
7276
7277 CellUtil.setSequenceId(cell, walKey.getWriteEntry().getWriteNumber());
7278 size += store.add(cell);
7279 doRollBackMemstore = true;
7280 }
7281 }
7282
7283
7284 allKVs.addAll(entry.getValue());
7285 }
7286
7287 size = this.addAndGetGlobalMemstoreSize(size);
7288 flush = isFlushSize(size);
7289 }
7290 } finally {
7291 this.updatesLock.readLock().unlock();
7292 }
7293
7294 } finally {
7295 rowLock.release();
7296 rowLock = null;
7297 }
7298
7299 if(txid != 0){
7300 syncOrDefer(txid, durability);
7301 }
7302 doRollBackMemstore = false;
7303 } finally {
7304 if (rowLock != null) {
7305 rowLock.release();
7306 }
7307
7308 WriteEntry we = walKey != null? walKey.getWriteEntry(): null;
7309 if (doRollBackMemstore) {
7310 rollbackMemstore(allKVs);
7311 if (we != null) mvcc.complete(we);
7312 } else if (we != null) {
7313 mvcc.completeAndWait(we);
7314 }
7315
7316 closeRegionOperation(op);
7317 }
7318
7319 if (this.metricsRegion != null) {
7320 this.metricsRegion.updateAppend();
7321 }
7322
7323 if (flush) {
7324
7325 requestFlush();
7326 }
7327
7328 return mutate.isReturnResults() ? Result.create(allKVs) : null;
7329 }
7330
7331 private static Cell getNewCell(final byte [] row, final long ts, final Cell cell,
7332 final Cell oldCell, final byte [] tagBytes) {
7333
7334 Cell newCell = new KeyValue(row.length, cell.getFamilyLength(),
7335 cell.getQualifierLength(), ts, KeyValue.Type.Put,
7336 oldCell.getValueLength() + cell.getValueLength(),
7337 tagBytes == null? 0: tagBytes.length);
7338
7339 System.arraycopy(cell.getRowArray(), cell.getRowOffset(),
7340 newCell.getRowArray(), newCell.getRowOffset(), cell.getRowLength());
7341 System.arraycopy(cell.getFamilyArray(), cell.getFamilyOffset(),
7342 newCell.getFamilyArray(), newCell.getFamilyOffset(),
7343 cell.getFamilyLength());
7344 System.arraycopy(cell.getQualifierArray(), cell.getQualifierOffset(),
7345 newCell.getQualifierArray(), newCell.getQualifierOffset(),
7346 cell.getQualifierLength());
7347
7348 System.arraycopy(oldCell.getValueArray(), oldCell.getValueOffset(),
7349 newCell.getValueArray(), newCell.getValueOffset(),
7350 oldCell.getValueLength());
7351 System.arraycopy(cell.getValueArray(), cell.getValueOffset(),
7352 newCell.getValueArray(),
7353 newCell.getValueOffset() + oldCell.getValueLength(),
7354 cell.getValueLength());
7355
7356 if (tagBytes != null) {
7357 System.arraycopy(tagBytes, 0, newCell.getTagsArray(), newCell.getTagsOffset(),
7358 tagBytes.length);
7359 }
7360 return newCell;
7361 }
7362
7363 private static Cell getNewCell(final Mutation mutate, final Cell cell) {
7364 Cell newCell = null;
7365 if (mutate.getTTL() != Long.MAX_VALUE) {
7366
7367 newCell = new KeyValue(cell.getRowArray(), cell.getRowOffset(),
7368 cell.getRowLength(),
7369 cell.getFamilyArray(), cell.getFamilyOffset(),
7370 cell.getFamilyLength(),
7371 cell.getQualifierArray(), cell.getQualifierOffset(),
7372 cell.getQualifierLength(),
7373 cell.getTimestamp(), KeyValue.Type.codeToType(cell.getTypeByte()),
7374 cell.getValueArray(), cell.getValueOffset(), cell.getValueLength(),
7375 carryForwardTTLTag(mutate));
7376 } else {
7377 newCell = cell;
7378 }
7379 return newCell;
7380 }
7381
7382 public Result increment(Increment increment) throws IOException {
7383 return increment(increment, HConstants.NO_NONCE, HConstants.NO_NONCE);
7384 }
7385
7386
7387
7388
7389
7390
7391
7392
7393
7394 @Override
7395 public Result increment(Increment mutation, long nonceGroup, long nonce)
7396 throws IOException {
7397 Operation op = Operation.INCREMENT;
7398 checkReadOnly();
7399 checkResources();
7400 checkRow(mutation.getRow(), op.toString());
7401 checkFamilies(mutation.getFamilyCellMap().keySet());
7402 startRegionOperation(op);
7403 this.writeRequestsCount.increment();
7404 try {
7405
7406
7407
7408
7409
7410
7411
7412
7413
7414
7415
7416
7417
7418
7419
7420
7421 return doIncrement(mutation, nonceGroup, nonce);
7422 } finally {
7423 if (this.metricsRegion != null) this.metricsRegion.updateIncrement();
7424 closeRegionOperation(op);
7425 }
7426 }
7427
7428 private Result doIncrement(Increment increment, long nonceGroup, long nonce) throws IOException {
7429 RowLock rowLock = null;
7430 WALKey walKey = null;
7431 boolean doRollBackMemstore = false;
7432 long accumulatedResultSize = 0;
7433 List<Cell> allKVs = new ArrayList<Cell>(increment.size());
7434 List<Cell> memstoreCells = new ArrayList<Cell>();
7435 Durability effectiveDurability = getEffectiveDurability(increment.getDurability());
7436 try {
7437 rowLock = getRowLock(increment.getRow());
7438 long txid = 0;
7439 try {
7440 lock(this.updatesLock.readLock());
7441 try {
7442
7443
7444 this.mvcc.await();
7445 if (this.coprocessorHost != null) {
7446 Result r = this.coprocessorHost.preIncrementAfterRowLock(increment);
7447 if (r != null) return r;
7448 }
7449 long now = EnvironmentEdgeManager.currentTime();
7450 final boolean writeToWAL = effectiveDurability != Durability.SKIP_WAL;
7451 WALEdit walEdits = null;
7452
7453
7454 Map<Store, List<Cell>> forMemStore = new HashMap<Store, List<Cell>>();
7455 for (Map.Entry<byte [], List<Cell>> entry: increment.getFamilyCellMap().entrySet()) {
7456 byte [] columnFamilyName = entry.getKey();
7457 List<Cell> increments = entry.getValue();
7458 Store store = this.stores.get(columnFamilyName);
7459
7460
7461 List<Cell> results = applyIncrementsToColumnFamily(increment, columnFamilyName,
7462 sort(increments, store.getComparator()), now,
7463 MultiVersionConcurrencyControl.NO_WRITE_NUMBER, allKVs, null);
7464 if (!results.isEmpty()) {
7465 forMemStore.put(store, results);
7466
7467 if (writeToWAL) {
7468 if (walEdits == null) walEdits = new WALEdit();
7469 walEdits.getCells().addAll(results);
7470 }
7471 }
7472 }
7473
7474 if (walEdits != null && !walEdits.isEmpty()) {
7475
7476
7477
7478 walKey = new HLogKey(this.getRegionInfo().getEncodedNameAsBytes(),
7479 this.htableDescriptor.getTableName(), WALKey.NO_SEQUENCE_ID, nonceGroup, nonce,
7480 getMVCC());
7481 txid =
7482 this.wal.append(this.htableDescriptor, this.getRegionInfo(), walKey, walEdits, true);
7483 } else {
7484
7485 walKey = this.appendEmptyEdit(this.wal);
7486 }
7487
7488 walKey.getWriteEntry();
7489
7490
7491 for (Map.Entry<Store, List<Cell>> entry: forMemStore.entrySet()) {
7492 Store store = entry.getKey();
7493 List<Cell> results = entry.getValue();
7494 if (store.getFamily().getMaxVersions() == 1) {
7495
7496 accumulatedResultSize += store.upsert(results, getSmallestReadPoint());
7497
7498 } else {
7499
7500 for (Cell cell: results) {
7501
7502 CellUtil.setSequenceId(cell, walKey.getWriteEntry().getWriteNumber());
7503 accumulatedResultSize += store.add(cell);
7504 doRollBackMemstore = true;
7505 }
7506 }
7507 }
7508 } finally {
7509 this.updatesLock.readLock().unlock();
7510 }
7511 } finally {
7512 rowLock.release();
7513 rowLock = null;
7514 }
7515
7516 if(txid != 0) {
7517 syncOrDefer(txid, durability);
7518 }
7519 doRollBackMemstore = false;
7520 } finally {
7521 if (rowLock != null) {
7522 rowLock.release();
7523 }
7524
7525 if (doRollBackMemstore) {
7526 rollbackMemstore(memstoreCells);
7527 if (walKey != null) mvcc.complete(walKey.getWriteEntry());
7528 } else {
7529 if (walKey != null) mvcc.completeAndWait(walKey.getWriteEntry());
7530 }
7531 }
7532
7533
7534 if (isFlushSize(this.addAndGetGlobalMemstoreSize(accumulatedResultSize))) requestFlush();
7535 return increment.isReturnResults() ? Result.create(allKVs) : null;
7536 }
7537
7538
7539
7540
7541 private static List<Cell> sort(List<Cell> cells, final Comparator<Cell> comparator) {
7542 Collections.sort(cells, comparator);
7543 return cells;
7544 }
7545
7546
7547
7548
7549
7550
7551
7552
7553
7554
7555
7556
7557 private List<Cell> applyIncrementsToColumnFamily(Increment increment, byte[] columnFamilyName,
7558 List<Cell> sortedIncrements, long now, long mvccNum, List<Cell> allKVs,
7559 final IsolationLevel isolation)
7560 throws IOException {
7561 List<Cell> results = new ArrayList<Cell>(sortedIncrements.size());
7562 byte [] row = increment.getRow();
7563
7564 List<Cell> currentValues =
7565 getIncrementCurrentValue(increment, columnFamilyName, sortedIncrements, isolation);
7566
7567
7568 int idx = 0;
7569 for (int i = 0; i < sortedIncrements.size(); i++) {
7570 Cell inc = sortedIncrements.get(i);
7571 long incrementAmount = getLongValue(inc);
7572
7573 boolean writeBack = (incrementAmount != 0);
7574
7575 List<Tag> tags = Tag.carryForwardTags(inc);
7576
7577 Cell currentValue = null;
7578 long ts = now;
7579 if (idx < currentValues.size() && CellUtil.matchingQualifier(currentValues.get(idx), inc)) {
7580 currentValue = currentValues.get(idx);
7581 ts = Math.max(now, currentValue.getTimestamp());
7582 incrementAmount += getLongValue(currentValue);
7583
7584 tags = Tag.carryForwardTags(tags, currentValue);
7585 if (i < (sortedIncrements.size() - 1) &&
7586 !CellUtil.matchingQualifier(inc, sortedIncrements.get(i + 1))) idx++;
7587 }
7588
7589
7590 byte [] qualifier = CellUtil.cloneQualifier(inc);
7591 byte [] incrementAmountInBytes = Bytes.toBytes(incrementAmount);
7592 tags = carryForwardTTLTag(tags, increment);
7593
7594 Cell newValue = new KeyValue(row, 0, row.length,
7595 columnFamilyName, 0, columnFamilyName.length,
7596 qualifier, 0, qualifier.length,
7597 ts, KeyValue.Type.Put,
7598 incrementAmountInBytes, 0, incrementAmountInBytes.length,
7599 tags);
7600
7601
7602
7603 if (mvccNum != MultiVersionConcurrencyControl.NO_WRITE_NUMBER) {
7604 CellUtil.setSequenceId(newValue, mvccNum);
7605 }
7606
7607
7608 if (coprocessorHost != null) {
7609 newValue = coprocessorHost.postMutationBeforeWAL(
7610 RegionObserver.MutationType.INCREMENT, increment, currentValue, newValue);
7611 }
7612 allKVs.add(newValue);
7613 if (writeBack) {
7614 results.add(newValue);
7615 }
7616 }
7617 return results;
7618 }
7619
7620
7621
7622
7623
7624 private static long getLongValue(final Cell cell) throws DoNotRetryIOException {
7625 int len = cell.getValueLength();
7626 if (len != Bytes.SIZEOF_LONG) {
7627
7628 throw new DoNotRetryIOException("Field is not a long, it's " + len + " bytes wide");
7629 }
7630 return Bytes.toLong(cell.getValueArray(), cell.getValueOffset(), len);
7631 }
7632
7633
7634
7635
7636
7637
7638
7639
7640
7641
7642 private List<Cell> getIncrementCurrentValue(final Increment increment, byte [] columnFamily,
7643 final List<Cell> increments, final IsolationLevel isolation)
7644 throws IOException {
7645 Get get = new Get(increment.getRow());
7646 if (isolation != null) get.setIsolationLevel(isolation);
7647 for (Cell cell: increments) {
7648 get.addColumn(columnFamily, CellUtil.cloneQualifier(cell));
7649 }
7650 TimeRange tr = increment.getTimeRange();
7651 if (tr != null) {
7652 get.setTimeRange(tr.getMin(), tr.getMax());
7653 }
7654 return get(get, false);
7655 }
7656
7657 private static List<Tag> carryForwardTTLTag(final Mutation mutation) {
7658 return carryForwardTTLTag(null, mutation);
7659 }
7660
7661
7662
7663
7664 private static List<Tag> carryForwardTTLTag(final List<Tag> tagsOrNull,
7665 final Mutation mutation) {
7666 long ttl = mutation.getTTL();
7667 if (ttl == Long.MAX_VALUE) return tagsOrNull;
7668 List<Tag> tags = tagsOrNull;
7669
7670
7671
7672 if (tags == null) tags = new ArrayList<Tag>(1);
7673 tags.add(new Tag(TagType.TTL_TAG_TYPE, Bytes.toBytes(ttl)));
7674 return tags;
7675 }
7676
7677
7678
7679
7680
7681 private void checkFamily(final byte [] family)
7682 throws NoSuchColumnFamilyException {
7683 if (!this.htableDescriptor.hasFamily(family)) {
7684 throw new NoSuchColumnFamilyException("Column family " +
7685 Bytes.toString(family) + " does not exist in region " + this
7686 + " in table " + this.htableDescriptor);
7687 }
7688 }
7689
7690 public static final long FIXED_OVERHEAD = ClassSize.align(
7691 ClassSize.OBJECT +
7692 ClassSize.ARRAY +
7693 44 * ClassSize.REFERENCE + 3 * Bytes.SIZEOF_INT +
7694 (14 * Bytes.SIZEOF_LONG) +
7695 5 * Bytes.SIZEOF_BOOLEAN);
7696
7697
7698
7699
7700
7701
7702
7703
7704
7705
7706
7707 public static final long DEEP_OVERHEAD = FIXED_OVERHEAD +
7708 ClassSize.OBJECT +
7709 (2 * ClassSize.ATOMIC_BOOLEAN) +
7710 (3 * ClassSize.ATOMIC_LONG) +
7711 (2 * ClassSize.CONCURRENT_HASHMAP) +
7712 WriteState.HEAP_SIZE +
7713 ClassSize.CONCURRENT_SKIPLISTMAP + ClassSize.CONCURRENT_SKIPLISTMAP_ENTRY +
7714 (2 * ClassSize.REENTRANT_LOCK) +
7715 MultiVersionConcurrencyControl.FIXED_SIZE
7716 + ClassSize.TREEMAP
7717 + 2 * ClassSize.ATOMIC_INTEGER
7718 ;
7719
7720 @Override
7721 public long heapSize() {
7722 long heapSize = DEEP_OVERHEAD;
7723 for (Store store : this.stores.values()) {
7724 heapSize += store.heapSize();
7725 }
7726
7727 return heapSize;
7728 }
7729
7730
7731
7732
7733
7734 private static void printUsageAndExit(final String message) {
7735 if (message != null && message.length() > 0) System.out.println(message);
7736 System.out.println("Usage: HRegion CATALOG_TABLE_DIR [major_compact]");
7737 System.out.println("Options:");
7738 System.out.println(" major_compact Pass this option to major compact " +
7739 "passed region.");
7740 System.out.println("Default outputs scan of passed region.");
7741 System.exit(1);
7742 }
7743
7744 @Override
7745 public boolean registerService(Service instance) {
7746
7747
7748
7749 Descriptors.ServiceDescriptor serviceDesc = instance.getDescriptorForType();
7750 if (coprocessorServiceHandlers.containsKey(serviceDesc.getFullName())) {
7751 LOG.error("Coprocessor service " + serviceDesc.getFullName() +
7752 " already registered, rejecting request from " + instance
7753 );
7754 return false;
7755 }
7756
7757 coprocessorServiceHandlers.put(serviceDesc.getFullName(), instance);
7758 if (LOG.isDebugEnabled()) {
7759 LOG.debug("Registered coprocessor service: region=" +
7760 Bytes.toStringBinary(getRegionInfo().getRegionName()) +
7761 " service=" + serviceDesc.getFullName());
7762 }
7763 return true;
7764 }
7765
7766 @Override
7767 public Message execService(RpcController controller, CoprocessorServiceCall call)
7768 throws IOException {
7769 String serviceName = call.getServiceName();
7770 String methodName = call.getMethodName();
7771 if (!coprocessorServiceHandlers.containsKey(serviceName)) {
7772 throw new UnknownProtocolException(null,
7773 "No registered coprocessor service found for name "+serviceName+
7774 " in region "+Bytes.toStringBinary(getRegionInfo().getRegionName()));
7775 }
7776
7777 Service service = coprocessorServiceHandlers.get(serviceName);
7778 Descriptors.ServiceDescriptor serviceDesc = service.getDescriptorForType();
7779 Descriptors.MethodDescriptor methodDesc = serviceDesc.findMethodByName(methodName);
7780 if (methodDesc == null) {
7781 throw new UnknownProtocolException(service.getClass(),
7782 "Unknown method "+methodName+" called on service "+serviceName+
7783 " in region "+Bytes.toStringBinary(getRegionInfo().getRegionName()));
7784 }
7785
7786 Message.Builder builder = service.getRequestPrototype(methodDesc).newBuilderForType();
7787 ProtobufUtil.mergeFrom(builder, call.getRequest());
7788 Message request = builder.build();
7789
7790 if (coprocessorHost != null) {
7791 request = coprocessorHost.preEndpointInvocation(service, methodName, request);
7792 }
7793
7794 final Message.Builder responseBuilder =
7795 service.getResponsePrototype(methodDesc).newBuilderForType();
7796 service.callMethod(methodDesc, controller, request, new RpcCallback<Message>() {
7797 @Override
7798 public void run(Message message) {
7799 if (message != null) {
7800 responseBuilder.mergeFrom(message);
7801 }
7802 }
7803 });
7804
7805 if (coprocessorHost != null) {
7806 coprocessorHost.postEndpointInvocation(service, methodName, request, responseBuilder);
7807 }
7808
7809 IOException exception = ResponseConverter.getControllerException(controller);
7810 if (exception != null) {
7811 throw exception;
7812 }
7813
7814 return responseBuilder.build();
7815 }
7816
7817
7818
7819
7820
7821
7822 private static void processTable(final FileSystem fs, final Path p,
7823 final WALFactory walFactory, final Configuration c,
7824 final boolean majorCompact)
7825 throws IOException {
7826 HRegion region;
7827 FSTableDescriptors fst = new FSTableDescriptors(c);
7828
7829 if (FSUtils.getTableName(p).equals(TableName.META_TABLE_NAME)) {
7830 final WAL wal = walFactory.getMetaWAL(
7831 HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes());
7832 region = HRegion.newHRegion(p, wal, fs, c,
7833 HRegionInfo.FIRST_META_REGIONINFO, fst.get(TableName.META_TABLE_NAME), null);
7834 } else {
7835 throw new IOException("Not a known catalog table: " + p.toString());
7836 }
7837 try {
7838 region.mvcc.advanceTo(region.initialize(null));
7839 if (majorCompact) {
7840 region.compact(true);
7841 } else {
7842
7843 Scan scan = new Scan();
7844
7845 RegionScanner scanner = region.getScanner(scan);
7846 try {
7847 List<Cell> kvs = new ArrayList<Cell>();
7848 boolean done;
7849 do {
7850 kvs.clear();
7851 done = scanner.next(kvs);
7852 if (kvs.size() > 0) LOG.info(kvs);
7853 } while (done);
7854 } finally {
7855 scanner.close();
7856 }
7857 }
7858 } finally {
7859 region.close();
7860 }
7861 }
7862
7863 boolean shouldForceSplit() {
7864 return this.splitRequest;
7865 }
7866
7867 byte[] getExplicitSplitPoint() {
7868 return this.explicitSplitPoint;
7869 }
7870
7871 void forceSplit(byte[] sp) {
7872
7873
7874 this.splitRequest = true;
7875 if (sp != null) {
7876 this.explicitSplitPoint = sp;
7877 }
7878 }
7879
7880 void clearSplit() {
7881 this.splitRequest = false;
7882 this.explicitSplitPoint = null;
7883 }
7884
7885
7886
7887
7888 protected void prepareToSplit() {
7889
7890 }
7891
7892
7893
7894
7895
7896
7897
7898 public byte[] checkSplit() {
7899
7900 if (this.getRegionInfo().isMetaTable() ||
7901 TableName.NAMESPACE_TABLE_NAME.equals(this.getRegionInfo().getTable())) {
7902 if (shouldForceSplit()) {
7903 LOG.warn("Cannot split meta region in HBase 0.20 and above");
7904 }
7905 return null;
7906 }
7907
7908
7909 if (this.isRecovering()) {
7910 LOG.info("Cannot split region " + this.getRegionInfo().getEncodedName() + " in recovery.");
7911 return null;
7912 }
7913
7914 if (!splitPolicy.shouldSplit()) {
7915 return null;
7916 }
7917
7918 byte[] ret = splitPolicy.getSplitPoint();
7919
7920 if (ret != null) {
7921 try {
7922 checkRow(ret, "calculated split");
7923 } catch (IOException e) {
7924 LOG.error("Ignoring invalid split", e);
7925 return null;
7926 }
7927 }
7928 return ret;
7929 }
7930
7931
7932
7933
7934 public int getCompactPriority() {
7935 int count = Integer.MAX_VALUE;
7936 for (Store store : stores.values()) {
7937 count = Math.min(count, store.getCompactPriority());
7938 }
7939 return count;
7940 }
7941
7942
7943
7944 @Override
7945 public RegionCoprocessorHost getCoprocessorHost() {
7946 return coprocessorHost;
7947 }
7948
7949
7950 public void setCoprocessorHost(final RegionCoprocessorHost coprocessorHost) {
7951 this.coprocessorHost = coprocessorHost;
7952 }
7953
7954 @Override
7955 public void startRegionOperation() throws IOException {
7956 startRegionOperation(Operation.ANY);
7957 }
7958
7959 @Override
7960 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="SF_SWITCH_FALLTHROUGH",
7961 justification="Intentional")
7962 public void startRegionOperation(Operation op) throws IOException {
7963 switch (op) {
7964 case GET:
7965 case SCAN:
7966 checkReadsEnabled();
7967 case INCREMENT:
7968 case APPEND:
7969 case SPLIT_REGION:
7970 case MERGE_REGION:
7971 case PUT:
7972 case DELETE:
7973 case BATCH_MUTATE:
7974 case COMPACT_REGION:
7975
7976 if (isRecovering() && (this.disallowWritesInRecovering ||
7977 (op != Operation.PUT && op != Operation.DELETE && op != Operation.BATCH_MUTATE))) {
7978 throw new RegionInRecoveryException(getRegionInfo().getRegionNameAsString() +
7979 " is recovering; cannot take reads");
7980 }
7981 break;
7982 default:
7983 break;
7984 }
7985 if (op == Operation.MERGE_REGION || op == Operation.SPLIT_REGION
7986 || op == Operation.COMPACT_REGION) {
7987
7988
7989 return;
7990 }
7991 if (this.closing.get()) {
7992 throw new NotServingRegionException(getRegionInfo().getRegionNameAsString() + " is closing");
7993 }
7994 lock(lock.readLock());
7995 if (this.closed.get()) {
7996 lock.readLock().unlock();
7997 throw new NotServingRegionException(getRegionInfo().getRegionNameAsString() + " is closed");
7998 }
7999 try {
8000 if (coprocessorHost != null) {
8001 coprocessorHost.postStartRegionOperation(op);
8002 }
8003 } catch (Exception e) {
8004 lock.readLock().unlock();
8005 throw new IOException(e);
8006 }
8007 }
8008
8009 @Override
8010 public void closeRegionOperation() throws IOException {
8011 closeRegionOperation(Operation.ANY);
8012 }
8013
8014
8015
8016
8017
8018
8019 public void closeRegionOperation(Operation operation) throws IOException {
8020 lock.readLock().unlock();
8021 if (coprocessorHost != null) {
8022 coprocessorHost.postCloseRegionOperation(operation);
8023 }
8024 }
8025
8026
8027
8028
8029
8030
8031
8032
8033
8034
8035 private void startBulkRegionOperation(boolean writeLockNeeded)
8036 throws NotServingRegionException, RegionTooBusyException, InterruptedIOException {
8037 if (this.closing.get()) {
8038 throw new NotServingRegionException(getRegionInfo().getRegionNameAsString() + " is closing");
8039 }
8040 if (writeLockNeeded) lock(lock.writeLock());
8041 else lock(lock.readLock());
8042 if (this.closed.get()) {
8043 if (writeLockNeeded) lock.writeLock().unlock();
8044 else lock.readLock().unlock();
8045 throw new NotServingRegionException(getRegionInfo().getRegionNameAsString() + " is closed");
8046 }
8047 }
8048
8049
8050
8051
8052
8053 private void closeBulkRegionOperation(){
8054 if (lock.writeLock().isHeldByCurrentThread()) lock.writeLock().unlock();
8055 else lock.readLock().unlock();
8056 }
8057
8058
8059
8060
8061
8062 private void recordMutationWithoutWal(final Map<byte [], List<Cell>> familyMap) {
8063 numMutationsWithoutWAL.increment();
8064 if (numMutationsWithoutWAL.get() <= 1) {
8065 LOG.info("writing data to region " + this +
8066 " with WAL disabled. Data may be lost in the event of a crash.");
8067 }
8068
8069 long mutationSize = 0;
8070 for (List<Cell> cells: familyMap.values()) {
8071 assert cells instanceof RandomAccess;
8072 int listSize = cells.size();
8073 for (int i=0; i < listSize; i++) {
8074 Cell cell = cells.get(i);
8075
8076 mutationSize += KeyValueUtil.keyLength(cell) + cell.getValueLength();
8077 }
8078 }
8079
8080 dataInMemoryWithoutWAL.add(mutationSize);
8081 }
8082
8083 private void lock(final Lock lock)
8084 throws RegionTooBusyException, InterruptedIOException {
8085 lock(lock, 1);
8086 }
8087
8088
8089
8090
8091
8092
8093 private void lock(final Lock lock, final int multiplier)
8094 throws RegionTooBusyException, InterruptedIOException {
8095 try {
8096 final long waitTime = Math.min(maxBusyWaitDuration,
8097 busyWaitDuration * Math.min(multiplier, maxBusyWaitMultiplier));
8098 if (!lock.tryLock(waitTime, TimeUnit.MILLISECONDS)) {
8099 throw new RegionTooBusyException(
8100 "failed to get a lock in " + waitTime + " ms. " +
8101 "regionName=" + (this.getRegionInfo() == null ? "unknown" :
8102 this.getRegionInfo().getRegionNameAsString()) +
8103 ", server=" + (this.getRegionServerServices() == null ? "unknown" :
8104 this.getRegionServerServices().getServerName()));
8105 }
8106 } catch (InterruptedException ie) {
8107 LOG.info("Interrupted while waiting for a lock");
8108 InterruptedIOException iie = new InterruptedIOException();
8109 iie.initCause(ie);
8110 throw iie;
8111 }
8112 }
8113
8114
8115
8116
8117
8118
8119
8120 private void syncOrDefer(long txid, Durability durability) throws IOException {
8121 if (this.getRegionInfo().isMetaRegion()) {
8122 this.wal.sync(txid);
8123 } else {
8124 switch(durability) {
8125 case USE_DEFAULT:
8126
8127 if (shouldSyncWAL()) {
8128 this.wal.sync(txid);
8129 }
8130 break;
8131 case SKIP_WAL:
8132
8133 break;
8134 case ASYNC_WAL:
8135
8136 break;
8137 case SYNC_WAL:
8138 case FSYNC_WAL:
8139
8140 this.wal.sync(txid);
8141 break;
8142 default:
8143 throw new RuntimeException("Unknown durability " + durability);
8144 }
8145 }
8146 }
8147
8148
8149
8150
8151 private boolean shouldSyncWAL() {
8152 return durability.ordinal() > Durability.ASYNC_WAL.ordinal();
8153 }
8154
8155
8156
8157
8158 private static final List<Cell> MOCKED_LIST = new AbstractList<Cell>() {
8159
8160 @Override
8161 public void add(int index, Cell element) {
8162
8163 }
8164
8165 @Override
8166 public boolean addAll(int index, Collection<? extends Cell> c) {
8167 return false;
8168 }
8169
8170 @Override
8171 public KeyValue get(int index) {
8172 throw new UnsupportedOperationException();
8173 }
8174
8175 @Override
8176 public int size() {
8177 return 0;
8178 }
8179 };
8180
8181
8182
8183
8184
8185
8186
8187
8188
8189
8190 public static void main(String[] args) throws IOException {
8191 if (args.length < 1) {
8192 printUsageAndExit(null);
8193 }
8194 boolean majorCompact = false;
8195 if (args.length > 1) {
8196 if (!args[1].toLowerCase().startsWith("major")) {
8197 printUsageAndExit("ERROR: Unrecognized option <" + args[1] + ">");
8198 }
8199 majorCompact = true;
8200 }
8201 final Path tableDir = new Path(args[0]);
8202 final Configuration c = HBaseConfiguration.create();
8203 final FileSystem fs = FileSystem.get(c);
8204 final Path logdir = new Path(c.get("hbase.tmp.dir"));
8205 final String logname = "wal" + FSUtils.getTableName(tableDir) + System.currentTimeMillis();
8206
8207 final Configuration walConf = new Configuration(c);
8208 FSUtils.setRootDir(walConf, logdir);
8209 final WALFactory wals = new WALFactory(walConf, null, logname);
8210 try {
8211 processTable(fs, tableDir, wals, c, majorCompact);
8212 } finally {
8213 wals.close();
8214
8215 BlockCache bc = new CacheConfig(c).getBlockCache();
8216 if (bc != null) bc.shutdown();
8217 }
8218 }
8219
8220 @Override
8221 public long getOpenSeqNum() {
8222 return this.openSeqNum;
8223 }
8224
8225 @Override
8226 public Map<byte[], Long> getMaxStoreSeqId() {
8227 return this.maxSeqIdInStores;
8228 }
8229
8230 @Override
8231 public long getOldestSeqIdOfStore(byte[] familyName) {
8232 return wal.getEarliestMemstoreSeqNum(getRegionInfo().getEncodedNameAsBytes(), familyName);
8233 }
8234
8235 @Override
8236 public CompactionState getCompactionState() {
8237 boolean hasMajor = majorInProgress.get() > 0, hasMinor = minorInProgress.get() > 0;
8238 return (hasMajor ? (hasMinor ? CompactionState.MAJOR_AND_MINOR : CompactionState.MAJOR)
8239 : (hasMinor ? CompactionState.MINOR : CompactionState.NONE));
8240 }
8241
8242 public void reportCompactionRequestStart(boolean isMajor){
8243 (isMajor ? majorInProgress : minorInProgress).incrementAndGet();
8244 }
8245
8246 public void reportCompactionRequestEnd(boolean isMajor, int numFiles, long filesSizeCompacted) {
8247 int newValue = (isMajor ? majorInProgress : minorInProgress).decrementAndGet();
8248
8249
8250 compactionsFinished.incrementAndGet();
8251 compactionNumFilesCompacted.addAndGet(numFiles);
8252 compactionNumBytesCompacted.addAndGet(filesSizeCompacted);
8253
8254 assert newValue >= 0;
8255 }
8256
8257
8258
8259
8260
8261 @VisibleForTesting
8262 public long getSequenceId() {
8263 return this.mvcc.getReadPoint();
8264 }
8265
8266
8267
8268
8269
8270
8271
8272
8273
8274 private WALKey appendEmptyEdit(final WAL wal) throws IOException {
8275
8276 @SuppressWarnings("deprecation")
8277 WALKey key = new HLogKey(getRegionInfo().getEncodedNameAsBytes(),
8278 getRegionInfo().getTable(), WALKey.NO_SEQUENCE_ID, 0, null,
8279 HConstants.NO_NONCE, HConstants.NO_NONCE, getMVCC());
8280
8281
8282
8283 try {
8284 wal.append(getTableDesc(), getRegionInfo(), key, WALEdit.EMPTY_WALEDIT, false);
8285 } catch (Throwable t) {
8286
8287 getMVCC().complete(key.getWriteEntry());
8288 }
8289 return key;
8290 }
8291
8292
8293
8294
8295 @Override
8296 public void onConfigurationChange(Configuration conf) {
8297
8298 }
8299
8300
8301
8302
8303 @Override
8304 public void registerChildren(ConfigurationManager manager) {
8305 configurationManager = Optional.of(manager);
8306 for (Store s : this.stores.values()) {
8307 configurationManager.get().registerObserver(s);
8308 }
8309 }
8310
8311
8312
8313
8314 @Override
8315 public void deregisterChildren(ConfigurationManager manager) {
8316 for (Store s : this.stores.values()) {
8317 configurationManager.get().deregisterObserver(s);
8318 }
8319 }
8320
8321
8322
8323
8324 public RegionSplitPolicy getSplitPolicy() {
8325 return this.splitPolicy;
8326 }
8327 }