This is an automated email from the ASF dual-hosted git repository. xiangfu0 pushed a commit to branch xiangfu0/codex2/ratis-broker-planning in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 05ee934d65108e81a9b0893b94a65feba4a05f1b Author: Xiang Fu <[email protected]> AuthorDate: Tue Oct 20 03:40:53 2026 -0700 Capture coherent native broker planning inputs --- .../requesthandler/BaseBrokerRequestHandler.java | 18 +- .../BaseSingleStageBrokerRequestHandler.java | 188 ++++++++++----- .../broker/routing/NativeBrokerRoutingImage.java | 147 ++++++++++-- .../broker/routing/NativeBrokerRoutingManager.java | 138 +++++++++++ .../BaseSingleStageBrokerRequestHandlerTest.java | 23 +- .../routing/NativeBrokerRoutingImageTest.java | 261 ++++++++++++++++++++- pinot-common/src/main/proto/metadata.proto | 9 +- pinot-coordination-ratis/CONTROLLER-API-PARITY.md | 3 +- pinot-coordination-ratis/README.md | 29 ++- .../ratis/metadata/MetadataCatalog.java | 26 +- .../ratis/metadata/OfflineRoutingProtocol.java | 14 +- .../ratis/transport/RoutingGrpcService.java | 62 ++++- .../ratis/transport/RoutingRpcCodec.java | 14 +- .../ratis/metadata/TaskCatalogTest.java | 44 +++- .../ratis/transport/MetadataGrpcTransportTest.java | 60 ++++- .../pinot/core/transport/ServerInstance.java | 32 +++ .../pinot/tools/service/RatisStartersTest.java | 13 +- 17 files changed, 968 insertions(+), 113 deletions(-) diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java index fcc86726946..8fb72efc719 100644 --- a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java +++ b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java @@ -285,6 +285,22 @@ public abstract class BaseBrokerRequestHandler implements BrokerRequestHandler { /// enableMultiClusterRouting=true protected void validatePhysicalTablesWithMultiClusterRouting(Set<String> tableNames, Map<String, String> queryOptions) { + validatePhysicalTablesWithMultiClusterRoutingAgainstCache(tableNames, queryOptions, _tableCache); + } + + /// Uses the same captured table metadata as the request's compilation and routing phases. + protected void validatePhysicalTablesWithMultiClusterRouting(Set<String> tableNames, + Map<String, String> queryOptions, TableCache tableCache) { + if (tableCache == _tableCache) { + // Preserve existing subclass validation on the default backend. + validatePhysicalTablesWithMultiClusterRouting(tableNames, queryOptions); + } else { + validatePhysicalTablesWithMultiClusterRoutingAgainstCache(tableNames, queryOptions, tableCache); + } + } + + private static void validatePhysicalTablesWithMultiClusterRoutingAgainstCache(Set<String> tableNames, + Map<String, String> queryOptions, TableCache tableCache) { Preconditions.checkNotNull(tableNames, "Table names cannot be null when validating multi-cluster routing"); Preconditions.checkNotNull(queryOptions, "Query options cannot be null"); boolean isMultiClusterRoutingEnabled = queryOptions.containsKey(QueryOptionKey.ENABLE_MULTI_CLUSTER_ROUTING) @@ -292,7 +308,7 @@ public abstract class BaseBrokerRequestHandler implements BrokerRequestHandler { if (isMultiClusterRoutingEnabled) { for (String tableName : tableNames) { - if (!_tableCache.isLogicalTable(tableName)) { + if (!tableCache.isLogicalTable(tableName)) { throw QueryErrorCode.QUERY_VALIDATION.asException( "Physical table '" + tableName + "' cannot be queried with enableMultiClusterRouting=true. " + "Multi-cluster routing is only supported for logical tables. " diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java index 38bd566f434..22bf6f3e60a 100644 --- a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java +++ b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java @@ -253,6 +253,41 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ _enableQueryCancellation); } + /// Providers captured once before metadata-dependent compilation. Legacy requests retain their existing live + /// providers; a backend can supply immutable providers and retain additional per-request identity in a subclass. + protected static class RequestEnvironment { + private final TableCache _tableCache; + @Nullable + private final RoutingManager _routingManager; + + public RequestEnvironment(TableCache tableCache, @Nullable RoutingManager routingManager) { + _tableCache = Preconditions.checkNotNull(tableCache); + _routingManager = routingManager; + } + + public final TableCache tableCache() { + return _tableCache; + } + + @Nullable + public final RoutingManager routingManager() { + return _routingManager; + } + } + + /// Captures metadata before subquery execution, case/database resolution, and schema/config reads. Implementations + /// can reject unsupported query forms here; no later planning phase substitutes another environment. A recursive + /// subquery captures its own environment; backends requiring a single physical image must reject such forms here. + protected RequestEnvironment captureRequestEnvironment(PinotQuery query, @Nullable HttpHeaders httpHeaders) { + return new RequestEnvironment(_tableCache, _routingManager); + } + + /// Diagnostic fallback for syntax unsupported by SSE, before a request environment can be captured. A backend + /// without multistage metadata can disable this diagnostic instead of consulting unrelated live metadata. + protected boolean canCompileWithMultiStageEngine(String query, String database) { + return ParserUtils.canCompileWithMultiStageEngine(query, database, _tableCache); + } + @Override public void start() { if (_enableMultistageMigrationMetric) { @@ -271,7 +306,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ String database = query.getRight(); // Check if the query is a v2 supported query - if (!ParserUtils.canCompileWithMultiStageEngine(queryString, database, _tableCache)) { + if (!canCompileWithMultiStageEngine(queryString, database)) { _brokerMetrics.addMeteredGlobalValue(BrokerMeter.SINGLE_STAGE_QUERIES_INVALID_MULTI_STAGE, 1); } } @@ -413,6 +448,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ final PinotQuery _pinotQuery; final PinotQuery _serverPinotQuery; final Schema _schema; + final RequestEnvironment _environment; final String _tableName; final String _rawTableName; /// Dimension tables read by `lookup()` calls, which are not part of the data source @@ -420,13 +456,14 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ final BrokerResponse _errorOrLiteralOnlyBrokerResponse; public CompileResult(PinotQuery pinotQuery, PinotQuery serverPinotQuery, Schema schema, String tableName, - String rawTableName, Set<String> lookupTableNames) { + String rawTableName, Set<String> lookupTableNames, RequestEnvironment environment) { _pinotQuery = pinotQuery; _serverPinotQuery = serverPinotQuery; _schema = schema; _tableName = tableName; _rawTableName = rawTableName; _lookupTableNames = lookupTableNames; + _environment = environment; _errorOrLiteralOnlyBrokerResponse = null; } @@ -437,6 +474,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ _tableName = null; _rawTableName = null; _lookupTableNames = Set.of(); + _environment = null; _errorOrLiteralOnlyBrokerResponse = errorOrLiteralOnlyBrokerResponse; } } @@ -469,12 +507,14 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ return compileResult._errorOrLiteralOnlyBrokerResponse; } + RequestEnvironment environment = compileResult._environment; + TableCache tableCache = environment.tableCache(); Schema schema = compileResult._schema; String tableName = compileResult._tableName; String rawTableName = compileResult._rawTableName; PinotQuery pinotQuery = compileResult._pinotQuery; PinotQuery serverPinotQuery = compileResult._serverPinotQuery; - LogicalTableConfig logicalTableConfig = _tableCache.getLogicalTableConfig(rawTableName); + LogicalTableConfig logicalTableConfig = tableCache.getLogicalTableConfig(rawTableName); String database = DatabaseUtils.extractDatabaseFromFullyQualifiedTableName(tableName); String workloadName = QueryOptionsUtils.getWorkloadName(sqlNodeAndOptions.getOptions()); long compilationEndTimeNs = System.nanoTime(); @@ -483,7 +523,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ // Unlike the MSE, the SSE has no centralized exception handler that converts QueryException into // BrokerResponseNative, so we must catch and convert here to return a proper JSON error response. try { - validatePhysicalTablesWithMultiClusterRouting(Set.of(tableName), sqlNodeAndOptions.getOptions()); + validatePhysicalTablesWithMultiClusterRouting(Set.of(tableName), sqlNodeAndOptions.getOptions(), tableCache); } catch (QueryException e) { LOGGER.warn("Request {}: {}", requestId, e.getMessage()); requestContext.setErrorCode(e.getErrorCode()); @@ -637,7 +677,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ if (_materializedViewHandler != null && QueryOptionsUtils.isMaterializedViewRewriteEnabled(serverPinotQuery.getQueryOptions())) { MaterializedViewCompileOutcome outcome = applyMaterializedViewRewriteAtCompile( - requestId, serverPinotQuery, tableName, rawTableName, schema, _tableCache.isIgnoreCase()); + requestId, serverPinotQuery, tableName, rawTableName, schema, tableCache.isIgnoreCase(), tableCache); materializedViewContext = outcome._materializedViewContext; if (outcome._serverPinotQuery != serverPinotQuery) { /// FULL_REWRITE — pick up the swapped state and recompute `serverBrokerRequest`. @@ -654,9 +694,9 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ // Get the appropriate routing manager based on query options RoutingManager selectedRoutingManager = _multiClusterRoutingContext != null && logicalTableConfig != null ? _multiClusterRoutingContext.getRoutingManager(pinotQuery.getQueryOptions()) - : _routingManager; + : environment.routingManager(); // Get the tables hit by the request - TableRouteInfo routeInfo = routeProvider.getTableRouteInfo(tableName, _tableCache, selectedRoutingManager); + TableRouteInfo routeInfo = routeProvider.getTableRouteInfo(tableName, tableCache, selectedRoutingManager); if (!routeInfo.isExists()) { LOGGER.info("Table not found for request {}: {}", requestId, @@ -728,8 +768,8 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ offlinePinotQuery.getDataSource().setTableName(offlineTableName); assert timeBoundaryInfo != null; attachTimeBoundary(offlinePinotQuery, timeBoundaryInfo, true); - handleExpressionOverride(offlinePinotQuery, _tableCache.getExpressionOverrideMap(offlineTableName)); - handleTimestampIndexOverride(offlinePinotQuery, offlineTableConfig); + handleExpressionOverride(offlinePinotQuery, tableCache.getExpressionOverrideMap(offlineTableName)); + handleTimestampIndexOverride(offlinePinotQuery, offlineTableConfig, tableCache); // Re-optimize after attaching the time boundary filter so that filter optimizers (e.g. NumericalFilterOptimizer, // FlattenAndOrFilterOptimizer, MergeRangeFilterOptimizer) are applied to the time boundary predicate. _queryOptimizer.optimize(offlinePinotQuery, schema); @@ -738,8 +778,8 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ PinotQuery realtimePinotQuery = serverPinotQuery.deepCopy(); realtimePinotQuery.getDataSource().setTableName(realtimeTableName); attachTimeBoundary(realtimePinotQuery, timeBoundaryInfo, false); - handleExpressionOverride(realtimePinotQuery, _tableCache.getExpressionOverrideMap(realtimeTableName)); - handleTimestampIndexOverride(realtimePinotQuery, realtimeTableConfig); + handleExpressionOverride(realtimePinotQuery, tableCache.getExpressionOverrideMap(realtimeTableName)); + handleTimestampIndexOverride(realtimePinotQuery, realtimeTableConfig, tableCache); _queryOptimizer.optimize(realtimePinotQuery, schema); realtimeBrokerRequest = CalciteSqlCompiler.convertToBrokerRequest(realtimePinotQuery); @@ -749,8 +789,8 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ } else if (routeInfo.isOffline()) { // OFFLINE only setTableName(serverBrokerRequest, offlineTableName); - handleExpressionOverride(serverPinotQuery, _tableCache.getExpressionOverrideMap(offlineTableName)); - handleTimestampIndexOverride(serverPinotQuery, offlineTableConfig); + handleExpressionOverride(serverPinotQuery, tableCache.getExpressionOverrideMap(offlineTableName)); + handleTimestampIndexOverride(serverPinotQuery, offlineTableConfig, tableCache); offlineBrokerRequest = serverBrokerRequest; requestContext.setFanoutType(RequestContext.FanoutType.OFFLINE); @@ -758,8 +798,8 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ } else { // REALTIME only setTableName(serverBrokerRequest, realtimeTableName); - handleExpressionOverride(serverPinotQuery, _tableCache.getExpressionOverrideMap(realtimeTableName)); - handleTimestampIndexOverride(serverPinotQuery, realtimeTableConfig); + handleExpressionOverride(serverPinotQuery, tableCache.getExpressionOverrideMap(realtimeTableName)); + handleTimestampIndexOverride(serverPinotQuery, realtimeTableConfig, tableCache); realtimeBrokerRequest = serverBrokerRequest; requestContext.setFanoutType(RequestContext.FanoutType.REALTIME); @@ -778,7 +818,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ if (offlineBrokerRequest == null && realtimeBrokerRequest == null) { return getEmptyBrokerOnlyResponse(pinotQuery, serverPinotQuery, requestContext, tableName, requesterIdentity, - schema, query, database, workloadName, queryWasLogged); + schema, query, database, workloadName, queryWasLogged, tableCache); } if (offlineBrokerRequest != null && isFilterAlwaysTrue(offlineBrokerRequest.getPinotQuery())) { @@ -857,7 +897,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ } else { // If no route is found, send an empty response return getEmptyBrokerOnlyResponse(pinotQuery, serverPinotQuery, requestContext, tableName, requesterIdentity, - schema, query, database, workloadName, queryWasLogged); + schema, query, database, workloadName, queryWasLogged, tableCache); } } long routingEndTimeNs = System.nanoTime(); @@ -882,11 +922,11 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ .getTimeoutMs() : null; if (offlineBrokerRequest != null) { remainingTimeMs = setQueryTimeout(offlineTableName, logicalTableQueryTimeout, - offlineBrokerRequest.getPinotQuery().getQueryOptions(), timeSpentMs); + offlineBrokerRequest.getPinotQuery().getQueryOptions(), timeSpentMs, tableCache); } if (realtimeBrokerRequest != null) { remainingTimeMs = Math.max(remainingTimeMs, setQueryTimeout(realtimeTableName, logicalTableQueryTimeout, - realtimeBrokerRequest.getPinotQuery().getQueryOptions(), timeSpentMs)); + realtimeBrokerRequest.getPinotQuery().getQueryOptions(), timeSpentMs, tableCache)); } } catch (TimeoutException e) { String errorMessage = e.getMessage(); @@ -952,7 +992,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ BrokerResponseNative viewSplitResponse = tryExecuteMaterializedViewSplit(requestId, query, sqlNodeAndOptions, requesterIdentity, requestContext, brokerRequest, serverPinotQuery, schema, materializedViewContext, routeInfo, remainingTimeMs, errorMsgs, numPrunedSegmentsTotal, database, tableName, userRawTableName, - rlsFiltersApplied.get(), serverStats, selectedRoutingManager, routeProvider, queryWasLogged); + rlsFiltersApplied.get(), serverStats, selectedRoutingManager, routeProvider, queryWasLogged, tableCache); if (viewSplitResponse != null) { return viewSplitResponse; } @@ -965,7 +1005,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ /// `processBrokerRequest` would dispatch the user's "base table" query with /// `ts >= boundary` baked in, silently dropping the historical half of the timeline. materializedViewContext = MaterializedViewContext.empty(); - routeInfo = routeProvider.getTableRouteInfo(tableName, _tableCache, selectedRoutingManager); + routeInfo = routeProvider.getTableRouteInfo(tableName, tableCache, selectedRoutingManager); if (routeInfo instanceof ImplicitHybridTableRouteInfo) { ((ImplicitHybridTableRouteInfo) routeInfo).setOfflineBrokerRequest(offlineBrokerRequest); ((ImplicitHybridTableRouteInfo) routeInfo).setRealtimeBrokerRequest(realtimeBrokerRequest); @@ -1014,7 +1054,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ new QueryServers(query, offlineExecutionServers, realtimeExecutionServers)); try { brokerResponse = processBrokerRequest(requestId, brokerRequest, serverBrokerRequest, routeInfo, - remainingTimeMs, serverStats, requestContext); + remainingTimeMs, serverStats, requestContext, environment); brokerResponse.setClientRequestId(clientRequestId); } finally { onQueryFinish(requestId); @@ -1022,7 +1062,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ } } else { brokerResponse = processBrokerRequest(requestId, brokerRequest, serverBrokerRequest, routeInfo, - remainingTimeMs, serverStats, requestContext); + remainingTimeMs, serverStats, requestContext, environment); } brokerResponse.setTablesQueried(Set.of(userRawTableName)); if (_materializedViewHandler != null) { @@ -1062,7 +1102,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ // server returns STRING as default dataType for all columns in (some) scenarios where no rows are returned // this is an attempt to return more faithful information based on other sources - fillEmptyResponseSchema(pinotQuery, brokerResponse, schema, database, query); + fillEmptyResponseSchema(pinotQuery, brokerResponse, schema, database, query, tableCache); // Set total query processing time long totalTimeMs = System.currentTimeMillis() - requestContext.getRequestArrivalTimeMillis(); @@ -1123,7 +1163,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ requestContext.setErrorCode(QueryErrorCode.SQL_PARSING); // Check if the query is a v2 supported query String database = DatabaseUtils.extractDatabaseFromQueryRequest(sqlNodeAndOptions.getOptions(), httpHeaders); - if (ParserUtils.canCompileWithMultiStageEngine(query, database, _tableCache)) { + if (canCompileWithMultiStageEngine(query, database)) { return new CompileResult(new BrokerResponseNative(QueryErrorCode.SQL_PARSING, "It seems that the query is only supported by the multi-stage query engine, please retry the query " + "using " @@ -1187,6 +1227,16 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ return new CompileResult(new BrokerResponseNative(QueryErrorCode.QUERY_VALIDATION, "Table name not found")); } + RequestEnvironment environment; + try { + environment = Preconditions.checkNotNull(captureRequestEnvironment(serverPinotQuery, httpHeaders)); + } catch (Exception e) { + QueryErrorCode errorCode = QueryErrorCode.fromThrowable(e, QueryErrorCode.QUERY_VALIDATION); + requestContext.setErrorCode(errorCode); + return new CompileResult(new BrokerResponseNative(errorCode, e.getMessage())); + } + TableCache tableCache = environment.tableCache(); + try { handleSubquery(serverPinotQuery, requestId, request, requesterIdentity, requestContext, httpHeaders, accessControl); @@ -1198,12 +1248,12 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ return new CompileResult(new BrokerResponseNative(errorCode, e.getMessage())); } - boolean ignoreCase = _tableCache.isIgnoreCase(); + boolean ignoreCase = tableCache.isIgnoreCase(); String tableName; try { tableName = getActualTableName(DatabaseUtils.translateTableName(dataSource.getTableName(), httpHeaders, ignoreCase), - _tableCache); + tableCache); } catch (DatabaseConflictException e) { LOGGER.info("{}. Request {}: {}", e.getMessage(), requestId, _queryLogger.redactQuery(query, requestContext.getQueryFingerprint())); @@ -1232,7 +1282,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ } try { - Map<String, String> columnNameMap = _tableCache.getColumnNameMap(rawTableName); + Map<String, String> columnNameMap = tableCache.getColumnNameMap(rawTableName); if (columnNameMap != null) { updateColumnNames(rawTableName, serverPinotQuery, ignoreCase, columnNameMap); } @@ -1257,15 +1307,16 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ handleQueryLimitOverride(serverPinotQuery, _queryResponseLimit); } handleSegmentPartitionedDistinctCountOverride(serverPinotQuery, - getSegmentPartitionedColumns(_tableCache, tableName)); + getSegmentPartitionedColumns(tableCache, tableName)); if (_enableDistinctCountBitmapOverride) { handleDistinctCountBitmapOverride(serverPinotQuery); } - Schema schema = _tableCache.getSchema(rawTableName); + Schema schema = tableCache.getSchema(rawTableName); _queryOptimizer.optimize(serverPinotQuery, schema); - return new CompileResult(pinotQuery, serverPinotQuery, schema, tableName, rawTableName, lookupTableNames); + return new CompileResult(pinotQuery, serverPinotQuery, schema, tableName, rawTableName, lookupTableNames, + environment); } /// Mutable holder returned from [#applyMaterializedViewRewriteAtCompile] — Java has no out @@ -1297,7 +1348,8 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ /// query — fall back to the base-table path and surface the failure via /// `QUERY_REWRITE_EXCEPTIONS`. private MaterializedViewCompileOutcome applyMaterializedViewRewriteAtCompile(long requestId, - PinotQuery serverPinotQuery, String tableName, String rawTableName, Schema schema, boolean ignoreCase) { + PinotQuery serverPinotQuery, String tableName, String rawTableName, Schema schema, boolean ignoreCase, + TableCache tableCache) { /// Skip rewrite when the user enabled multi-cluster routing — the broker's /// `validatePhysicalTablesWithMultiClusterRouting` check (in `doHandleRequest`) inspects the /// post-rewrite table name and hard-rejects MV tables that aren't registered in the @@ -1324,7 +1376,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ /// per `validateMaterializedViewInvariants`. String mvLookupTableNameWithType = TableNameBuilder.getTableTypeFromTableName(tableName) != null ? tableName : TableNameBuilder.OFFLINE.tableNameWithType(rawTableName); - TableConfig userTableConfig = _tableCache.getTableConfig(mvLookupTableNameWithType); + TableConfig userTableConfig = tableCache.getTableConfig(mvLookupTableNameWithType); if (userTableConfig != null && userTableConfig.isMaterializedView()) { return new MaterializedViewCompileOutcome( serverPinotQuery, tableName, rawTableName, schema, MaterializedViewContext.empty()); @@ -1333,7 +1385,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ MaterializedViewContext materializedViewContext; try { materializedViewContext = _materializedViewHandler.compile(new MaterializedViewCompileContext( - serverPinotQuery, tableName, rawTableName, _tableCache)); + serverPinotQuery, tableName, rawTableName, tableCache)); } catch (Exception e) { LOGGER.error("Materialized view rewrite failed for request {} on table {}; " + "falling back to base-table query path", requestId, rawTableName, e); @@ -1354,7 +1406,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ /// rewrite engine does not have access to the table cache) so the broker emits the matching /// observability signal and continues against the base table unchanged. String baseRawTableName = TableNameBuilder.extractRawTableName(tableName); - if (_tableCache.getTableConfig(TableNameBuilder.REALTIME.tableNameWithType(baseRawTableName)) != null) { + if (tableCache.getTableConfig(TableNameBuilder.REALTIME.tableNameWithType(baseRawTableName)) != null) { LOGGER.warn("FULL_REWRITE skipped for request {} on hybrid base table {}: MV would drop " + "realtime data; falling back to base-table query path", requestId, baseRawTableName); _brokerMetrics.addMeteredTableValue(baseRawTableName, BrokerMeter.QUERY_REWRITE_EXCEPTIONS, 1); @@ -1381,7 +1433,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ serverQueryOptions.put(QueryOptionKey.MATERIALIZED_VIEW_REWRITE, "true"); String rewrittenTableName = plan.getMaterializedViewTableNameWithType(); String rewrittenRawTableName = TableNameBuilder.extractRawTableName(rewrittenTableName); - Schema rewrittenSchema = _tableCache.getSchema(rewrittenRawTableName); + Schema rewrittenSchema = tableCache.getSchema(rewrittenRawTableName); /// The handler's `compile()` already null-checks the MV schema before returning a plan, but /// the broker re-reads the schema here on a fresh `TableCache` query — between those two /// reads a concurrent MV drop or zk listener-driven eviction can leave the second lookup @@ -1403,7 +1455,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ /// rewritten MV query may reference columns whose case differs from the MV schema's /// canonical case. Wrap in try/catch so a case-sensitive cluster with a case-different MV /// column name falls back to the base-table path instead of failing the query. - Map<String, String> mvColumnNameMap = _tableCache.getColumnNameMap(rewrittenRawTableName); + Map<String, String> mvColumnNameMap = tableCache.getColumnNameMap(rewrittenRawTableName); if (mvColumnNameMap != null) { try { updateColumnNames(rewrittenRawTableName, rewrittenServerQuery, ignoreCase, mvColumnNameMap); @@ -1498,7 +1550,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ private BrokerResponseNative getEmptyBrokerOnlyResponse(PinotQuery pinotQuery, PinotQuery serverPinotQuery, RequestContext requestContext, String tableName, @Nullable RequesterIdentity requesterIdentity, Schema schema, - String query, String database, String workloadName, boolean queryWasLogged) { + String query, String database, String workloadName, boolean queryWasLogged, TableCache tableCache) { if (pinotQuery.isExplain()) { // EXPLAIN PLAN results to show that query is evaluated exclusively by Broker. return BrokerResponseNative.BROKER_ONLY_EXPLAIN_PLAN_OUTPUT; @@ -1534,7 +1586,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ } } } - fillEmptyResponseSchema(pinotQuery, brokerResponse, schema, database, query); + fillEmptyResponseSchema(pinotQuery, brokerResponse, schema, database, query, tableCache); } catch (Exception e) { LOGGER.warn("Caught exception while building empty response for request {}: {}, {}", requestContext.getRequestId(), query, e.getMessage()); @@ -1547,20 +1599,21 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ } private void fillEmptyResponseSchema(PinotQuery pinotQuery, BrokerResponse brokerResponse, Schema schema, - String database, String query) { + String database, String query, TableCache tableCache) { if (brokerResponse.getNumRowsResultSet() == 0) { boolean useMSE = QueryOptionsUtils.isUseMSEToFillEmptySchema(pinotQuery.getQueryOptions(), _useMSEToFillEmptyResponseSchema); - EmptyResponseUtils.fillEmptyResponseSchema(useMSE, brokerResponse, _tableCache, schema, database, query); + EmptyResponseUtils.fillEmptyResponseSchema(useMSE, brokerResponse, tableCache, schema, database, query); } } - private void handleTimestampIndexOverride(PinotQuery pinotQuery, @Nullable TableConfig tableConfig) { + private void handleTimestampIndexOverride(PinotQuery pinotQuery, @Nullable TableConfig tableConfig, + TableCache tableCache) { if (tableConfig == null || tableConfig.getFieldConfigList() == null) { return; } - Set<String> timestampIndexColumns = _tableCache.getTimestampIndexColumns(tableConfig.getTableName()); + Set<String> timestampIndexColumns = tableCache.getTimestampIndexColumns(tableConfig.getTableName()); if (CollectionUtils.isEmpty(timestampIndexColumns)) { return; } @@ -2370,7 +2423,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ /// For the overall query timeout, use query-level timeout (in the query options) if exists, or use table-level /// timeout (in the table config) if exists, or use instance-level timeout (in the broker config). private long setQueryTimeout(String tableNameWithType, Long logicalTableQueryTimeout, - Map<String, String> queryOptions, long timeSpentMs) + Map<String, String> queryOptions, long timeSpentMs, TableCache tableCache) throws TimeoutException { long queryTimeoutMs; Long queryLevelTimeoutMs = QueryOptionsUtils.getTimeoutMs(queryOptions); @@ -2378,7 +2431,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ // Use query-level timeout if exists queryTimeoutMs = queryLevelTimeoutMs; } else { - TableConfig tableConfig = _tableCache.getTableConfig(tableNameWithType); + TableConfig tableConfig = tableCache.getTableConfig(tableNameWithType); if (tableConfig != null && tableConfig.getQueryConfig() != null && tableConfig.getQueryConfig().getTimeoutMs() != null) { @@ -2581,13 +2634,14 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ MaterializedViewContext materializedViewContext, TableRouteInfo routeInfo, long remainingTimeMs, List<QueryProcessingException> errorMsgs, int numPrunedSegmentsTotal, String database, String tableName, String userRawTableName, boolean rlsFiltersApplied, ServerStats serverStats, - RoutingManager selectedRoutingManager, TableRouteProvider routeProvider, boolean queryWasLogged) { + RoutingManager selectedRoutingManager, TableRouteProvider routeProvider, boolean queryWasLogged, + TableCache tableCache) { String clientRequestId = extractClientRequestId(sqlNodeAndOptions); MaterializedViewSplitDispatcher dispatcher = (originalReq, baseQ, baseRoute, baseSch, viewQ, viewTable, viewSch, timeoutMs) -> dispatchMaterializedViewSplit(requestId, originalReq, baseQ, baseRoute, baseSch, viewQ, viewTable, viewSch, timeoutMs, serverStats, requestContext, selectedRoutingManager, routeProvider, - clientRequestId, query); + clientRequestId, query, tableCache); MaterializedViewSplitExecutionContext splitCtx = MaterializedViewSplitExecutionContext.builder() .originalBrokerRequest(brokerRequest) .baseServerPinotQuery(serverPinotQuery) @@ -2617,7 +2671,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ viewSplitResponse.addException(errorMsg); } viewSplitResponse.setNumSegmentsPrunedByBroker(numPrunedSegmentsTotal); - fillEmptyResponseSchema(brokerRequest.getPinotQuery(), viewSplitResponse, schema, database, query); + fillEmptyResponseSchema(brokerRequest.getPinotQuery(), viewSplitResponse, schema, database, query, tableCache); long totalTimeMs = System.currentTimeMillis() - requestContext.getRequestArrivalTimeMillis(); viewSplitResponse.setTimeUsedMs(totalTimeMs); augmentStatistics(requestContext, viewSplitResponse); @@ -2640,12 +2694,13 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ BrokerRequest originalBrokerRequest, PinotQuery baseQueryWithTimeFilter, TableRouteInfo baseRouteInfo, Schema baseSchema, PinotQuery viewQueryWithTimeFilter, String viewTableNameWithType, Schema viewSchema, long timeoutMs, ServerStats serverStats, RequestContext requestContext, RoutingManager routingManager, - TableRouteProvider routeProvider, @Nullable String clientRequestId, String query) throws Exception { + TableRouteProvider routeProvider, @Nullable String clientRequestId, String query, TableCache tableCache) + throws Exception { /// 1. Optimize and route the base-table side (handles hybrid offline/realtime split). _queryOptimizer.optimize(baseQueryWithTimeFilter, baseSchema); BrokerRequest baseBrokerRequest = CalciteSqlCompiler.convertToBrokerRequest(baseQueryWithTimeFilter); ImplicitHybridTableRouteInfo hybridBaseRoute = - prepareBaseTableHybridRoute(baseBrokerRequest, baseRouteInfo, baseSchema); + prepareBaseTableHybridRoute(baseBrokerRequest, baseRouteInfo, baseSchema, tableCache); BrokerRequest baseOfflineBrokerRequest = hybridBaseRoute.getOfflineBrokerRequest(); BrokerRequest baseRealtimeBrokerRequest = hybridBaseRoute.getRealtimeBrokerRequest(); @@ -2664,11 +2719,11 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ /// Apply MV-table expression overrides and `$ts$DAY` timestamp-index hints before /// optimization, mirroring the non-split path. Without this, overrides configured on the /// MV table would be silently dropped on the SPLIT path. - handleExpressionOverride(viewQueryWithTimeFilter, _tableCache.getExpressionOverrideMap(viewTableNameWithType)); - handleTimestampIndexOverride(viewQueryWithTimeFilter, _tableCache.getTableConfig(viewTableNameWithType)); + handleExpressionOverride(viewQueryWithTimeFilter, tableCache.getExpressionOverrideMap(viewTableNameWithType)); + handleTimestampIndexOverride(viewQueryWithTimeFilter, tableCache.getTableConfig(viewTableNameWithType), tableCache); _queryOptimizer.optimize(viewQueryWithTimeFilter, viewSchema); BrokerRequest viewBrokerRequest = CalciteSqlCompiler.convertToBrokerRequest(viewQueryWithTimeFilter); - TableRouteInfo viewRouteInfo = routeProvider.getTableRouteInfo(viewTableNameWithType, _tableCache, routingManager); + TableRouteInfo viewRouteInfo = routeProvider.getTableRouteInfo(viewTableNameWithType, tableCache, routingManager); /// The MV table can disappear between compile (broker MV cache) and execute (route lookup) /// — operator dropped it, controller mid-state-transition, etc. Treat missing-route the /// same way the base path does (`BROKER_RESOURCE_MISSING` with a graceful message); the @@ -2758,7 +2813,7 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ /// without these, table-level overrides configured on the base table would be silently /// dropped on the SPLIT path, producing divergent results vs. the base fallback path. private ImplicitHybridTableRouteInfo prepareBaseTableHybridRoute(BrokerRequest baseBrokerRequest, - TableRouteInfo baseRouteInfo, Schema schema) { + TableRouteInfo baseRouteInfo, Schema schema, TableCache tableCache) { Preconditions.checkState(baseRouteInfo instanceof ImplicitHybridTableRouteInfo, "MV split execution requires ImplicitHybridTableRouteInfo but got: %s", baseRouteInfo.getClass().getSimpleName()); @@ -2778,8 +2833,8 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ if (timeBoundaryInfo != null) { attachTimeBoundary(offlinePinotQuery, timeBoundaryInfo, true); } - handleExpressionOverride(offlinePinotQuery, _tableCache.getExpressionOverrideMap(offlineTableName)); - handleTimestampIndexOverride(offlinePinotQuery, offlineTableConfig); + handleExpressionOverride(offlinePinotQuery, tableCache.getExpressionOverrideMap(offlineTableName)); + handleTimestampIndexOverride(offlinePinotQuery, offlineTableConfig, tableCache); _queryOptimizer.optimize(offlinePinotQuery, schema); BrokerRequest offlineBrokerRequest = CalciteSqlCompiler.convertToBrokerRequest(offlinePinotQuery); @@ -2788,8 +2843,8 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ if (timeBoundaryInfo != null) { attachTimeBoundary(realtimePinotQuery, timeBoundaryInfo, false); } - handleExpressionOverride(realtimePinotQuery, _tableCache.getExpressionOverrideMap(realtimeTableName)); - handleTimestampIndexOverride(realtimePinotQuery, realtimeTableConfig); + handleExpressionOverride(realtimePinotQuery, tableCache.getExpressionOverrideMap(realtimeTableName)); + handleTimestampIndexOverride(realtimePinotQuery, realtimeTableConfig, tableCache); _queryOptimizer.optimize(realtimePinotQuery, schema); BrokerRequest realtimeBrokerRequest = CalciteSqlCompiler.convertToBrokerRequest(realtimePinotQuery); @@ -2798,14 +2853,14 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ } else if (baseRouteInfo.isOffline()) { setTableName(baseBrokerRequest, offlineTableName); handleExpressionOverride(baseBrokerRequest.getPinotQuery(), - _tableCache.getExpressionOverrideMap(offlineTableName)); - handleTimestampIndexOverride(baseBrokerRequest.getPinotQuery(), offlineTableConfig); + tableCache.getExpressionOverrideMap(offlineTableName)); + handleTimestampIndexOverride(baseBrokerRequest.getPinotQuery(), offlineTableConfig, tableCache); hybridRoute.setOfflineBrokerRequest(baseBrokerRequest); } else { setTableName(baseBrokerRequest, realtimeTableName); handleExpressionOverride(baseBrokerRequest.getPinotQuery(), - _tableCache.getExpressionOverrideMap(realtimeTableName)); - handleTimestampIndexOverride(baseBrokerRequest.getPinotQuery(), realtimeTableConfig); + tableCache.getExpressionOverrideMap(realtimeTableName)); + handleTimestampIndexOverride(baseBrokerRequest.getPinotQuery(), realtimeTableConfig, tableCache); hybridRoute.setRealtimeBrokerRequest(baseBrokerRequest); } return hybridRoute; @@ -2833,6 +2888,15 @@ public abstract class BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ } + /// Dispatch seam retaining the exact providers used during compilation and routing. Existing transports keep + /// their implementation; a backend requiring captured identities can override this overload explicitly. + protected BrokerResponseNative processBrokerRequest(long requestId, BrokerRequest originalBrokerRequest, + BrokerRequest serverBrokerRequest, TableRouteInfo route, long timeoutMs, ServerStats serverStats, + RequestContext requestContext, RequestEnvironment environment) throws Exception { + return processBrokerRequest(requestId, originalBrokerRequest, serverBrokerRequest, route, timeoutMs, serverStats, + requestContext); + } + /// Processes the optimized broker requests for both OFFLINE and REALTIME table. /// TODO: Directly take PinotQuery protected abstract BrokerResponseNative processBrokerRequest(long requestId, BrokerRequest originalBrokerRequest, diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/NativeBrokerRoutingImage.java b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/NativeBrokerRoutingImage.java index a7943bca6c7..1bf5fd5fd39 100644 --- a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/NativeBrokerRoutingImage.java +++ b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/NativeBrokerRoutingImage.java @@ -35,16 +35,23 @@ import java.util.UUID; import org.apache.pinot.common.config.provider.NativeOfflineSegmentAction.ObjectVersion; import org.apache.pinot.common.config.provider.NativeOfflineSegmentAction.Participant; import org.apache.pinot.common.config.provider.NativeOfflineSegmentAction.Role; +import org.apache.pinot.common.config.provider.StaticTableCache; +import org.apache.pinot.common.config.provider.TableCache; import org.apache.pinot.common.coordination.NativeOfflineSegmentWorker; import org.apache.pinot.common.request.context.NativeQueryContext; import org.apache.pinot.common.request.context.NativeQueryContext.ProcessIdentity; import org.apache.pinot.common.request.context.NativeQueryContext.Segment; import org.apache.pinot.common.request.context.NativeQueryDeadline; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc; +import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.CatalogRecord; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ParticipantRole; +import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ParticipantState; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ParticipantToken; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ReplicaRouting; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.RoutingSnapshot; +import org.apache.pinot.core.routing.RoutingManager; +import org.apache.pinot.core.routing.SegmentsToQuery; +import org.apache.pinot.core.transport.ServerInstance; import org.apache.pinot.spi.config.table.TableType; import org.apache.pinot.spi.utils.builder.TableNameBuilder; @@ -60,6 +67,7 @@ public final class NativeBrokerRoutingImage { public static final int MAX_RECORDS = 4096; public static final int MAX_BYTES = 2 * 1024 * 1024; private final RoutingSnapshot _snapshot; + private final Map<String, ParticipantState> _serverAuthorities; public NativeBrokerRoutingImage(RoutingSnapshot snapshot, ParticipantToken broker, String tableName) throws IOException { @@ -75,6 +83,14 @@ public final class NativeBrokerRoutingImage { checkArgument(identities.add(object(table.getTable(), broker.getHistoryId()).objectId()) && identities.add(object(table.getSchema(), broker.getHistoryId()).objectId()), "Routing table/schema identities overlap"); + checkArgument(table.hasTableConfig() == table.hasSchemaRecord(), "Partial native planning metadata"); + if (table.hasTableConfig()) { + validatePlanningRecord(table.getTableConfig(), table.getTable(), tableName, snapshot.getRevision().getIndex()); + validatePlanningRecord(table.getSchemaRecord(), table.getSchema(), table.getSchemaName(), + snapshot.getRevision().getIndex()); + NativeOfflineSegmentWorker.decodeTable(table.getTableConfig()); + NativeOfflineSegmentWorker.decodeSchema(table.getSchemaRecord()); + } Set<String> names = new HashSet<>(); Set<UUID> materializations = new HashSet<>(); Set<UUID> installations = new HashSet<>(); @@ -111,6 +127,33 @@ public final class NativeBrokerRoutingImage { "Routing image mixes query endpoint approvals"); } } + Map<String, ParticipantState> authorities = new HashMap<>(); + for (ParticipantState state : table.getServerAuthoritiesList()) { + checkArgument(++records <= MAX_RECORDS, "Server authority exceeds routing record bound"); + ParticipantToken token = state.getParticipant(); + participant(token, broker.getHistoryId(), ParticipantRole.SERVER); + checkArgument(state.hasParticipant() && token.equals(servers.get(token.getParticipantId())) + && authorities.put(token.getParticipantId(), state) == null && state.getReady() + && !state.getRetired() && !state.getDraining() && state.getExpiryRound() > snapshot.getControl().getRound() + && state.getLastModifiedIndex() > 0 && state.getLastModifiedIndex() <= snapshot.getRevision().getIndex() + && state.getLastRenewalSequence() >= 0 && state.hasAppliedRevision() + && state.getAppliedRevision().getHistoryId().equals(broker.getHistoryId()) + && state.getAppliedRevision().getIndex() >= 0 + && state.getAppliedRevision().getIndex() <= snapshot.getRevision().getIndex(), + "Invalid, repeated or unreferenced routing server authority"); + checkArgument(state.getCapabilitiesCount() == 0 + || state.getCapabilitiesList().equals(List.of(MetadataRpc.ServingCapability.SERVER_QUERY)), + "Invalid routing server capabilities"); + String endpoint = endpoints.get(token.getParticipantId()); + checkArgument(endpoint.isEmpty() + || state.getCapabilitiesList().contains(MetadataRpc.ServingCapability.SERVER_QUERY) + && endpoint.equals(state.getEndpointsMap().get("queryGrpc")) + && "2".equals(state.getEndpointsMap().get("nativeQueryVersion")), + "Query endpoint differs from its current server authority"); + } + checkArgument(!table.hasTableConfig() && authorities.isEmpty() || authorities.keySet().equals(servers.keySet()), + "Incomplete native planning server authorities"); + _serverAuthorities = Map.copyOf(authorities); _snapshot = snapshot; } @@ -143,12 +186,83 @@ public final class NativeBrokerRoutingImage { return _snapshot.getTable().getSegmentsCount(); } + public MetadataRpc.MembershipControl control() { + return _snapshot.getControl(); + } + + public Map<String, ParticipantState> serverAuthorities() { + return _serverAuthorities; + } + + /// Exact planning/content/assignment/session/endpoint identity, excluding routine server lease renewals. + public MetadataRpc.TableRouting queryIdentity() { + return _snapshot.getTable().toBuilder().clearServerAuthorities().build(); + } + + public boolean hasPlanningMetadata() { + return _snapshot.getTable().hasTableConfig() && _snapshot.getTable().hasSchemaRecord(); + } + + /// Derives ordinary Pinot name/column/expression maps from freshly decoded request-owned values. Virtual columns + /// or later planner mutations cannot modify the canonical snapshot or another request's cache. + public TableCache newTableCache(boolean ignoreCase) { + checkState(hasPlanningMetadata(), "Native planning requires complete captured table and schema records"); + try { + return new StaticTableCache( + List.of(NativeOfflineSegmentWorker.decodeTable(_snapshot.getTable().getTableConfig())), + List.of(NativeOfflineSegmentWorker.decodeSchema(_snapshot.getTable().getSchemaRecord())), + List.of(), ignoreCase); + } catch (IOException e) { + throw new IllegalStateException("Validated native planning records cannot be decoded", e); + } + } + + public RoutingManager newRoutingManager() { + checkState(hasPlanningMetadata(), "Native planning requires complete captured table and schema records"); + Selection selection = select(); + Map<ServerInstance, SegmentsToQuery> routes = new HashMap<>(); + selection.segments().forEach((id, segments) -> { + URI endpoint = endpoint(selection.targets().get(id).getEndpoint()); + routes.put(ServerInstance.forNativeGrpc(id, endpoint.getHost(), endpoint.getPort()), + new SegmentsToQuery(segments.stream().map(Segment::name).toList(), List.of())); + }); + return new NativeBrokerRoutingManager(tableName(), routes); + } + + /// Both normal Pinot providers belong to this one immutable table image. Create a new pair for each request. + public PlanningMetadata planningMetadata(boolean ignoreCase) { + return new PlanningMetadata(newTableCache(ignoreCase), newRoutingManager()); + } + + public record PlanningMetadata(TableCache tableCache, RoutingManager routingManager) { + } + /// Selects exactly one approved replica for every segment, in deterministic order. The same originating deadline /// is attached to every fanout request; callers must preserve it across retries and independently qualify its clock. public List<ServerRequest> requests(UUID queryId, UUID attemptId, NativeQueryDeadline deadline) { requireNonNull(queryId, "queryId"); requireNonNull(attemptId, "attemptId"); requireNonNull(deadline, "deadline"); + var table = _snapshot.getTable(); + Selection selection = select(); + ParticipantToken broker = _snapshot.getParticipant().getParticipant(); + ProcessIdentity process = new ProcessIdentity(broker.getParticipantId(), uuid(broker.getIncarnation()), + broker.getRegistrationEpoch()); + List<ServerRequest> requests = new ArrayList<>(); + selection.segments().forEach((id, segments) -> { + ReplicaRouting replica = selection.targets().get(id); + ParticipantToken server = replica.getServer(); + NativeQueryContext context = new NativeQueryContext(NativeQueryContext.VERSION, uuid(broker.getHistoryId()), + process, new Participant(id, Role.SERVER, uuid(server.getIncarnation()), server.getRegistrationEpoch()), + queryId, attemptId, table.getTableName(), object(table.getTable(), broker.getHistoryId()), + table.getSchemaName(), object(table.getSchema(), broker.getHistoryId()), segments, List.of()) + .withDeadline(deadline); + requests.add(new ServerRequest(endpoint(replica.getEndpoint()), context)); + }); + return List.copyOf(requests); + } + + private Selection select() { var table = _snapshot.getTable(); checkState(table.getSegmentsCount() > 0, "Native empty-table query execution is not supported"); Map<String, List<Segment>> selected = new TreeMap<>(); @@ -165,27 +279,23 @@ public final class NativeBrokerRoutingImage { object(replica.getAssignment(), _snapshot.getRevision().getHistoryId()), uuid(replica.getMaterializationId()))); } - ParticipantToken broker = _snapshot.getParticipant().getParticipant(); - ProcessIdentity process = new ProcessIdentity(broker.getParticipantId(), uuid(broker.getIncarnation()), - broker.getRegistrationEpoch()); - List<ServerRequest> requests = new ArrayList<>(); - selected.forEach((id, segments) -> { - ReplicaRouting replica = targets.get(id); - ParticipantToken server = replica.getServer(); - NativeQueryContext context = new NativeQueryContext(NativeQueryContext.VERSION, uuid(broker.getHistoryId()), - process, new Participant(id, Role.SERVER, uuid(server.getIncarnation()), server.getRegistrationEpoch()), - queryId, attemptId, table.getTableName(), object(table.getTable(), broker.getHistoryId()), - table.getSchemaName(), object(table.getSchema(), broker.getHistoryId()), segments, List.of()) - .withDeadline(deadline); - requests.add(new ServerRequest(endpoint(replica.getEndpoint()), context)); - }); - return List.copyOf(requests); + return new Selection(selected, targets); + } + + private record Selection(Map<String, List<Segment>> segments, Map<String, ReplicaRouting> targets) { } /// An exact destination and identity envelope, not a dispatched or admitted query. public record ServerRequest(URI endpoint, NativeQueryContext context) { } + private static void validatePlanningRecord(CatalogRecord record, MetadataRpc.ObjectVersion expected, String name, + long index) { + checkArgument(record.getObjectId().equals(expected.getObjectId()) && record.getVersion() == expected.getVersion() + && record.getName().equals(name) && record.getLastModifiedIndex() <= index, + "Planning record differs from captured routing identity or revision"); + } + private static void validateHeader(RoutingSnapshot snapshot, ParticipantToken broker) throws IOException { NativeOfflineSegmentWorker.validateMessage(snapshot); participant(broker, broker.getHistoryId(), ParticipantRole.BROKER); @@ -213,6 +323,13 @@ public final class NativeBrokerRoutingImage { checkArgument(policy.getVersion() > 0 && policy.getMinTickIntervalMillis() > 0 && policy.getLeaseRounds() >= 2 && policy.getPermitLifetimeMillis() > 0 && policy.getMaxQueryLifetimeMillis() > 0 && policy.getMaxRevisionLag() > 0, "Invalid routing membership policy"); + checkArgument((policy.getPermitLifetimeMillis() - 1) / policy.getMinTickIntervalMillis() + < policy.getLeaseRounds() - 1 + && control.getMinimumTickDelayMillis() >= policy.getMinTickIntervalMillis() + && control.getMaxConfiguredPermitLifetimeMillis() >= policy.getPermitLifetimeMillis() + && (control.getMaxConfiguredPermitLifetimeMillis() == policy.getPermitLifetimeMillis() + || control.getMinimumTickDelayMillis() >= control.getMaxConfiguredPermitLifetimeMillis()), + "Routing control timing does not protect serving authority"); } private static ObjectVersion object(MetadataRpc.ObjectVersion value, String history) { diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/NativeBrokerRoutingManager.java b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/NativeBrokerRoutingManager.java new file mode 100644 index 00000000000..d4b1d62c43e --- /dev/null +++ b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/NativeBrokerRoutingManager.java @@ -0,0 +1,138 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.broker.routing; + +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Function; +import java.util.stream.Collectors; +import javax.annotation.Nullable; +import org.apache.pinot.common.request.BrokerRequest; +import org.apache.pinot.common.utils.config.QueryOptionsUtils; +import org.apache.pinot.core.routing.RoutingManager; +import org.apache.pinot.core.routing.RoutingTable; +import org.apache.pinot.core.routing.SegmentsToQuery; +import org.apache.pinot.core.routing.TablePartitionInfo; +import org.apache.pinot.core.routing.TablePartitionReplicatedServersInfo; +import org.apache.pinot.core.routing.timeboundary.TimeBoundaryInfo; +import org.apache.pinot.core.transport.ServerInstance; + +import static com.google.common.base.Preconditions.checkArgument; + + +/// Normal Pinot routing interface over one captured, completely covered OFFLINE table. It neither changes with +/// later metadata updates nor performs I/O, sampling or partition pruning; execution still requires native fences. +final class NativeBrokerRoutingManager implements RoutingManager { + private final String _tableName; + private final Map<String, ServerInstance> _servers; + private final RoutingTable _route; + private final List<String> _segments; + + NativeBrokerRoutingManager(String tableName, Map<ServerInstance, SegmentsToQuery> routes) { + _tableName = tableName; + _servers = Map.copyOf(routes.keySet().stream() + .collect(Collectors.toMap(ServerInstance::getInstanceId, Function.identity()))); + _route = new RoutingTable(Map.copyOf(routes), List.of(), 0); + _segments = routes.values().stream().flatMap(value -> value.getSegments().stream()).sorted().toList(); + } + + @Override + public Map<String, ServerInstance> getEnabledServerInstanceMap() { + return _servers; + } + + @Nullable + @Override + public RoutingTable getRoutingTable(BrokerRequest request, long requestId) { + return getRoutingTable(request, request.getQuerySource().getTableName(), requestId); + } + + @Nullable + @Override + public RoutingTable getRoutingTable(BrokerRequest request, String tableName, long requestId) { + rejectSampling(request, null); + if (!_tableName.equals(tableName)) { + return null; + } + checkArgument(tableName.equals(request.getQuerySource().getTableName()), + "Native routing cannot redirect another table's request"); + return _route; + } + + @Nullable + @Override + public List<String> getSegments(BrokerRequest request) { + return getSegments(request, null); + } + + @Nullable + @Override + public List<String> getSegments(BrokerRequest request, @Nullable String samplerName) { + rejectSampling(request, samplerName); + return _tableName.equals(request.getQuerySource().getTableName()) ? _segments : null; + } + + @Nullable + @Override + public Set<String> getPrunedSegments(BrokerRequest request) { + rejectSampling(request, null); + return _tableName.equals(request.getQuerySource().getTableName()) ? Set.of() : null; + } + + @Override + public boolean routingExists(String tableName) { + return _tableName.equals(tableName); + } + + @Override + public Set<String> getRoutableTables() { + return Set.of(_tableName); + } + + @Nullable + @Override + public TimeBoundaryInfo getTimeBoundaryInfo(String tableName) { + return null; + } + + @Nullable + @Override + public TablePartitionInfo getTablePartitionInfo(String tableName) { + return null; + } + + @Nullable + @Override + public TablePartitionReplicatedServersInfo getTablePartitionReplicatedServersInfo(String tableName) { + return null; + } + + @Nullable + @Override + public Set<String> getServingInstances(String tableName) { + return _tableName.equals(tableName) ? _servers.keySet() : null; + } + + private static void rejectSampling(BrokerRequest request, @Nullable String samplerName) { + checkArgument(samplerName == null && (!request.isSetPinotQuery() + || QueryOptionsUtils.getTableSampler(request.getPinotQuery().getQueryOptions()) == null), + "Native captured routing does not support table sampling"); + } +} diff --git a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandlerTest.java b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandlerTest.java index 6c26207b9ae..dfca00fb70d 100644 --- a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandlerTest.java +++ b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandlerTest.java @@ -31,6 +31,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import javax.annotation.Nullable; import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager; @@ -101,6 +102,8 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertSame; public class BaseSingleStageBrokerRequestHandlerTest { @@ -191,12 +194,16 @@ public class BaseSingleStageBrokerRequestHandlerTest { /// it continue to link. Invokes the 9-arg constructor (no trailing `MaterializedViewHandler` /// parameter) and asserts it instantiates cleanly — and that /// `processMaterializedViewSplitBrokerRequest` is a concrete method on the base class (would - /// fail to compile this anonymous subclass if it were abstract). + /// fail to compile this anonymous subclass if it were abstract). Captured-environment overloads must + /// continue to invoke the existing validation and dispatch overrides for the default backend. @Test - public void testLegacyConstructorIsAvailableForCustomSubclasses() { + public void testLegacyConstructorIsAvailableForCustomSubclasses() + throws Exception { PinotConfiguration config = new PinotConfiguration(); QueryQuotaManager queryQuotaManager = mock(QueryQuotaManager.class); TableCache tableCache = mock(TableCache.class); + AtomicInteger validationCalls = new AtomicInteger(); + BrokerResponseNative response = new BrokerResponseNative(); BaseSingleStageBrokerRequestHandler handler = new BaseSingleStageBrokerRequestHandler(config, "testBrokerId", new BrokerRequestIdGenerator(), @@ -210,15 +217,25 @@ public class BaseSingleStageBrokerRequestHandlerTest { public void shutDown() { } + @Override + protected void validatePhysicalTablesWithMultiClusterRouting(Set<String> tableNames, + Map<String, String> queryOptions) { + validationCalls.incrementAndGet(); + } + @Override protected BrokerResponseNative processBrokerRequest(long requestId, BrokerRequest originalBrokerRequest, BrokerRequest serverBrokerRequest, TableRouteInfo route, long timeoutMs, ServerStats serverStats, RequestContext requestContext) { - return new BrokerResponseNative(); + return response; } }; Assert.assertNotNull(handler); + handler.validatePhysicalTablesWithMultiClusterRouting(Set.of("testTable"), Map.of(), tableCache); + assertEquals(validationCalls.get(), 1); + assertSame(handler.processBrokerRequest(1, null, null, null, 1000, null, null, + new BaseSingleStageBrokerRequestHandler.RequestEnvironment(tableCache, null)), response); } @Test diff --git a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/NativeBrokerRoutingImageTest.java b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/NativeBrokerRoutingImageTest.java index 80a4c46423d..49edf4fac89 100644 --- a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/NativeBrokerRoutingImageTest.java +++ b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/NativeBrokerRoutingImageTest.java @@ -18,21 +18,59 @@ */ package org.apache.pinot.broker.routing; +import com.google.protobuf.ByteString; +import java.io.IOException; +import java.util.HashMap; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.UUID; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import javax.annotation.Nullable; +import javax.ws.rs.core.HttpHeaders; +import org.apache.pinot.broker.broker.AllowAllAccessControlFactory; +import org.apache.pinot.broker.queryquota.QueryQuotaManager; +import org.apache.pinot.broker.requesthandler.BaseSingleStageBrokerRequestHandler; +import org.apache.pinot.broker.requesthandler.BrokerRequestIdGenerator; +import org.apache.pinot.common.config.provider.TableCache; +import org.apache.pinot.common.request.BrokerRequest; +import org.apache.pinot.common.request.PinotQuery; +import org.apache.pinot.common.request.QuerySource; import org.apache.pinot.common.request.context.NativeQueryContext; import org.apache.pinot.common.request.context.NativeQueryDeadline; +import org.apache.pinot.common.response.broker.BrokerResponseNative; +import org.apache.pinot.common.response.broker.ResultTable; +import org.apache.pinot.common.utils.DataSchema; +import org.apache.pinot.common.utils.DataSchema.ColumnDataType; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc; +import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.CatalogRecord; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ParticipantRole; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ParticipantToken; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ReplicaRouting; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.RoutingSnapshot; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.SegmentRouting; +import org.apache.pinot.core.routing.RoutingManager; +import org.apache.pinot.core.routing.TableRouteInfo; +import org.apache.pinot.core.transport.ServerInstance; +import org.apache.pinot.spi.accounting.ThreadAccountantUtils; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.data.FieldSpec.DataType; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.env.PinotConfiguration; +import org.apache.pinot.spi.eventlistener.query.BrokerQueryEventListenerFactory; +import org.apache.pinot.spi.trace.RequestContext; +import org.apache.pinot.spi.utils.JsonUtils; +import org.apache.pinot.spi.utils.builder.TableConfigBuilder; import org.testng.annotations.Test; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNull; import static org.testng.Assert.assertSame; import static org.testng.Assert.assertThrows; import static org.testng.Assert.assertTrue; @@ -82,6 +120,42 @@ public class NativeBrokerRoutingImageTest { assertEquals(image.segmentCount(), 2); assertEquals(image.requests(query, attempt, deadline).getFirst().context().toProto(), requests.getFirst().context().toProto()); + var planning = image.planningMetadata(true); + var otherRequest = image.planningMetadata(true); + assertEquals(planning.tableCache().getActualTableName("ORDERS"), "orders"); + planning.tableCache().getTableConfig("orders_OFFLINE").setTableName("changed"); + planning.tableCache().getSchema("orders").setSchemaName("changed"); + assertEquals(otherRequest.tableCache().getTableConfig("orders_OFFLINE").getTableName(), "orders_OFFLINE"); + assertEquals(otherRequest.tableCache().getSchema("orders").getSchemaName(), "orders"); + assertEquals(image.newTableCache(false).getSchema("orders").getSchemaName(), "orders"); + BrokerRequest compiled = new BrokerRequest(); + QuerySource source = new QuerySource(); + source.setTableName("orders_OFFLINE"); + compiled.setQuerySource(source); + var routing = planning.routingManager(); + var route = routing.getRoutingTable(compiled, 1); + assertTrue(route.getUnavailableSegments().isEmpty()); + assertEquals(routing.getSegments(compiled), List.of("one", "two")); + assertEquals(routing.getServingInstances("orders_OFFLINE"), Set.of("Server_a", "Server_b")); + assertNull(routing.getRoutingTable(compiled, "other_OFFLINE", 1)); + assertFalse(routing.routingExists("orders_REALTIME")); + route.getServerInstanceToSegmentsMap().forEach((server, segments) -> { + var request = requests.stream().filter(value -> value.context().server().participantId() + .equals(server.getInstanceId())).findFirst().orElseThrow(); + assertEquals(server.getHostname(), request.endpoint().getHost()); + assertEquals(server.getGrpcPort(), request.endpoint().getPort()); + assertEquals(server.getPort(), -1); + assertEquals(server.getNettyTlsPort(), -1); + assertNull(server.getAdminEndpoint()); + assertEquals(segments.getSegments(), request.context().requiredNames()); + assertTrue(segments.getOptionalSegments().isEmpty()); + assertThrows(UnsupportedOperationException.class, () -> segments.getSegments().clear()); + }); + assertThrows(UnsupportedOperationException.class, () -> route.getServerInstanceToSegmentsMap().clear()); + assertThrows(UnsupportedOperationException.class, () -> image.serverAuthorities().clear()); + assertThrows(IllegalArgumentException.class, () -> routing.getSegments(compiled, "sample")); + assertThrows(IllegalArgumentException.class, () -> ServerInstance.forNativeGrpc("Server_a", "SERVER-A", 8090)); + assertThrows(IllegalArgumentException.class, () -> ServerInstance.forNativeGrpc("Server_a", "server-a", -1)); } @Test @@ -94,19 +168,22 @@ public class NativeBrokerRoutingImageTest { assertEquals(image.segmentCount(), 2, "Missing replicas must remain represented in complete table metadata"); assertThrows(IllegalStateException.class, () -> image.requests(UUID.randomUUID(), UUID.randomUUID(), new NativeQueryDeadline("test-clock", 100, 150))); + assertThrows(IllegalStateException.class, () -> image.planningMetadata(false)); ReplicaRouting unapproved = fixture.replica("Server_b", "grpcs://server-b:8090").toBuilder() .clearEndpoint().build(); var notEligible = new NativeBrokerRoutingImage(fixture.snapshot(fixture.segment("not-eligible", unapproved)), fixture._broker, "orders_OFFLINE"); assertThrows(IllegalStateException.class, () -> notEligible.requests(UUID.randomUUID(), UUID.randomUUID(), new NativeQueryDeadline("test-clock", 100, 150))); + assertThrows(IllegalStateException.class, () -> notEligible.newRoutingManager()); var empty = new NativeBrokerRoutingImage(fixture.snapshot(), fixture._broker, "orders_OFFLINE"); assertThrows(IllegalStateException.class, () -> empty.requests(UUID.randomUUID(), UUID.randomUUID(), new NativeQueryDeadline("test-clock", 100, 150))); + assertThrows(IllegalStateException.class, () -> empty.newRoutingManager()); } @Test - public void testCrossServerIdentityCollisionsAndEndpointConfusionAreRejected() { + public void testCrossServerIdentityCollisionsAndEndpointConfusionAreRejected() throws Exception { Fixture fixture = new Fixture(); ReplicaRouting first = fixture.replica("Server_a", "grpcs://server-a:8090"); ReplicaRouting second = fixture.replica("Server_b", "grpcs://server-b:8090"); @@ -129,6 +206,156 @@ public class NativeBrokerRoutingImageTest { fixture._broker, "orders_OFFLINE")); } + @Test + public void testPlanningRequiresExactRecordsAndCompleteCurrentServerAuthorities() throws Exception { + Fixture fixture = new Fixture(); + RoutingSnapshot snapshot = fixture.snapshot(fixture.segment("one", + fixture.replica("Server_a", "grpcs://server-a:8090"))); + var table = snapshot.getTable(); + var authority = table.getServerAuthorities(0); + NativeBrokerRoutingImage image = new NativeBrokerRoutingImage(snapshot, fixture._broker, "orders_OFFLINE"); + assertTrue(image.hasPlanningMetadata()); + assertEquals(image.control(), snapshot.getControl()); + var renewed = snapshot.toBuilder().setRevision(snapshot.getRevision().toBuilder().setIndex(11)) + .setTable(table.toBuilder().setServerAuthorities(0, authority.toBuilder() + .setLastRenewalSequence(1).setExpiryRound(4).setLastModifiedIndex(11))).build(); + assertEquals(new NativeBrokerRoutingImage(renewed, fixture._broker, "orders_OFFLINE").queryIdentity(), + image.queryIdentity(), "Lease renewal must not change captured query identity"); + for (var invalid : List.of( + table.toBuilder().clearTableConfig().build(), + table.toBuilder().setTableConfig(table.getTableConfig().toBuilder().setVersion(2)).build(), + table.toBuilder().setTableConfig(table.getTableConfig().toBuilder() + .setObjectId(UUID.randomUUID().toString())).build(), + table.toBuilder().setSchemaRecord(table.getSchemaRecord().toBuilder().setLastModifiedIndex(11)).build(), + table.toBuilder().setSchemaRecord(table.getSchemaRecord().toBuilder().setName("other")).build(), + table.toBuilder().clearServerAuthorities().build(), + table.toBuilder().addServerAuthorities(authority).build(), + table.toBuilder().setServerAuthorities(0, authority.toBuilder().setExpiryRound(0)).build(), + table.toBuilder().setServerAuthorities(0, authority.toBuilder().setReady(false)).build(), + table.toBuilder().setServerAuthorities(0, authority.toBuilder().setDraining(true)).build(), + table.toBuilder().setServerAuthorities(0, authority.toBuilder().clearCapabilities()).build(), + table.toBuilder().setServerAuthorities(0, authority.toBuilder().setParticipant( + authority.getParticipant().toBuilder().setIncarnation(UUID.randomUUID().toString()))).build(), + table.toBuilder().setServerAuthorities(0, authority.toBuilder() + .putEndpoints("queryGrpc", "grpcs://other:8090")).build())) { + assertThrows(IllegalArgumentException.class, () -> new NativeBrokerRoutingImage( + snapshot.toBuilder().setTable(invalid).build(), fixture._broker, "orders_OFFLINE")); + } + assertThrows(IOException.class, () -> new NativeBrokerRoutingImage(snapshot.toBuilder() + .setTable(table.toBuilder().setSchemaRecord(table.getSchemaRecord().toBuilder() + .setValue(MetadataRpc.CatalogValue.newBuilder().setSchemaJson(ByteString.copyFromUtf8("[]"))))).build(), + fixture._broker, "orders_OFFLINE")); + assertThrows(IOException.class, () -> new NativeBrokerRoutingImage(snapshot.toBuilder() + .setTable(table.toBuilder().setTableConfig(table.getTableConfig().toBuilder() + .setKind(MetadataRpc.RecordKind.SCHEMA))).build(), fixture._broker, "orders_OFFLINE")); + for (var invalid : List.of( + snapshot.getControl().toBuilder().setMinimumTickDelayMillis(99).build(), + snapshot.getControl().toBuilder().setMaxConfiguredPermitLifetimeMillis(199).build(), + snapshot.getControl().toBuilder().setMaxConfiguredPermitLifetimeMillis(201).build())) { + assertThrows(IllegalArgumentException.class, () -> new NativeBrokerRoutingImage( + snapshot.toBuilder().setControl(invalid).build(), fixture._broker, "orders_OFFLINE")); + } + + NativeBrokerRoutingImage legacy = new NativeBrokerRoutingImage(snapshot.toBuilder().setTable(table.toBuilder() + .clearTableConfig().clearSchemaRecord().clearServerAuthorities()).build(), fixture._broker, "orders_OFFLINE"); + assertFalse(legacy.hasPlanningMetadata(), "Old observation images do not acquire planning authority"); + assertThrows(IllegalStateException.class, () -> legacy.planningMetadata(false)); + } + + @Test + public void testNormalCompilerRetainsCapturedSchemaAndRoutingAcrossRefresh() throws Exception { + Fixture fixture = new Fixture(); + NativeBrokerRoutingImage initial = new NativeBrokerRoutingImage( + fixture.snapshot(fixture.segment("one", fixture.replica("Server_a", "grpcs://server-a:8090"))), + fixture._broker, "orders_OFFLINE"); + var next = fixture.snapshot( + fixture.segment("two", fixture.replica("Server_b", "grpcs://server-b:8090"))).toBuilder(); + Schema changedSchema = new Schema.SchemaBuilder().setSchemaName("orders") + .addSingleValueDimension("newValue", DataType.INT).build(); + next.getRevisionBuilder().setIndex(20); + next.getTableBuilder().setSchema(fixture._schema.toBuilder().setVersion(2)) + .setSchemaRecord(next.getTable().getSchemaRecord().toBuilder().setVersion(2).setLastModifiedIndex(20) + .setValue(MetadataRpc.CatalogValue.newBuilder() + .setSchemaJson(ByteString.copyFrom(JsonUtils.objectToBytes(changedSchema))))); + NativeBrokerRoutingImage replacement = + new NativeBrokerRoutingImage(next.build(), fixture._broker, "orders_OFFLINE"); + AtomicReference<NativeBrokerRoutingImage> current = new AtomicReference<>(initial); + AtomicReference<NativeBrokerRoutingImage.PlanningMetadata> captured = new AtomicReference<>(); + AtomicInteger captures = new AtomicInteger(); + AtomicInteger dispatches = new AtomicInteger(); + PinotConfiguration config = new PinotConfiguration(); + BrokerQueryEventListenerFactory.init(config); + QueryQuotaManager quotas = mock(QueryQuotaManager.class); + when(quotas.acquire(anyString())).thenReturn(true); + when(quotas.acquireDatabase(anyString())).thenReturn(true); + when(quotas.acquireApplication(anyString())).thenReturn(true); + TableCache unusedCache = mock(TableCache.class, invocation -> { + throw new AssertionError("Compilation consulted the global table cache: " + invocation.getMethod()); + }); + RoutingManager unusedRouting = mock(RoutingManager.class, invocation -> { + throw new AssertionError("Compilation consulted the global routing manager: " + invocation.getMethod()); + }); + BaseSingleStageBrokerRequestHandler handler = new BaseSingleStageBrokerRequestHandler(config, "Broker_native", + new BrokerRequestIdGenerator(), unusedRouting, new AllowAllAccessControlFactory(), quotas, unusedCache, + ThreadAccountantUtils.getNoOpAccountant(), null) { + @Override + protected RequestEnvironment captureRequestEnvironment(PinotQuery query, @Nullable HttpHeaders headers) { + captures.incrementAndGet(); + // Simulate publication of a different schema and route immediately after this request captures its image. + var metadata = current.getAndSet(replacement).planningMetadata(true); + captured.set(metadata); + return new RequestEnvironment(metadata.tableCache(), metadata.routingManager()); + } + + @Override + protected boolean canCompileWithMultiStageEngine(String query, String database) { + return false; + } + + @Override + protected BrokerResponseNative processBrokerRequest(long requestId, BrokerRequest original, + BrokerRequest server, TableRouteInfo route, long timeoutMs, ServerStats stats, RequestContext context) { + throw new AssertionError("Captured metadata was lost before dispatch"); + } + + @Override + protected BrokerResponseNative processBrokerRequest(long requestId, BrokerRequest original, + BrokerRequest server, TableRouteInfo route, long timeoutMs, ServerStats stats, RequestContext context, + RequestEnvironment environment) { + assertSame(environment.tableCache(), captured.get().tableCache()); + assertSame(environment.routingManager(), captured.get().routingManager()); + assertEquals(server.getPinotQuery().getDataSource().getTableName(), "orders_OFFLINE"); + assertEquals(route.getOfflineRoutingTable().size(), 1); + var target = route.getOfflineRoutingTable().entrySet().iterator().next(); + boolean first = dispatches.getAndIncrement() == 0; + assertEquals(target.getKey().getInstanceId(), first ? "Server_a" : "Server_b"); + assertEquals(target.getKey().getGrpcPort(), 8090); + assertEquals(target.getValue().getSegments(), List.of(first ? "one" : "two")); + assertTrue(target.getValue().getOptionalSegments().isEmpty()); + BrokerResponseNative response = new BrokerResponseNative(); + response.setResultTable(new ResultTable(new DataSchema(new String[]{"total"}, + new ColumnDataType[]{ColumnDataType.LONG}), List.<Object[]>of(new Object[]{6L}))); + return response; + } + }; + try { + assertTrue(handler.handleRequest("SELECT SUM(VALUE) AS total FROM ORDERS WHERE VALUE > 1") + .getExceptions().isEmpty()); + assertEquals(captures.get(), 1); + assertEquals(dispatches.get(), 1); + assertSame(current.get(), replacement); + assertFalse(handler.handleRequest("SELECT SUM(VALUE) FROM ORDERS").getExceptions().isEmpty(), + "The next request must observe the replacement schema"); + assertEquals(dispatches.get(), 1); + assertTrue(handler.handleRequest("SELECT SUM(NEWVALUE) AS total FROM ORDERS WHERE NEWVALUE > 1") + .getExceptions().isEmpty()); + assertEquals(captures.get(), 3); + assertEquals(dispatches.get(), 2); + } finally { + handler.shutDown(); + } + } + private static final class Fixture { private final String _history = UUID.randomUUID().toString(); private final ParticipantToken _broker = token("Broker_native", ParticipantRole.BROKER); @@ -156,7 +383,7 @@ public class NativeBrokerRoutingImageTest { .addAllReplicas(List.of(replicas)).build(); } - private RoutingSnapshot snapshot(SegmentRouting... segments) { + private RoutingSnapshot snapshot(SegmentRouting... segments) throws IOException { var revision = MetadataRpc.Revision.newBuilder().setHistoryId(_history).setIndex(10); var control = MetadataRpc.MembershipControl.newBuilder().setControlEpoch(MetadataRpc.ControlToken.newBuilder() .setHistoryId(_history).setRaftTerm(1).setBarrierIndex(1).setLeaderId("controller") @@ -164,11 +391,39 @@ public class NativeBrokerRoutingImageTest { .setMaxConfiguredPermitLifetimeMillis(200).setPolicy(MetadataRpc.MembershipPolicy.newBuilder().setVersion(1) .setMinTickIntervalMillis(100).setLeaseRounds(3).setPermitLifetimeMillis(200) .setMaxQueryLifetimeMillis(300).setMaxRevisionLag(100)); + Map<String, MetadataRpc.ParticipantState> authorities = new HashMap<>(); + for (SegmentRouting segment : segments) { + for (ReplicaRouting replica : segment.getReplicasList()) { + var state = MetadataRpc.ParticipantState.newBuilder().setParticipant(replica.getServer()) + .setReady(true).setExpiryRound(3).setLastModifiedIndex(1).setAppliedRevision(revision) + .addCapabilities(MetadataRpc.ServingCapability.SERVER_QUERY); + if (replica.hasEndpoint()) { + state.putEndpoints("queryGrpc", replica.getEndpoint()).putEndpoints("nativeQueryVersion", "2"); + } + authorities.putIfAbsent(replica.getServer().getParticipantId(), state.build()); + } + } + var tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("orders").build(); + var schema = new Schema.SchemaBuilder().setSchemaName("orders") + .addSingleValueDimension("value", DataType.INT).addSingleValueDimension("label", DataType.STRING).build(); return RoutingSnapshot.newBuilder().setStatus(MetadataRpc.ResultStatus.OK).setRevision(revision) .setParticipant(MetadataRpc.ParticipantState.newBuilder().setParticipant(_broker).setExpiryRound(3) .setLastModifiedIndex(1).setAppliedRevision(revision)) .setControl(control).setTable(MetadataRpc.TableRouting.newBuilder().setTableName("orders_OFFLINE") - .setTable(_table).setSchemaName("orders").setSchema(_schema).addAllSegments(List.of(segments))).build(); + .setTable(_table).setSchemaName("orders").setSchema(_schema).addAllSegments(List.of(segments)) + .setTableConfig(record(_table, "orders_OFFLINE", MetadataRpc.RecordKind.TABLE_CONFIG, + MetadataRpc.CatalogValue.newBuilder() + .setTableConfigJson(ByteString.copyFrom(JsonUtils.objectToBytes(tableConfig))).build())) + .setSchemaRecord(record(_schema, "orders", MetadataRpc.RecordKind.SCHEMA, + MetadataRpc.CatalogValue.newBuilder() + .setSchemaJson(ByteString.copyFrom(JsonUtils.objectToBytes(schema))).build())) + .addAllServerAuthorities(authorities.values())).build(); + } + + private CatalogRecord record(MetadataRpc.ObjectVersion identity, String name, MetadataRpc.RecordKind kind, + MetadataRpc.CatalogValue value) { + return CatalogRecord.newBuilder().setKind(kind).setName(name).setObjectId(identity.getObjectId()) + .setVersion(identity.getVersion()).setSchemaVersion(1).setLastModifiedIndex(1).setValue(value).build(); } } } diff --git a/pinot-common/src/main/proto/metadata.proto b/pinot-common/src/main/proto/metadata.proto index 72ab2e971fe..935fdd4d961 100644 --- a/pinot-common/src/main/proto/metadata.proto +++ b/pinot-common/src/main/proto/metadata.proto @@ -57,8 +57,8 @@ service MinionTaskService { rpc Outcome(TaskOutcomeRequest) returns (CommandResult); } -// BROKER-only complete routing observations. TLS policy selects the stable identity; committed policy authorizes -// its table tenants. Responses contain identities and installation evidence, never catalog records or storage URIs. +// BROKER-only complete planning observations. TLS policy selects the stable identity; committed policy authorizes +// its table tenants. Only the requested table/schema and installed candidates accompany routing evidence. service BrokerRoutingService { rpc Snapshot(RoutingSnapshotRequest) returns (RoutingSnapshot); } @@ -88,6 +88,11 @@ message TableRouting { string schema_name = 3; ObjectVersion schema = 4; repeated SegmentRouting segments = 5; + // Exact current planning records at the same revision as the routes. Absent in legacy observations. + CatalogRecord table_config = 6; + CatalogRecord schema_record = 7; + // One exact current membership state per distinct installed replica candidate, ordered by stable server ID. + repeated ParticipantState server_authorities = 8; } message SegmentRouting { diff --git a/pinot-coordination-ratis/CONTROLLER-API-PARITY.md b/pinot-coordination-ratis/CONTROLLER-API-PARITY.md index cff6c43d95a..aa803bd39b9 100644 --- a/pinot-coordination-ratis/CONTROLLER-API-PARITY.md +++ b/pinot-coordination-ratis/CONTROLLER-API-PARITY.md @@ -38,7 +38,8 @@ The standard resource package has **264 annotated HTTP methods across 43 resourc This does not provide replacement/deletion, general rebalance, physical cleanup or query-serving qualification. - Native broker routing: authenticated, tenant-scoped table discovery and complete OFFLINE routing observations, immutable broker images and query-identity construction. Typed broker certificate/session/tenant grants now support - local server admission and guarded gRPC delivery. Normal query planning, dispatch and serving activation remain + local server admission and guarded gRPC delivery. Coherent table/schema planning inputs and a captured environment + in the shared single-stage compiler are available. Normal broker admission, dispatch and serving activation remain outstanding; installed replicas without query authority are not selectable. - Outstanding: remaining offline lifecycle, full existing task API/plugin parity, realtime, remaining feature families and query-serving qualification. A full replacement is not yet implemented. diff --git a/pinot-coordination-ratis/README.md b/pinot-coordination-ratis/README.md index 3acd9d9044a..4f33741bfb2 100644 --- a/pinot-coordination-ratis/README.md +++ b/pinot-coordination-ratis/README.md @@ -392,11 +392,12 @@ There are no wildcard or implicit policy grants. Normal omitted table tenants re convention. Policy changes take effect through subsequent committed reads; participant advertisements cannot grant tenant access or destination approval. -Each table image carries exact table/schema versions and every published immutable segment, including segments -with zero installed replicas. Installed candidates require current configuration and assignment identities, a live +Each table image carries exact table/schema records and versions and every published immutable segment, including +segments with zero installed replicas. Installed candidates require current configuration and assignment identities, a live ready server incarnation, placement-policy eligibility and matching installation evidence from the durable reconciliation operation. A completed proof does not expire merely because its former controller owner expires. -Archive locations, loader configuration, task configuration and unrelated cluster metadata are not exposed. +The full authorized table configuration and schema are included, preserving their committed record identities. +Segment archive locations, standalone task records and unrelated cluster metadata are not exposed. A candidate includes a query destination only if its server also has `SERVER_QUERY`, advertises the exact approved `queryGrpc` address and declares `nativeQueryVersion=2`. This is routing metadata, not a serving grant or an @@ -414,6 +415,28 @@ unsupported; no fallback can silently turn required segments into optional ones. These observations and contexts do not authorize queries. Connecting the normal query handlers requires the remaining serving-permit, routing-revision, transport-identity, clock and response-delivery checks together. +## Captured broker planning inputs + +The table configuration, schema, segment identities, installed candidates and deduplicated candidate membership +states are captured under the same catalog lock and consensus read barrier. They share the existing whole-image +record and byte limits; an oversized configuration cannot produce a truncated routing image. Discovery still +exposes only table names, and this projection does not grant access to generic metadata APIs. + +`NativeBrokerRoutingImage` verifies the supplied planning records against their kind, name, incarnation, version +and snapshot revision. Identity-only legacy observations remain readable but cannot provide planning inputs. +Each planning capture decodes its own table/schema objects and derives Pinot's normal `TableCache` maps, keeping +compiler mutations separate from the immutable protobuf image. Its `RoutingManager` adapter uses the same +deterministic, complete replica selection as the native identity envelopes without constructing Helix records. + +The normal single-stage handler now captures a request environment before metadata-dependent compilation and +passes its table cache and routing manager through optimization, route construction and dispatch. The default +environment uses the existing providers. Native callers can supply one captured image without replacing global +cache state or copying the SQL compiler. + +This is planning infrastructure, not a native query listener. Normal broker admission/deadline origination, +authenticated dispatch, receiving and final-response fences, quotas and qualified clock integration remain +required before activation. Unsupported native query forms must be rejected explicitly by that integration. + ## Authenticated native query admission The optional `principals` object in `pinot.coordination.ratis.broker.routing` maps lowercase SHA-256 leaf-certificate diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MetadataCatalog.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MetadataCatalog.java index 12444ee666c..04d7baf7835 100644 --- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MetadataCatalog.java +++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MetadataCatalog.java @@ -476,8 +476,11 @@ public final class MetadataCatalog { require(schema != null && !schema.deleted(), Status.NOT_READY, "Routing schema is absent"); TableRouting tableHeader = new TableRouting(tableName, routingVersion(table), schemaName, routingVersion(schema), List.of()); - projection.add(tableHeader, 2); + projection.add(tableHeader, 0); + projection.addRecord(table, _recordBytes.get(table.objectId())); + projection.addRecord(schema, _recordBytes.get(schema.objectId())); List<SegmentRouting> segments = new ArrayList<>(); + Map<String, ParticipantState> serverAuthorities = new HashMap<>(); String contentPrefix = table.objectId() + "/"; String serverTenant = NativeReplicaPlacementPolicy.tenant(config); List<String> approvedServers = placement.servers().keySet().stream().sorted().toList(); @@ -502,6 +505,11 @@ public final class MetadataCatalog { } ReplicaRouting replica = routingReplica(view, policy, content, record, table, schema, serverId); if (replica != null) { + if (!serverAuthorities.containsKey(serverId)) { + ParticipantState authority = view.participant(serverId); + projection.add(authority, 1); + serverAuthorities.put(serverId, authority); + } projection.add(replica, 1); replicas.add(replica); } @@ -510,7 +518,9 @@ public final class MetadataCatalog { } segments.sort(Comparator.comparing(SegmentRouting::name)); return new RoutingSnapshot(Status.OK, header.revision(), "", participant, control, List.of(), - new TableRouting(tableName, tableHeader.table(), schemaName, tableHeader.schema(), segments)); + new TableRouting(tableName, tableHeader.table(), schemaName, tableHeader.schema(), segments, table, schema, + serverAuthorities.values().stream() + .sorted(Comparator.comparing(authority -> authority.participant().participantId())).toList())); } catch (Rejection e) { return routingSnapshotError(e._status, e.getMessage()); } catch (CoordinationRejection e) { @@ -615,8 +625,8 @@ public final class MetadataCatalog { assignment.materializationId(), report.installationId(), policy.endpoint(server)); } - /// Only bounded identity DTOs are encoded while holding apply. Config and archive payloads never enter this image; - /// reject before retaining each additional element and let transport separately check the final protobuf size. + /// Config/schema use maintained canonical sizes so large planning records are not serialized under apply. + /// Reject before retaining each element; transport separately checks the complete protobuf byte and record bounds. private static final class RoutingProjection { private final int _maxRecords; private final int _maxBytes; @@ -637,6 +647,14 @@ public final class MetadataCatalog { checkBounds(); } + private void addRecord(CatalogRecord record, int bytes) { + require(record.kind() == RecordKind.TABLE_CONFIG || record.kind() == RecordKind.SCHEMA, Status.NOT_READY, + "Routing planning record has an unexpected kind"); + _records++; + _bytes += bytes + 1L; + checkBounds(); + } + private void checkBounds() { require(_records <= _maxRecords && _bytes <= _maxBytes, Status.RESPONSE_TOO_LARGE, "Routing snapshot exceeds its record or byte bound"); diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineRoutingProtocol.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineRoutingProtocol.java index 69fe8bb2657..61a243bc599 100644 --- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineRoutingProtocol.java +++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineRoutingProtocol.java @@ -24,6 +24,7 @@ import javax.annotation.Nullable; import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ControlState; import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantState; import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantToken; +import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.CatalogRecord; import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ObjectVersion; import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Revision; import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status; @@ -31,7 +32,8 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status; /// Complete bounded OFFLINE routing observations, never serving grants. Every current segment is represented, /// including segments without an installed replica. A null endpoint means installed evidence is not selectable. -/// Config payloads, archive locations and other participants' private metadata are deliberately not exposed. +/// Planning includes only the authorized table's exact config/schema and the membership of its installed candidates. +/// Archive locations, cluster configuration and unrelated participants are deliberately not exposed. public final class OfflineRoutingProtocol { public static final int MAX_SNAPSHOT_RECORDS = 4096; public static final int MIN_SNAPSHOT_BYTES = 1024; @@ -52,9 +54,17 @@ public final class OfflineRoutingProtocol { } public record TableRouting(String tableName, ObjectVersion table, String schemaName, ObjectVersion schema, - List<SegmentRouting> segments) { + List<SegmentRouting> segments, @Nullable CatalogRecord tableConfig, + @Nullable CatalogRecord schemaRecord, List<ParticipantState> serverAuthorities) { + /// Legacy observations lack planning inputs; consumers must not use them for native query compilation. + public TableRouting(String tableName, ObjectVersion table, String schemaName, ObjectVersion schema, + List<SegmentRouting> segments) { + this(tableName, table, schemaName, schema, segments, null, null, List.of()); + } + public TableRouting { segments = List.copyOf(segments); + serverAuthorities = List.copyOf(serverAuthorities); } } diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingGrpcService.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingGrpcService.java index b61cb979a17..9365b884387 100644 --- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingGrpcService.java +++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingGrpcService.java @@ -23,12 +23,21 @@ import io.grpc.Status; import io.grpc.StatusRuntimeException; import io.grpc.stub.StreamObserver; import java.io.IOException; +import java.util.HashSet; import java.util.Objects; +import java.util.Set; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; +import javax.annotation.Nullable; +import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantToken; import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Role; import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol; +import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.CatalogRecord; +import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ObjectVersion; +import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RecordKind; +import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Revision; import org.apache.pinot.coordination.ratis.metadata.OfflineRoutingProtocol; +import org.apache.pinot.coordination.ratis.metadata.OfflineRoutingProtocol.RoutingSnapshot; import org.apache.pinot.coordination.ratis.rpc.BrokerRoutingServiceGrpc; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc; import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy.ParticipantIdentity; @@ -79,11 +88,13 @@ final class RoutingGrpcService extends BrokerRoutingServiceGrpc.BrokerRoutingSer || !snapshot.tables().isEmpty()) { throw Status.INTERNAL.withDescription("Failed routing projection contains data").asRuntimeException(); } + long records = snapshot.status() == MetadataProtocol.Status.OK ? validatePlanning(snapshot) : 0; MetadataRpc.RoutingSnapshot response = RoutingRpcCodec.snapshot(snapshot); if (response.getSerializedSize() > request.getMaxBytes() - || response.getSerializedSize() > OfflineRoutingProtocol.MAX_SNAPSHOT_BYTES) { + || response.getSerializedSize() > OfflineRoutingProtocol.MAX_SNAPSHOT_BYTES + || records > request.getMaxRecords()) { response = MetadataRpc.RoutingSnapshot.newBuilder().setStatus(MetadataRpc.ResultStatus.RESPONSE_TOO_LARGE) - .setRevision(response.getRevision()).setDetail("Routing projection exceeds its wire byte bound").build(); + .setRevision(response.getRevision()).setDetail("Routing projection exceeds its wire bounds").build(); } remainingMillis(); if (!identity.equals(authorize())) { @@ -113,6 +124,53 @@ final class RoutingGrpcService extends BrokerRoutingServiceGrpc.BrokerRoutingSer } } + /// Planning extends only the requested table image, never the generic catalog authorization scope. + private static long validatePlanning(RoutingSnapshot snapshot) { + var table = snapshot.table(); + if (table == null) { + return 2L + snapshot.tables().size(); + } + if (!snapshot.tables().isEmpty() + || (table.tableConfig() == null) != (table.schemaRecord() == null) + || !matches(table.tableConfig(), RecordKind.TABLE_CONFIG, table.tableName(), table.table(), snapshot.revision()) + || !matches(table.schemaRecord(), RecordKind.SCHEMA, table.schemaName(), table.schema(), snapshot.revision())) { + throw Status.INTERNAL.withDescription("Routing planning records differ from their identities") + .asRuntimeException(); + } + long records = 4L + table.segments().size() + table.serverAuthorities().size(); + Set<ParticipantToken> candidates = new HashSet<>(); + for (var segment : table.segments()) { + records += segment.replicas().size(); + segment.replicas().forEach(replica -> candidates.add(replica.server())); + } + Set<String> servers = new HashSet<>(); + for (var authority : table.serverAuthorities()) { + var token = authority.participant(); + if (token.role() != Role.SERVER || !token.historyId().equals(snapshot.revision().historyId()) + || !servers.add(token.participantId()) || !candidates.remove(token) + || authority.retired() || authority.draining() || !authority.ready() + || authority.expiryRound() <= snapshot.control().round() + || authority.lastModifiedIndex() <= 0 || authority.lastModifiedIndex() > snapshot.revision().index()) { + throw Status.INTERNAL.withDescription("Routing server authority differs from its candidates") + .asRuntimeException(); + } + } + if (table.tableConfig() != null && !candidates.isEmpty()) { + throw Status.INTERNAL.withDescription("Routing planning lacks candidate membership").asRuntimeException(); + } + return records; + } + + private static boolean matches(@Nullable CatalogRecord record, RecordKind kind, String name, ObjectVersion version, + Revision revision) { + // Older identity-only observations remain representable; a planning consumer must reject absent values. + return record == null || record.kind() == kind && record.name().equals(name) && !record.deleted() + && record.schemaVersion() == 1 && record.value() != null && record.objectId().equals(version.objectId()) + && record.version() > 0 && record.version() == version.version() + && version.historyId().equals(revision.historyId()) && record.lastModifiedIndex() > 0 + && record.lastModifiedIndex() <= revision.index(); + } + private ParticipantIdentity authorize() { String fingerprint = MetadataGrpcServer.PEER_FINGERPRINT.get(); ParticipantIdentity identity = fingerprint == null ? null : _policy.authorize(fingerprint); diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingRpcCodec.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingRpcCodec.java index eb5d4169d7d..fdf19158386 100644 --- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingRpcCodec.java +++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/RoutingRpcCodec.java @@ -18,6 +18,7 @@ */ package org.apache.pinot.coordination.ratis.transport; +import java.io.IOException; import java.util.UUID; import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantToken; import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Role; @@ -29,7 +30,7 @@ import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy.Par import static com.google.common.base.Preconditions.checkArgument; -/// Closed, identity-only broker routing vocabulary. The authenticated certificate supplies the stable broker ID; +/// Closed, table-scoped broker planning vocabulary. The authenticated certificate supplies the stable broker ID; /// the request can select only its session and one table or authorized discovery. final class RoutingRpcCodec { private RoutingRpcCodec() { @@ -48,7 +49,7 @@ final class RoutingRpcCodec { UUID.fromString(request.getSession().getIncarnation()), request.getSession().getRegistrationEpoch()); } - static MetadataRpc.RoutingSnapshot snapshot(RoutingSnapshot snapshot) { + static MetadataRpc.RoutingSnapshot snapshot(RoutingSnapshot snapshot) throws IOException { var response = MetadataRpc.RoutingSnapshot.newBuilder() .setStatus(MetadataRpc.ResultStatus.valueOf(snapshot.status().name())).setDetail(snapshot.detail()) .setRevision(MetadataRpc.Revision.newBuilder().setHistoryId(snapshot.revision().historyId().toString()) @@ -64,6 +65,15 @@ final class RoutingRpcCodec { var encodedTable = MetadataRpc.TableRouting.newBuilder().setTableName(table.tableName()) .setTable(OfflineSegmentRpcCodec.version(table.table())).setSchemaName(table.schemaName()) .setSchema(OfflineSegmentRpcCodec.version(table.schema())); + if (table.tableConfig() != null) { + encodedTable.setTableConfig(MetadataRpcCodec.record(table.tableConfig())); + } + if (table.schemaRecord() != null) { + encodedTable.setSchemaRecord(MetadataRpcCodec.record(table.schemaRecord())); + } + for (var server : table.serverAuthorities()) { + encodedTable.addServerAuthorities(MembershipRpcCodec.participant(server)); + } for (var segment : table.segments()) { var encodedSegment = MetadataRpc.SegmentRouting.newBuilder().setName(segment.name()) .setContent(OfflineSegmentRpcCodec.version(segment.content())).setContentSha256(segment.contentSha256()); diff --git a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalogTest.java b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalogTest.java index baa75e62242..0f6dae96c76 100644 --- a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalogTest.java +++ b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalogTest.java @@ -728,6 +728,10 @@ public class TaskCatalogTest { Fixture f = new Fixture(); ParticipantToken broker = f.routingBroker(0); f.routingPolicies("DefaultTenant", "grpcs://localhost:8090", true); + assertEquals(f.mutate(Mutation.updateTableConfig(f.objectVersion(f.record(RecordKind.TABLE_CONFIG, TABLE)), + new TableConfigBuilder(TableType.OFFLINE).setTableName("orders") + .setCustomConfig(new TableCustomConfig(Map.of("planning", "x".repeat(5000)))).build())).status(), + Status.OK); UUID hiddenTable = UUID.randomUUID(); assertEquals(f.mutate(Mutation.createTableConfig(f._history, hiddenTable, new TableConfigBuilder(TableType.OFFLINE).setTableName("private").setBrokerTenant("PrivateTenant").build())) @@ -741,6 +745,11 @@ public class TaskCatalogTest { assertEquals(f.routing(broker, TABLE).table().segments().getFirst().replicas().size(), 0, "Desired assignment is not installation evidence"); f.install(f.assignments(installed).getFirst()); + Content second = f.publishRoutingContent("installed-second"); + ReconcileContent secondReconcile = new ReconcileContent(f._owner, second.segmentObjectId()); + assertEquals(f.segment(secondReconcile).status(), Status.OK); + assertEquals(f.segment(secondReconcile).status(), Status.OK); + f.install(f.assignments(second).getFirst(), "second-install"); RoutingSnapshot discovery = f.routing(broker, ""); assertEquals(discovery.status(), Status.OK); assertEquals(discovery.tables(), List.of(TABLE)); @@ -748,12 +757,24 @@ public class TaskCatalogTest { assertTrue(discovery.participant().capabilities().isEmpty()); RoutingSnapshot image = f.routing(broker, TABLE); assertEquals(image.status(), Status.OK); - assertEquals(image.table().segments().size(), 2, "Uninstalled published content must remain explicit"); - assertEquals(image.table().segments().get(1).name(), "pending"); - assertTrue(image.table().segments().get(1).replicas().isEmpty()); + assertEquals(image.revision(), new Revision(f._history, f._index)); + assertEquals(image.control(), f._catalog.membershipControl()); + assertEquals(image.table().tableConfig(), f.record(RecordKind.TABLE_CONFIG, TABLE)); + assertEquals(image.table().schemaRecord(), f.record(RecordKind.SCHEMA, "orders")); + assertEquals(image.table().table(), f.objectVersion(image.table().tableConfig())); + assertEquals(image.table().schema(), f.objectVersion(image.table().schemaRecord())); + assertTrue(image.table().tableConfig().lastModifiedIndex() <= image.revision().index()); + assertTrue(image.table().schemaRecord().lastModifiedIndex() <= image.revision().index()); + assertEquals(image.table().segments().size(), 3, "Uninstalled published content must remain explicit"); + assertEquals(image.table().segments().get(2).name(), "pending"); + assertTrue(image.table().segments().get(2).replicas().isEmpty()); var replica = image.table().segments().getFirst().replicas().getFirst(); assertEquals(replica.server(), server); assertEquals(replica.endpoint(), null, "Installed SERVER without SERVER_QUERY is not selectable"); + assertEquals(image.table().serverAuthorities(), List.of(MembershipProtocol.decodeParticipant( + f.record(RecordKind.PARTICIPANT, server.participantId()).value()))); + assertEquals(f._catalog.routingSnapshot(broker, TABLE, 10, OfflineRoutingProtocol.MAX_SNAPSHOT_BYTES), image, + "Multiple installed segments share one bounded server authority"); String json = new String(MetadataProtocol.encodeValue(image), StandardCharsets.UTF_8); assertFalse(json.contains("private-archive")); assertFalse(json.contains("never-expose")); @@ -765,8 +786,8 @@ public class TaskCatalogTest { assertEquals(f.routing(broker, "orders_OFFLINE_OFFLINE").status(), Status.INVALID_REQUEST); assertEquals(f.routing(broker, ".orders_OFFLINE").status(), Status.INVALID_REQUEST); for (RoutingSnapshot rejected : List.of( - f._catalog.routingSnapshot(broker, TABLE, 6, OfflineRoutingProtocol.MAX_SNAPSHOT_BYTES), - f._catalog.routingSnapshot(broker, TABLE, 4096, 1024), + f._catalog.routingSnapshot(broker, TABLE, 9, OfflineRoutingProtocol.MAX_SNAPSHOT_BYTES), + f._catalog.routingSnapshot(broker, TABLE, 4096, 5000), f._catalog.routingSnapshot(broker, "", 2, OfflineRoutingProtocol.MAX_SNAPSHOT_BYTES))) { assertEquals(rejected.status(), Status.RESPONSE_TOO_LARGE); assertEquals(rejected.table(), null); @@ -842,6 +863,13 @@ public class TaskCatalogTest { .addSingleValueDimension("newField", DataType.STRING).build())).status(), Status.OK); assertTrue(f.routing(broker, TABLE).table().segments().getFirst().replicas().isEmpty(), "A proof against a previous exact schema version is not a current candidate"); + RoutingSnapshot changed = f.routing(broker, TABLE); + assertEquals(changed.revision(), new Revision(f._history, f._index)); + assertEquals(changed.table().schemaRecord(), f.record(RecordKind.SCHEMA, "orders")); + assertEquals(changed.table().schema().version(), schema.version() + 1); + assertTrue(changed.table().schemaRecord().value().toString().contains("newField")); + assertTrue(changed.table().serverAuthorities().isEmpty(), + "An obsolete installation must not leave unreferenced server authority"); for (String policy : List.of("{}", "{\"brokers\":{},\"servers\":{},\"extra\":{}}", "{\"brokers\":{\"routing-broker\":[\"*\"]},\"servers\":{}}", "{\"brokers\":{\"routing-broker\":[\"DefaultTenant\",\"DefaultTenant\"]},\"servers\":{}}", @@ -1136,6 +1164,10 @@ public class TaskCatalogTest { } private void install(Assignment assignment) throws IOException { + install(assignment, "install"); + } + + private void install(Assignment assignment, String streamId) throws IOException { UUID operation = NativeReplicaPlacementPolicy.operationId(_history, assignment.materializationId()); var work = OfflineInstallationProtocol.decodeWorkResult(_catalog.query(MetadataProtocol.encode( new GetInstallationWork(_history, assignment.server(), operation)))); @@ -1147,7 +1179,7 @@ public class TaskCatalogTest { assertEquals(apply(new InstallationSessionCommand(assignment.server(), new RegisterClient(_history, client, assignment.server().incarnation(), 0))).status(), Status.OK); assertEquals(apply(new InstallationSessionCommand(assignment.server(), new NativeInstallationCommand(_history, - client, 1, "install", 1, report))).status(), Status.OK); + client, 1, streamId, 1, report))).status(), Status.OK); } private Result submit() throws IOException { diff --git a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcTransportTest.java b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcTransportTest.java index dc25c3bb0f9..d998aa33315 100644 --- a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcTransportTest.java +++ b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcTransportTest.java @@ -1625,12 +1625,13 @@ public class MetadataGrpcTransportTest { assertStatus(Status.Code.PERMISSION_DENIED, () -> worker.openSnapshot(fixture.open())); assertEquals(quorum._access.execute(new MetadataProtocol.RegisterClient(history, "native-test", UUID.randomUUID(), 0), 5000).status(), MetadataProtocol.Status.OK); + TableConfig plannedTable = new TableConfigBuilder(TableType.OFFLINE).setTableName("routes") + .setNumReplicas(1).build(); + Schema plannedSchema = new Schema.SchemaBuilder().setSchemaName("routes") + .addSingleValueDimension("id", DataType.INT).build(); commitNativeTest(quorum, new MetadataProtocol.Transaction(history, "native-test", 1, "config", 1, - List.of(MetadataProtocol.Mutation.createTableConfig(history, UUID.randomUUID(), - new TableConfigBuilder(TableType.OFFLINE).setTableName("routes").setNumReplicas(1).build()), - MetadataProtocol.Mutation.createSchema(history, UUID.randomUUID(), - new Schema.SchemaBuilder().setSchemaName("routes").addSingleValueDimension("id", DataType.INT) - .build()), + List.of(MetadataProtocol.Mutation.createTableConfig(history, UUID.randomUUID(), plannedTable), + MetadataProtocol.Mutation.createSchema(history, UUID.randomUUID(), plannedSchema), MetadataProtocol.Mutation.createClusterConfig(history, UUID.randomUUID(), Map.of( NativeBrokerRoutingPolicy.CONFIG_KEY, "{\"brokers\":{\"Broker_worker\":[\"DefaultTenant\"]},\"servers\":{}}", @@ -1651,6 +1652,50 @@ public class MetadataGrpcTransportTest { assertTrue(table.getTable().hasSchema()); assertEquals(table.getTable().getTableName(), "routes_OFFLINE"); assertEquals(table.getTable().getSegmentsCount(), 0); + MetadataRpc.CatalogRecord tableConfig = table.getTable().getTableConfig(); + MetadataRpc.CatalogRecord schemaRecord = table.getTable().getSchemaRecord(); + assertEquals(tableConfig.getKind(), MetadataRpc.RecordKind.TABLE_CONFIG); + assertEquals(tableConfig.getName(), "routes_OFFLINE"); + assertEquals(tableConfig.getObjectId(), table.getTable().getTable().getObjectId()); + assertEquals(tableConfig.getVersion(), table.getTable().getTable().getVersion()); + // The catalog canonicalizes numeric JSON (e.g. 0.0 to 0); preserve the committed value exactly on the wire. + var committedTable = quorum._access.query(new MetadataProtocol.GetRecord(history, + MetadataProtocol.RecordKind.TABLE_CONFIG, "routes_OFFLINE"), 5000).record(); + assertEquals(tableConfig.getValue().getTableConfigJson().toByteArray(), + JsonUtils.objectToBytes(committedTable.value())); + assertEquals(schemaRecord.getKind(), MetadataRpc.RecordKind.SCHEMA); + assertEquals(schemaRecord.getName(), "routes"); + assertEquals(schemaRecord.getObjectId(), table.getTable().getSchema().getObjectId()); + assertEquals(schemaRecord.getVersion(), table.getTable().getSchema().getVersion()); + var committedSchema = quorum._access.query(new MetadataProtocol.GetRecord(history, + MetadataProtocol.RecordKind.SCHEMA, "routes"), 5000).record(); + assertEquals(schemaRecord.getValue().getSchemaJson().toByteArray(), + JsonUtils.objectToBytes(committedSchema.value())); + assertTrue(tableConfig.getLastModifiedIndex() <= table.getRevision().getIndex()); + assertTrue(schemaRecord.getLastModifiedIndex() <= table.getRevision().getIndex()); + assertEquals(tableConfig.getLastModifiedIndex(), schemaRecord.getLastModifiedIndex()); + assertEquals(table.getTable().getServerAuthoritiesCount(), 0); + MetadataRpc.RoutingSnapshot tableBound = worker.routingSnapshot(request.toBuilder() + .setTableName("routes_OFFLINE").setMaxRecords(3).build()); + assertEquals(tableBound.getStatus(), MetadataRpc.ResultStatus.RESPONSE_TOO_LARGE); + assertFalse(tableBound.hasTable(), "A bound failure must not expose partial planning records"); + access._routingResult.set(image -> { + var routed = image.table(); + var record = routed.tableConfig(); + var wrongKind = new MetadataProtocol.CatalogRecord(MetadataProtocol.RecordKind.CLUSTER_CONFIG, + record.name(), record.objectId(), record.version(), record.schemaVersion(), record.lastModifiedIndex(), + false, record.value()); + return new OfflineRoutingProtocol.RoutingSnapshot(image.status(), image.revision(), image.detail(), + image.participant(), image.control(), image.tables(), new OfflineRoutingProtocol.TableRouting( + routed.tableName(), routed.table(), routed.schemaName(), routed.schema(), routed.segments(), + wrongKind, routed.schemaRecord(), routed.serverAuthorities())); + }); + try { + assertStatus(Status.Code.INTERNAL, () -> worker.routingSnapshot( + request.toBuilder().setTableName("routes_OFFLINE").build())); + } finally { + access._routingResult.set(UnaryOperator.identity()); + } assertEquals(other.routingSnapshot(request).getStatus(), MetadataRpc.ResultStatus.FENCED); assertStatus(Status.Code.PERMISSION_DENIED, () -> controller.routingSnapshot(request)); assertEquals(worker.routingSnapshot(request.toBuilder().setHistoryId(UUID.randomUUID().toString()).build()) @@ -2000,6 +2045,8 @@ public class MetadataGrpcTransportTest { private final AtomicReference<Runnable> _beforeRouting = new AtomicReference<>(() -> { }); private final AtomicReference<UnaryOperator<OfflineInstallationProtocol.InstallationSnapshot>> _installationResult = new AtomicReference<>(UnaryOperator.identity()); + private final AtomicReference<UnaryOperator<OfflineRoutingProtocol.RoutingSnapshot>> _routingResult = + new AtomicReference<>(UnaryOperator.identity()); private final CountDownLatch _entered = new CountDownLatch(4); private final CountDownLatch _release = new CountDownLatch(1); private volatile boolean _block; @@ -2034,7 +2081,8 @@ public class MetadataGrpcTransportTest { public OfflineRoutingProtocol.RoutingSnapshot routingSnapshot(MembershipProtocol.ParticipantToken broker, String tableName, int maxRecords, int maxBytes, long timeoutMillis) throws IOException { _beforeRouting.get().run(); - return ((RoutingAccess) _delegate).routingSnapshot(broker, tableName, maxRecords, maxBytes, timeoutMillis); + return _routingResult.get().apply( + ((RoutingAccess) _delegate).routingSnapshot(broker, tableName, maxRecords, maxBytes, timeoutMillis)); } @Override diff --git a/pinot-core/src/main/java/org/apache/pinot/core/transport/ServerInstance.java b/pinot-core/src/main/java/org/apache/pinot/core/transport/ServerInstance.java index b0ca541bf16..7b62bfe421a 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/transport/ServerInstance.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/transport/ServerInstance.java @@ -20,7 +20,10 @@ package org.apache.pinot.core.transport; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; +import java.net.URI; +import java.net.URISyntaxException; import java.util.HashSet; +import java.util.Locale; import java.util.Map; import java.util.Set; import org.apache.commons.lang3.StringUtils; @@ -129,6 +132,35 @@ public final class ServerInstance { _pool = FALLBACK_POOL_ID; } + /// Represents an approved native gRPC destination without constructing Helix instance metadata. + /// Other transports and administrative endpoints remain unavailable; this value grants no query authority. + public static ServerInstance forNativeGrpc(String participantId, String hostname, int grpcPort) { + Preconditions.checkArgument(participantId != null && !participantId.isBlank() && participantId.length() <= 128 + && participantId.chars().noneMatch(Character::isISOControl), "Invalid native server participant ID"); + Preconditions.checkArgument(hostname != null && !hostname.isBlank() + && hostname.equals(hostname.toLowerCase(Locale.ROOT)) && grpcPort > 0 && grpcPort <= 65535, + "Invalid native gRPC host or port"); + try { + URI endpoint = new URI("grpcs", null, hostname, grpcPort, null, null, null); + Preconditions.checkArgument(hostname.equals(endpoint.getHost()), "Noncanonical native gRPC hostname"); + } catch (URISyntaxException e) { + throw new IllegalArgumentException("Invalid native gRPC hostname", e); + } + return new ServerInstance(participantId, hostname, grpcPort); + } + + private ServerInstance(String participantId, String hostname, int grpcPort) { + _instanceId = participantId.intern(); + _hostname = hostname; + _grpcPort = grpcPort; + _port = INVALID_PORT; + _nettyTlsPort = INVALID_PORT; + _queryServicePort = INVALID_PORT; + _queryMailboxPort = INVALID_PORT; + _adminEndpoint = null; + _pool = FALLBACK_POOL_ID; + } + /// Returns the interned instance id. public String getInstanceId() { return _instanceId; diff --git a/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java b/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java index 3ee3b056e6a..d5f4044d4c9 100644 --- a/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java +++ b/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java @@ -500,8 +500,11 @@ public class RatisStartersTest { assertEquals(recovered, reinstalled, "Controller restart must retain the current installation and proof"); assertEquals(awaitInstallation(voterConfig, replacement, uploaded.content()), uploadedAgain, "Controller restart must not create another materialization for an uploaded segment"); + // Heartbeats advance server authority; the planning records and installation proof must remain unchanged. assertEquals(assertRoutingInstallation(brokerClient, brokerState.getParticipant(), uploaded.content(), - replacement, uploadedAgain), recoveredRouting, "Routing must retain the recovered installation proof"); + replacement, uploadedAgain).toBuilder().clearServerAuthorities().build(), + recoveredRouting.toBuilder().clearServerAuthorities().build(), + "Routing must retain the recovered installation proof"); var taskState = JsonUtils.stringToJsonNode(get(http, controllerUrl + "/tasks/native/nativeTasks", true).body()); assertEquals(taskState.path("tasks").size(), 1); assertEquals(taskState.path("publications").size(), 1); @@ -582,6 +585,14 @@ public class RatisStartersTest { assertEquals(replica.getMaterializationId(), installed.assignment().materializationId().toString()); assertEquals(replica.getInstallationId(), installed.installationId().toString()); assertFalse(replica.hasEndpoint(), "An approved address cannot grant a missing server query capability"); + assertEquals(snapshot.getTable().getServerAuthoritiesCount(), 1); + MetadataRpc.ParticipantState authority = snapshot.getTable().getServerAuthorities(0); + assertEquals(authority.getParticipant(), server); + assertTrue(authority.getReady()); + assertFalse(authority.getRetired()); + assertFalse(authority.getDraining()); + assertTrue(authority.getExpiryRound() > snapshot.getControl().getRound()); + assertEquals(authority.getCapabilitiesCount(), 0, "Native query listeners must remain disabled"); return snapshot.getTable(); } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
