You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by vj...@apache.org on 2024/07/28 17:51:30 UTC
(phoenix) branch master updated: PHOENIX-7363 Protect server side metadata cache updates for the given PTable (#1943)
This is an automated email from the ASF dual-hosted git repository.
vjasani pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/master by this push:
new a96449825e PHOENIX-7363 Protect server side metadata cache updates for the given PTable (#1943)
a96449825e is described below
commit a96449825ec4d685de11f410933afaad35db7fe3
Author: Viraj Jasani <vj...@apache.org>
AuthorDate: Sun Jul 28 09:51:26 2024 -0800
PHOENIX-7363 Protect server side metadata cache updates for the given PTable (#1943)
---
.../org/apache/phoenix/query/QueryServices.java | 9 ++++
.../phoenix/coprocessor/MetaDataEndpointImpl.java | 52 +++++++++++++++++-----
2 files changed, 51 insertions(+), 10 deletions(-)
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
index f4f0574a1d..1a42cf5d3b 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
@@ -510,6 +510,15 @@ public interface QueryServices extends SQLCloseable {
String PHOENIX_GET_METADATA_READ_LOCK_ENABLED = "phoenix.get.metadata.read.lock.enabled";
+ /**
+ * If server side metadata cache is empty, take Phoenix writeLock for the given row
+ * and make sure we can acquire the writeLock within the configurable duration.
+ */
+ String PHOENIX_METADATA_CACHE_UPDATE_ROWLOCK_TIMEOUT =
+ "phoenix.metadata.update.rowlock.timeout";
+
+ long DEFAULT_PHOENIX_METADATA_CACHE_UPDATE_ROWLOCK_TIMEOUT = 60000;
+
/**
* Get executor service used for parallel scans
*/
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
index 4048086022..b54fd8c1e9 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
@@ -195,6 +195,7 @@ import org.apache.phoenix.expression.LiteralExpression;
import org.apache.phoenix.expression.ProjectedColumnExpression;
import org.apache.phoenix.expression.RowKeyColumnExpression;
import org.apache.phoenix.expression.visitor.StatelessTraverseAllExpressionVisitor;
+import org.apache.phoenix.hbase.index.LockManager;
import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
import org.apache.phoenix.hbase.index.util.GenericKeyValueBuilder;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
@@ -326,6 +327,9 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements RegionCopr
private static final byte[] PHYSICAL_TABLE_BYTES =
new byte[]{PTable.LinkType.PHYSICAL_TABLE.getSerializedValue()};
+ private LockManager lockManager;
+ private long metadataCacheRowLockTimeout;
+
// KeyValues for Table
private static final Cell TABLE_TYPE_KV = createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY,
TABLE_FAMILY_BYTES, TABLE_TYPE_BYTES);
@@ -642,8 +646,12 @@ TABLE_FAMILY_BYTES, TABLE_SEQ_NUM_BYTES);
throw new CoprocessorException("Must be loaded on a table region!");
}
+ this.lockManager = new LockManager();
phoenixAccessCoprocessorHost = new PhoenixMetaDataCoprocessorHost(this.env);
Configuration config = env.getConfiguration();
+ this.metadataCacheRowLockTimeout =
+ config.getLong(QueryServices.PHOENIX_METADATA_CACHE_UPDATE_ROWLOCK_TIMEOUT,
+ QueryServices.DEFAULT_PHOENIX_METADATA_CACHE_UPDATE_ROWLOCK_TIMEOUT);
this.accessCheckEnabled = config.getBoolean(QueryServices.PHOENIX_ACLS_ENABLED,
QueryServicesOptions.DEFAULT_PHOENIX_ACLS_ENABLED);
this.blockWriteRebuildIndex = config.getBoolean(QueryServices.INDEX_FAILURE_BLOCK_WRITE,
@@ -765,6 +773,7 @@ TABLE_FAMILY_BYTES, TABLE_SEQ_NUM_BYTES);
Scan scan = MetaDataUtil.newTableRowsScan(key, MIN_TABLE_TIMESTAMP, clientTimeStamp);
Cache<ImmutableBytesPtr, PMetaDataEntity> metaDataCache = GlobalCache.getInstance(this.env).getMetaDataCache();
PTable newTable;
+ region.startRegionOperation();
try (RegionScanner scanner = region.getScanner(scan)) {
PTable oldTable = (PTable) metaDataCache.getIfPresent(cacheKey);
long tableTimeStamp = oldTable == null ? MIN_TABLE_TIMESTAMP - 1 : oldTable.getTimeStamp();
@@ -782,6 +791,8 @@ TABLE_FAMILY_BYTES, TABLE_SEQ_NUM_BYTES);
}
metaDataCache.put(cacheKey, newTable);
}
+ } finally {
+ region.closeRegionOperation();
}
return newTable;
}
@@ -3920,8 +3931,7 @@ TABLE_FAMILY_BYTES, TABLE_SEQ_NUM_BYTES);
rowLock = acquireLock(region, key, null, true);
}
PTable table =
- getTableFromCache(cacheKey, clientTimeStamp, clientVersion);
- table = modifyIndexStateForOldClient(clientVersion, table);
+ getTableFromCacheWithModifiedIndexState(clientTimeStamp, clientVersion, cacheKey);
// We only cache the latest, so we'll end up building the table with every call if the
// client connection has specified an SCN.
// TODO: If we indicate to the client that we're returning an older version, but there's a
@@ -3934,17 +3944,32 @@ TABLE_FAMILY_BYTES, TABLE_SEQ_NUM_BYTES);
}
return table;
}
- // Query for the latest table first, since it's not cached
- table =
+ // take Phoenix row level write-lock as we need to protect metadata cache update
+ // after scanning SYSTEM.CATALOG to retrieve the PTable object
+ LockManager.RowLock phoenixRowLock =
+ lockManager.lockRow(key, this.metadataCacheRowLockTimeout);
+ try {
+ table = getTableFromCacheWithModifiedIndexState(clientTimeStamp, clientVersion,
+ cacheKey);
+ if (table != null && table.getTimeStamp() < clientTimeStamp) {
+ if (isTableDeleted(table)) {
+ return null;
+ }
+ return table;
+ }
+ // Query for the latest table first, since it's not cached
+ table =
buildTable(key, cacheKey, region, HConstants.LATEST_TIMESTAMP, clientVersion);
- if ((table != null && table.getTimeStamp() <= clientTimeStamp) ||
- (blockWriteRebuildIndex && table.getIndexDisableTimestamp() > 0)) {
+ if ((table != null && table.getTimeStamp() <= clientTimeStamp) || (
+ blockWriteRebuildIndex && table.getIndexDisableTimestamp() > 0)) {
+ return table;
+ }
+ // Otherwise, query for an older version of the table - it won't be cached
+ table = buildTable(key, cacheKey, region, clientTimeStamp, clientVersion);
return table;
+ } finally {
+ phoenixRowLock.release();
}
- // Otherwise, query for an older version of the table - it won't be cached
- table =
- buildTable(key, cacheKey, region, clientTimeStamp, clientVersion);
- return table;
} finally {
if (!wasLocked && rowLock != null) {
rowLock.release();
@@ -3952,6 +3977,13 @@ TABLE_FAMILY_BYTES, TABLE_SEQ_NUM_BYTES);
}
}
+ private PTable getTableFromCacheWithModifiedIndexState(long clientTimeStamp, int clientVersion,
+ ImmutableBytesPtr cacheKey) throws SQLException {
+ PTable table = getTableFromCache(cacheKey, clientTimeStamp, clientVersion);
+ table = modifyIndexStateForOldClient(clientVersion, table);
+ return table;
+ }
+
private List<PFunction> doGetFunctions(List<byte[]> keys, long clientTimeStamp) throws IOException, SQLException {
Cache<ImmutableBytesPtr, PMetaDataEntity> metaDataCache =
GlobalCache.getInstance(this.env).getMetaDataCache();