From 02e94b033bc2a04b0e9f42bd29a935a38cfff23d Mon Sep 17 00:00:00 2001 From: Kadir Ozdemir Date: Sun, 27 Sep 2020 15:59:20 -0700 Subject: [PATCH 1/2] PHOENIX-6160 Simplifying concurrent mutation handling for global Indexes --- .../hbase/index/IndexRegionObserver.java | 242 +++++++++++------- 1 file changed, 144 insertions(+), 98 deletions(-) diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java index edc281ec8f2..1dc33cd5188 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java @@ -71,7 +71,6 @@ import org.apache.phoenix.hbase.index.builder.IndexBuildManager; import org.apache.phoenix.hbase.index.builder.IndexBuilder; import org.apache.phoenix.hbase.index.covered.IndexMetaData; -import org.apache.phoenix.hbase.index.covered.update.ColumnReference; import org.apache.phoenix.hbase.index.metrics.MetricsIndexerSource; import org.apache.phoenix.hbase.index.metrics.MetricsIndexerSourceFactory; import org.apache.phoenix.hbase.index.table.HTableInterfaceReference; @@ -91,6 +90,8 @@ import org.apache.phoenix.util.ServerUtil.ConnectionType; import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import static org.apache.phoenix.coprocessor.IndexRebuildRegionScanner.applyNew; import static org.apache.phoenix.coprocessor.IndexRebuildRegionScanner.removeColumn; @@ -117,12 +118,15 @@ public class IndexRegionObserver extends BaseRegionObserver { * Class to represent pending data table rows */ private static class PendingRow { - private boolean concurrent = false; private long count = 1; + private BatchMutateContext lastContext; - public void add() { + PendingRow(BatchMutateContext context) { + lastContext = context; + } + public void add(BatchMutateContext context) { count++; - concurrent = true; + lastContext = context; } public void remove() { @@ -133,8 +137,8 @@ public long getCount() { return count; } - public boolean isConcurrent() { - return concurrent; + public BatchMutateContext getLastContext() { + return lastContext; } } @@ -150,9 +154,14 @@ public static void setFailDataTableUpdatesForTesting(boolean fail) { failDataTableUpdatesForTesting = fail; } + public enum BatchMutatePhase { + PRE, POST, FAILED + } // Hack to get around not being able to save any state between // coprocessor calls. TODO: remove after HBASE-18127 when available + private static class BatchMutateContext { + private BatchMutatePhase currentPhase = BatchMutatePhase.PRE; private final int clientVersion; // The collection of index mutations that will be applied before the data table mutations. The empty column (i.e., // the verified column) will have the value false ("unverified") on these mutations @@ -166,12 +175,36 @@ private static class BatchMutateContext { private HashSet rowsToLock = new HashSet<>(); // The current and next states of the data rows corresponding to the pending mutations private HashMap> dataRowStates; - // Data table pending mutations + // The previous concurrent batch contexts + private HashMap lastConcurrentBatchContext = null; + // The latches of the threads waiting for this batch to complete + private List waitList = null; private Map multiMutationMap; private BatchMutateContext(int clientVersion) { this.clientVersion = clientVersion; } + + public BatchMutatePhase getCurrentPhase() { + return currentPhase; + } + + public Put getNextDataRowState(ImmutableBytesPtr rowKeyPtr) { + Pair rowState = dataRowStates.get(rowKeyPtr); + if (rowState != null) { + return dataRowStates.get(rowKeyPtr).getSecond(); + } + return null; + } + + public CountDownLatch getCountDownLatch() { + if (waitList == null) { + waitList = new ArrayList<>(); + } + CountDownLatch countDownLatch = new CountDownLatch(1); + waitList.add(countDownLatch); + return countDownLatch; + } } private ThreadLocal batchMutateContext = @@ -211,9 +244,11 @@ private BatchMutateContext(int clientVersion) { private long slowIndexPrepareThreshold; private long slowPreIncrementThreshold; private int rowLockWaitDuration; + private int concurrentMutationWaitDuration; private String dataTableName; private static final int DEFAULT_ROWLOCK_WAIT_DURATION = 30000; + private static final int DEFAULT_CONCURRENT_MUTATION_WAIT_DURATION_IN_MS = 1000; @Override public void start(CoprocessorEnvironment e) throws IOException { @@ -245,6 +280,8 @@ public void start(CoprocessorEnvironment e) throws IOException { this.rowLockWaitDuration = env.getConfiguration().getInt("hbase.rowlock.wait.duration", DEFAULT_ROWLOCK_WAIT_DURATION); this.lockManager = new LockManager(); + this.concurrentMutationWaitDuration = env.getConfiguration().getInt("phoenix.index.concurrent.wait.duration.ms", + DEFAULT_CONCURRENT_MUTATION_WAIT_DURATION_IN_MS); // Metrics impl for the Indexer -- avoiding unnecessary indirection for hadoop-1/2 compat this.metricSource = MetricsIndexerSourceFactory.getInstance().getIndexerSource(); @@ -396,15 +433,22 @@ private void lockRows(BatchMutateContext context) throws IOException { } } + private void unlockRows(BatchMutateContext context) throws IOException { + for (RowLock rowLock : context.rowLocks) { + rowLock.release(); + } + context.rowLocks.clear(); + } + private void populatePendingRows(BatchMutateContext context) { for (RowLock rowLock : context.rowLocks) { ImmutableBytesPtr rowKey = rowLock.getRowKey(); PendingRow pendingRow = pendingRows.get(rowKey); if (pendingRow == null) { - pendingRows.put(rowKey, new PendingRow()); + pendingRows.put(rowKey, new PendingRow(context)); } else { // m is a mutation on a row that has already a pending mutation in progress from another batch - pendingRow.add(); + pendingRow.add(context); } } } @@ -593,15 +637,31 @@ private void handleLocalIndexUpdates(TableName table, private void getCurrentRowStates(ObserverContext c, BatchMutateContext context) throws IOException { Set keys = new HashSet(context.rowsToLock.size()); + context.dataRowStates = new HashMap>(context.rowsToLock.size()); for (ImmutableBytesPtr rowKeyPtr : context.rowsToLock) { - keys.add(PVarbinary.INSTANCE.getKeyRange(rowKeyPtr.get())); + PendingRow pendingRow = pendingRows.get(rowKeyPtr); + if (pendingRow != null && pendingRow.getLastContext().getCurrentPhase() == BatchMutatePhase.PRE) { + if (context.lastConcurrentBatchContext == null) { + context.lastConcurrentBatchContext = new HashMap<>(); + } + context.lastConcurrentBatchContext.put(rowKeyPtr, pendingRow.getLastContext()); + Put put = pendingRow.getLastContext().getNextDataRowState(rowKeyPtr); + if (put != null) { + context.dataRowStates.put(rowKeyPtr, new Pair(put, new Put(put))); + } + } + else { + keys.add(PVarbinary.INSTANCE.getKeyRange(rowKeyPtr.get())); + } + } + if (keys.isEmpty()) { + return; } Scan scan = new Scan(); ScanRanges scanRanges = ScanRanges.createPointLookup(new ArrayList(keys)); scanRanges.initializeScan(scan); SkipScanFilter skipScanFilter = scanRanges.getSkipScanFilter(); scan.setFilter(skipScanFilter); - context.dataRowStates = new HashMap>(context.rowsToLock.size()); try (RegionScanner scanner = c.getEnvironment().getRegion().getScanner(scan)) { boolean more = true; while(more) { @@ -750,43 +810,11 @@ protected PhoenixIndexMetaData getPhoenixIndexMetaData(ObserverContext> familyMap = multiMutation.getFamilyCellMap(); - for (ColumnReference columnReference : indexMaintainer.getIndexedColumns()) { - byte[] family = columnReference.getFamily(); - List cellList = familyMap.get(family); - if (cellList == null) { - return false; - } - boolean has = false; - for (Cell cell : cellList) { - if (CellUtil.matchingColumn(cell, family, columnReference.getQualifier())) { - has = true; - break; - } - } - if (!has) { - return false; - } - } - return true; - } - - private void preparePostIndexMutations(TableName table, - BatchMutateContext context, + private void preparePostIndexMutations(BatchMutateContext context, long now, - PhoenixIndexMetaData indexMetaData) - throws Throwable { + PhoenixIndexMetaData indexMetaData) { context.postIndexUpdates = ArrayListMultimap.create(); List maintainers = indexMetaData.getIndexMaintainers(); - // Check if we need to skip post index update for any of the rows for (IndexMaintainer indexMaintainer : maintainers) { byte[] emptyCF = indexMaintainer.getEmptyKeyValueFamily().copyBytesIfNecessary(); byte[] emptyCQ = indexMaintainer.getEmptyKeyValueQualifier(); @@ -794,43 +822,17 @@ private void preparePostIndexMutations(TableName table, new HTableInterfaceReference(new ImmutableBytesPtr(indexMaintainer.getIndexTableName())); List> updates = context.indexUpdates.get(hTableInterfaceReference); for (Pair update : updates) { - // Are there concurrent updates on the data table row? if so, skip post index updates - // and let read repair resolve conflicts - ImmutableBytesPtr rowKey = new ImmutableBytesPtr(update.getSecond()); - PendingRow pendingRow = pendingRows.get(rowKey); - if (!pendingRow.isConcurrent()) { - Mutation m = update.getFirst(); - if (m instanceof Put) { - Put verifiedPut = new Put(m.getRow()); - // Set the status of the index row to "verified" - verifiedPut.addColumn(emptyCF, emptyCQ, now, VERIFIED_BYTES); - context.postIndexUpdates.put(hTableInterfaceReference, verifiedPut); - } else { - context.postIndexUpdates.put(hTableInterfaceReference, m); - } + Mutation m = update.getFirst(); + if (m instanceof Put) { + Put verifiedPut = new Put(m.getRow()); + // Set the status of the index row to "verified" + verifiedPut.addColumn(emptyCF, emptyCQ, now, VERIFIED_BYTES); + context.postIndexUpdates.put(hTableInterfaceReference, verifiedPut); } else { - if (!hasAllIndexedColumns(indexMaintainer, context.multiMutationMap.get(rowKey))) { - // This batch needs to be retried since one of the concurrent mutations does not have the value - // for an indexed column. Not including an index column may lead to incorrect index row key - // generation for concurrent mutations since concurrent mutations are not serialized entirely - // and do not see each other's effect on data table. Throwing an IOException will result in - // retries of this batch. Before throwing exception, we need to remove reference counts and - // locks for the rows of this batch - removePendingRows(context); - context.indexUpdates.clear(); - for (RowLock rowLock : context.rowLocks) { - rowLock.release(); - } - context.rowLocks.clear(); - throw new IOException("One of the concurrent mutations does not have all indexed columns. " + - "The batch needs to be retried " + table.getNameAsString()); - } + context.postIndexUpdates.put(hTableInterfaceReference, m); } } } - - // We are done with handling concurrent mutations. So we can remove the rows of this batch from - // the collection of pending rows removePendingRows(context); context.indexUpdates.clear(); } @@ -853,41 +855,75 @@ private static boolean hasLocalIndex(PhoenixIndexMetaData indexMetaData) { return false; } + private void waitForPreviousConcurrentBatch(TableName table, BatchMutateContext context) + throws Throwable { + boolean done; + BatchMutatePhase phase; + long start = EnvironmentEdgeManager.currentTimeMillis(); + done = true; + for (BatchMutateContext lastContext : context.lastConcurrentBatchContext.values()) { + phase = lastContext.getCurrentPhase(); + if (phase == BatchMutatePhase.FAILED) { + done = false; + break; + } + if (phase == BatchMutatePhase.PRE) { + if (EnvironmentEdgeManager.currentTimeMillis() - start > concurrentMutationWaitDuration) { + done = false; + break; + } + CountDownLatch countDownLatch = lastContext.getCountDownLatch(); + // Release the locks so that the previous concurrent mutation can go into the post phase + unlockRows(context); + countDownLatch.await(DEFAULT_CONCURRENT_MUTATION_WAIT_DURATION_IN_MS, TimeUnit.MILLISECONDS); + // Acquire the locks again before letting the region proceed with data table updates + lockRows(context); + } + } + if (!done) { + // This batch needs to be retried since one of the previous concurrent batches has not completed yet. + // Throwing an IOException will result in retries of this batch. Before throwing exception, + // we need to remove reference counts and locks for the rows of this batch + removePendingRows(context); + context.indexUpdates.clear(); + for (RowLock rowLock : context.rowLocks) { + rowLock.release(); + } + context.rowLocks.clear(); + throw new IOException("One of the previous concurrent mutations has not completed. " + + "The batch needs to be retried " + table.getNameAsString()); + } + } + public void preBatchMutateWithExceptions(ObserverContext c, MiniBatchOperationInProgress miniBatchOp) throws Throwable { ignoreAtomicOperations(miniBatchOp); PhoenixIndexMetaData indexMetaData = getPhoenixIndexMetaData(c, miniBatchOp); BatchMutateContext context = new BatchMutateContext(indexMetaData.getClientVersion()); setBatchMutateContext(c, context); - Mutation firstMutation = miniBatchOp.getOperation(0); /* * Exclusively lock all rows so we get a consistent read * while determining the index updates */ populateRowsToLock(miniBatchOp, context); + // early exit if it turns out we don't have any update for indexes + if (context.rowsToLock.isEmpty()) { + return; + } lockRows(context); - long now = EnvironmentEdgeManager.currentTimeMillis(); - // Unless we're replaying edits to rebuild the index, we update the time stamp - // of the data table to prevent overlapping time stamps (which prevents index + // Update the timestamps of the data table mutations to prevent overlapping timestamps (which prevents index // inconsistencies as this case isn't handled correctly currently). setTimestamps(miniBatchOp, builder, now); - // Group all the updates for a single row into a single update to be processed (for local indexes, and global index retries) - Collection mutations = groupMutations(miniBatchOp, context); - // early exit if it turns out we don't have any edits - if (mutations == null || mutations.isEmpty()) { - return; - } - TableName table = c.getEnvironment().getRegion().getRegionInfo().getTable(); if (hasGlobalIndex(indexMetaData)) { + // Prepare current and next data rows states for pending mutations (for global indexes) + prepareDataRowStates(c, miniBatchOp, context, now); // Add the table rows in the mini batch to the collection of pending rows. This will be used to detect // concurrent updates populatePendingRows(context); - // Prepare current and next data rows states for pending mutations (for global indexes) - prepareDataRowStates(c, miniBatchOp, context, now); // early exit if it turns out we don't have any edits long start = EnvironmentEdgeManager.currentTimeMillis(); preparePreIndexMutations(context, now, indexMetaData); @@ -901,17 +937,19 @@ public void preBatchMutateWithExceptions(ObserverContext mutations = groupMutations(miniBatchOp, context); handleLocalIndexUpdates(table, miniBatchOp, mutations, indexMetaData); } if (failDataTableUpdatesForTesting) { @@ -942,9 +980,17 @@ public void postBatchMutateIndispensably(ObserverContext Date: Sun, 11 Oct 2020 16:13:17 -0700 Subject: [PATCH 2/2] Updated based on review comments --- .../hbase/index/IndexRegionObserver.java | 41 ++++++++++++++----- 1 file changed, 31 insertions(+), 10 deletions(-) diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java index 1dc33cd5188..74b8088f77a 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java @@ -118,12 +118,14 @@ public class IndexRegionObserver extends BaseRegionObserver { * Class to represent pending data table rows */ private static class PendingRow { - private long count = 1; + private int count; private BatchMutateContext lastContext; PendingRow(BatchMutateContext context) { + count = 1; lastContext = context; } + public void add(BatchMutateContext context) { count++; lastContext = context; @@ -133,7 +135,7 @@ public void remove() { count--; } - public long getCount() { + public int getCount() { return count; } @@ -157,11 +159,22 @@ public static void setFailDataTableUpdatesForTesting(boolean fail) { public enum BatchMutatePhase { PRE, POST, FAILED } + // Hack to get around not being able to save any state between // coprocessor calls. TODO: remove after HBASE-18127 when available + /** + * The concurrent batch of mutations is a set such that every pair of batches in this set has at least one common row. + * Since a BatchMutateContext object of a batch is modified only after the row locks for all the rows that are mutated + * by this batch are acquired, there can be only one thread can acquire the locks for its batch and safely access + * all the batch contexts in the set of concurrent batches. Because of this, we do not read atomic variables or + * additional locks to serialize the access to the BatchMutateContext objects. + */ + private static class BatchMutateContext { private BatchMutatePhase currentPhase = BatchMutatePhase.PRE; + // The max of reference counts on the pending rows of this batch at the time this batch arrives + private int maxPendingRowCount = 0; private final int clientVersion; // The collection of index mutations that will be applied before the data table mutations. The empty column (i.e., // the verified column) will have the value false ("unverified") on these mutations @@ -192,7 +205,7 @@ public BatchMutatePhase getCurrentPhase() { public Put getNextDataRowState(ImmutableBytesPtr rowKeyPtr) { Pair rowState = dataRowStates.get(rowKeyPtr); if (rowState != null) { - return dataRowStates.get(rowKeyPtr).getSecond(); + return rowState.getSecond(); } return null; } @@ -205,6 +218,10 @@ public CountDownLatch getCountDownLatch() { waitList.add(countDownLatch); return countDownLatch; } + + public int getMaxPendingRowCount() { + return maxPendingRowCount; + } } private ThreadLocal batchMutateContext = @@ -248,7 +265,7 @@ public CountDownLatch getCountDownLatch() { private String dataTableName; private static final int DEFAULT_ROWLOCK_WAIT_DURATION = 30000; - private static final int DEFAULT_CONCURRENT_MUTATION_WAIT_DURATION_IN_MS = 1000; + private static final int DEFAULT_CONCURRENT_MUTATION_WAIT_DURATION_IN_MS = 100; @Override public void start(CoprocessorEnvironment e) throws IOException { @@ -645,6 +662,9 @@ private void getCurrentRowStates(ObserverContext c context.lastConcurrentBatchContext = new HashMap<>(); } context.lastConcurrentBatchContext.put(rowKeyPtr, pendingRow.getLastContext()); + if (context.maxPendingRowCount < pendingRow.getCount()) { + context.maxPendingRowCount = pendingRow.getCount(); + } Put put = pendingRow.getLastContext().getNextDataRowState(rowKeyPtr); if (put != null) { context.dataRowStates.put(rowKeyPtr, new Pair(put, new Put(put))); @@ -859,7 +879,6 @@ private void waitForPreviousConcurrentBatch(TableName table, BatchMutateContext throws Throwable { boolean done; BatchMutatePhase phase; - long start = EnvironmentEdgeManager.currentTimeMillis(); done = true; for (BatchMutateContext lastContext : context.lastConcurrentBatchContext.values()) { phase = lastContext.getCurrentPhase(); @@ -868,14 +887,16 @@ private void waitForPreviousConcurrentBatch(TableName table, BatchMutateContext break; } if (phase == BatchMutatePhase.PRE) { - if (EnvironmentEdgeManager.currentTimeMillis() - start > concurrentMutationWaitDuration) { - done = false; - break; - } CountDownLatch countDownLatch = lastContext.getCountDownLatch(); // Release the locks so that the previous concurrent mutation can go into the post phase unlockRows(context); - countDownLatch.await(DEFAULT_CONCURRENT_MUTATION_WAIT_DURATION_IN_MS, TimeUnit.MILLISECONDS); + // Wait for at most one concurrentMutationWaitDuration for each level in the dependency tree of batches. + // lastContext.getMaxPendingRowCount() is the depth of the subtree rooted at the batch pointed by lastContext + if (!countDownLatch.await((lastContext.getMaxPendingRowCount() + 1) * concurrentMutationWaitDuration, + TimeUnit.MILLISECONDS)) { + done = false; + break; + } // Acquire the locks again before letting the region proceed with data table updates lockRows(context); }