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