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