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