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