[
https://issues.apache.org/jira/browse/PHOENIX-7026?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17784655#comment-17784655
]
ASF GitHub Bot commented on PHOENIX-7026:
-----------------------------------------
shahrs87 commented on code in PR #1726:
URL: https://github.com/apache/phoenix/pull/1726#discussion_r1388810619
##########
phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricType.java:
##########
@@ -150,6 +150,7 @@ public enum MetricType {
CLIENT_METADATA_CACHE_MISS_COUNTER("cmcm", "Number of cache misses for the
CQSI cache.", LogLevel.DEBUG, PLong.INSTANCE),
CLIENT_METADATA_CACHE_HIT_COUNTER("cmch", "Number of cache hits for the
CQSI cache.", LogLevel.DEBUG, PLong.INSTANCE),
PAGED_ROWS_COUNTER("prc", "Number of dummy rows returned to client due to
paging.", LogLevel.DEBUG, PLong.INSTANCE),
+ STALE_METADATA_CACHE_EXCEPTION_COUNTER("smcem", "Number of
StaleMetadataCacheException encountered.", LogLevel.DEBUG, PLong.INSTANCE),
Review Comment:
Why `smcem` ? What does the last m indicate? (metric?) If yes, then we
don't have in other metrics.
##########
phoenix-core/src/main/java/org/apache/phoenix/util/ValidateLastDDLTimestampUtil.java:
##########
@@ -0,0 +1,170 @@
+package org.apache.phoenix.util;
+
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
+import org.apache.hadoop.hbase.util.ByteStringer;
+import org.apache.phoenix.coprocessor.PhoenixRegionServerEndpoint;
+import org.apache.phoenix.coprocessor.generated.RegionServerEndpointProtos;
+import org.apache.phoenix.exception.StaleMetadataCacheException;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.schema.PName;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.PTableKey;
+import org.apache.phoenix.schema.PTableType;
+import org.apache.phoenix.schema.TableNotFoundException;
+import org.apache.phoenix.schema.TableRef;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.SQLException;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * Utility class for last ddl timestamp validation from the client.
+ */
+public class ValidateLastDDLTimestampUtil {
+
+ private static final Logger LOGGER = LoggerFactory
+ .getLogger(ValidateLastDDLTimestampUtil.class);
+
+ public static String getInfoString(PName tenantId, List<TableRef>
tableRefs) {
+ StringBuilder sb = new StringBuilder();
+ sb.append(String.format("Tenant: %s, ", tenantId));
+ for (TableRef tableRef : tableRefs) {
+ sb.append(String.format("{Schema: %s, Table: %s},",
+ tableRef.getTable().getSchemaName(),
+ tableRef.getTable().getTableName()));
+ }
+ return sb.toString();
+ }
+
+ /**
+ * Verifies that table metadata for given tables is up-to-date in client
cache with server.
+ * A random live region server is picked for invoking the RPC to validate
LastDDLTimestamp.
+ * Retry once if there was an error performing the RPC, otherwise throw
the Exception.
+ * @param tableRefs
+ * @throws SQLException
+ */
+ public static void validateLastDDLTimestamp(
+ PhoenixConnection conn, List<TableRef> tableRefs, boolean
isWritePath, boolean doRetry)
+ throws SQLException {
+
+ String infoString = getInfoString(conn.getTenantId(), tableRefs);
+ try (Admin admin = conn.getQueryServices().getAdmin()) {
+ // get all live region servers
+ List<ServerName> regionServers
+ = conn.getQueryServices().getLiveRegionServers();
+ // pick one at random
+ ServerName regionServer
+ =
regionServers.get(ThreadLocalRandom.current().nextInt(regionServers.size()));
+
+ LOGGER.debug("Sending DDL timestamp validation request for {} to
regionserver {}",
+ infoString, regionServer);
+
+ // RPC
+ CoprocessorRpcChannel channel =
admin.coprocessorService(regionServer);
+ PhoenixRegionServerEndpoint.BlockingInterface service
+ = PhoenixRegionServerEndpoint.newBlockingStub(channel);
+ RegionServerEndpointProtos.ValidateLastDDLTimestampRequest request
+ = getValidateDDLTimestampRequest(conn, tableRefs,
isWritePath);
+ service.validateLastDDLTimestamp(null, request);
+ } catch (Exception e) {
+ SQLException parsedException = ServerUtil.parseServerException(e);
+ if (parsedException instanceof StaleMetadataCacheException) {
+ throw parsedException;
+ }
+ //retry once for any exceptions other than
StaleMetadataCacheException
+ LOGGER.error("Error in validating DDL timestamp for {}",
infoString, parsedException);
+ if (doRetry) {
+ // update the list of live region servers
+ conn.getQueryServices().refreshLiveRegionServers();
+ validateLastDDLTimestamp(conn, tableRefs, isWritePath, false);
+ return;
+ }
+ throw parsedException;
+ }
+ }
+
+ /**
+ * Build a request for the validateLastDDLTimestamp RPC for the given
tables.
+ * 1. For a view, we need to add all its ancestors to the request in case
something changed in the hierarchy.
+ * 2. For an index, we need to add its parent table to the request in case
the index was dropped.
+ * 3. On the write path, we need to add all indexes of a table/view in
case index state was changed.
+ * @param tableRefs
+ * @return ValidateLastDDLTimestampRequest for the table in tableRef
+ */
+ private static RegionServerEndpointProtos.ValidateLastDDLTimestampRequest
+ getValidateDDLTimestampRequest(PhoenixConnection conn, List<TableRef>
tableRefs,
+ boolean isWritePath) throws
TableNotFoundException {
+
+ RegionServerEndpointProtos.ValidateLastDDLTimestampRequest.Builder
requestBuilder
+ =
RegionServerEndpointProtos.ValidateLastDDLTimestampRequest.newBuilder();
+ RegionServerEndpointProtos.LastDDLTimestampRequest.Builder
innerBuilder;
+
+ for (TableRef tableRef : tableRefs) {
+ innerBuilder =
RegionServerEndpointProtos.LastDDLTimestampRequest.newBuilder();
+
+ //when querying an index, we need to validate its parent table
+ //in case the index was dropped
+ if (PTableType.INDEX.equals(tableRef.getTable().getType())) {
+ PTableKey key = new PTableKey(conn.getTenantId(),
+ tableRef.getTable().getParentName().getString());
+ PTable parentTable = conn.getTable(key);
+ setLastDDLTimestampRequestParameters(innerBuilder,
conn.getTenantId(), parentTable);
+ requestBuilder.addLastDDLTimestampRequests(innerBuilder);
+ }
+
+ // add the tableRef to the request
+ innerBuilder =
RegionServerEndpointProtos.LastDDLTimestampRequest.newBuilder();
+ setLastDDLTimestampRequestParameters(innerBuilder,
conn.getTenantId(), tableRef.getTable());
+ requestBuilder.addLastDDLTimestampRequests(innerBuilder);
+
+ //when querying a view, we need to validate last ddl timestamps
for all its ancestors
+ if (PTableType.VIEW.equals(tableRef.getTable().getType())) {
+ PTable pTable = tableRef.getTable();
+ while (pTable.getParentName() != null) {
+ PTableKey key = new PTableKey(conn.getTenantId(),
+ pTable.getParentName().getString());
+ PTable parentTable = conn.getTable(key);
+ innerBuilder =
RegionServerEndpointProtos.LastDDLTimestampRequest.newBuilder();
+ setLastDDLTimestampRequestParameters(innerBuilder,
conn.getTenantId(), parentTable);
+ requestBuilder.addLastDDLTimestampRequests(innerBuilder);
+ pTable = parentTable;
+ }
+ }
+
+ //on the write path, we need to validate all indexes of a
table/view
+ //in case index state was changed
+ if (isWritePath) {
+ for (PTable idxPTable : tableRef.getTable().getIndexes()) {
+ innerBuilder =
RegionServerEndpointProtos.LastDDLTimestampRequest.newBuilder();
+ setLastDDLTimestampRequestParameters(innerBuilder,
conn.getTenantId(), idxPTable);
+ requestBuilder.addLastDDLTimestampRequests(innerBuilder);
+ }
+ }
+ }
+
Review Comment:
Still an extra line is present.
> Validate LAST_DDL_TIMESTAMP for write requests.
> -----------------------------------------------
>
> Key: PHOENIX-7026
> URL: https://issues.apache.org/jira/browse/PHOENIX-7026
> Project: Phoenix
> Issue Type: Sub-task
> Components: core
> Reporter: Rushabh Shah
> Assignee: Palash Chauhan
> Priority: Major
>
> We need to validate LAST_DDL_TIMESTAMP for write requests.
> We can't create an extra RPC like we do for read requests (PHOENIX-7025)
> since the semantics for executeQuery() and executeMutation() is very
> different.
> The usual semantic for batch write requests (assuming auto commit OFF) is:
> {noformat}
> conn = DriverManager.getConnection(getUrl(), props);
> conn.createStatement().execute("upsert into foo values (1,2)");
> conn.createStatement().execute("upsert into foo values (3,4)");
> conn.createStatement().execute("upsert into foo values (5,6)");
> conn.createStatement().execute("upsert into bar values ('a','b')");
> conn.createStatement().execute("upsert into bar values ('c','d')");
> conn.commit();
> {noformat}
> If we introduce an RPC for every execute() method, then it increases the
> latency of the write request.
> There are 2 options I can think of:
> # Call validateDDLTimestamp rpc for the very first execute() method for
> every table in the batch.
> ** Pros:
> *** An extra RPC is made only once in the whole batch for every table.
> *** Easy to implement.
> *** In case of StaleMetadataCacheException, the retry is very easy to
> implement.
> ** Cons:
> # Call validateDDLTimestamp rpc for all the tables in the batch when we call
> conn.commit()
> ** Pros
> *** We add only 1 extra RPC irrespective of the number of tables in the
> batch.
> ** Cons
> *** In case of StaleMetadataCacheException, the retry becomes very complex.
> We will have to re-create all the mutations that we have created in the
> executeMutation phase.
> *** If autocommit is turned ON, then we introduce an extra RPC for every
> execute() call which will introduce regression in this feature.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
