palashc commented on code in PR #1666:
URL: https://github.com/apache/phoenix/pull/1666#discussion_r1325169932
##########
phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java:
##########
@@ -317,16 +331,83 @@ protected QueryPlan optimizeQuery(CompilableStatement
stmt) throws SQLException
protected PhoenixResultSet executeQuery(final CompilableStatement stmt,
final QueryLogger queryLogger)
throws SQLException {
- return executeQuery(stmt, true, queryLogger, false);
+ return executeQuery(stmt, true, queryLogger, false,
this.ddlTimestampValidationEnabled);
}
protected PhoenixResultSet executeQuery(final CompilableStatement stmt,
final QueryLogger queryLogger, boolean noCommit)
throws SQLException {
- return executeQuery(stmt, true, queryLogger, noCommit);
+ return executeQuery(stmt, true, queryLogger, noCommit,
this.ddlTimestampValidationEnabled);
+ }
+
+ /**
+ * Build a request for the validateLastDDLTimestamp RPC.
+ * @param tableRef
+ * @return
+ */
+ private RegionServerEndpointProtos.ValidateLastDDLTimestampRequest
getValidateDDLTimestampRequest(TableRef tableRef) {
+ RegionServerEndpointProtos.ValidateLastDDLTimestampRequest.Builder
requestBuilder
+ =
RegionServerEndpointProtos.ValidateLastDDLTimestampRequest.newBuilder();
+ RegionServerEndpointProtos.LastDDLTimestampRequest.Builder innerBuilder
+ =
RegionServerEndpointProtos.LastDDLTimestampRequest.newBuilder();
+
+ byte[] tenantIDBytes = this.connection.getTenantId() == null ?
HConstants.EMPTY_BYTE_ARRAY : this.connection.getTenantId().getBytes();
+ byte[] schemaBytes = tableRef.getTable().getSchemaName() == null ?
HConstants.EMPTY_BYTE_ARRAY : tableRef.getTable().getSchemaName().getBytes();
+
+ innerBuilder.setTenantId(ByteStringer.wrap(tenantIDBytes));
+ innerBuilder.setSchemaName(ByteStringer.wrap(schemaBytes));
+
innerBuilder.setTableName(ByteStringer.wrap(tableRef.getTable().getTableName().getBytes()));
+
innerBuilder.setLastDDLTimestamp(tableRef.getTable().getLastDDLTimestamp());
+ requestBuilder.addLastDDLTimestampRequests(innerBuilder);
+ return requestBuilder.build();
+ }
+
+ /**
+ * Verifies that table metadata in client cache is up-to-date 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 tableRef
+ * @throws SQLException
+ */
+ private void validateDDLTimestamp(TableRef tableRef, boolean doRetry)
throws SQLException {
+
+ try (Admin admin = this.connection.getQueryServices().getAdmin()) {
+ // get all live region servers
+ List<ServerName> regionServers = new
ArrayList<>(admin.getRegionServers(true));
+ // pick one at random
+ ServerName regionServer =
regionServers.get(ThreadLocalRandom.current().nextInt(regionServers.size()));
+
+ LOGGER.debug("Sending DDL timestamp validation request for table
{} at regionserver {}",
+ tableRef.getTable().getTableName().getString(),
regionServer.toString());
+
+ // RPC
+ CoprocessorRpcChannel channel =
admin.coprocessorService(regionServer);
+ PhoenixRegionServerEndpoint.BlockingInterface service =
PhoenixRegionServerEndpoint.newBlockingStub(channel);
+ service.validateLastDDLTimestamp(null,
getValidateDDLTimestampRequest(tableRef));
+ }
+ // handle server side exceptions
+ catch (ServiceException | SQLException | IOException e) {
+ if (e instanceof ServiceException) {
Review Comment:
@shahrs87 any server side exception is wrapped in a
`com.google.protobuf.ServiceException`, that is what the RPC returns. I have
tried to differentiate between that case and the case when there is an
IOException or SQLException raised by `CQSI.getAdmin` or
`admin.getRegionServers`. We should parse server exception only in the former
case, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
