You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2026/10/20 10:41:07 UTC
(pinot) branch xiangfu0/codex2/ratis-broker-planning created (now 05ee934d651)
This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a change to branch xiangfu0/codex2/ratis-broker-planning
in repository https://gitbox.apache.org/repos/asf/pinot.git
at 05ee934d651 Capture coherent native broker planning inputs
This branch includes the following new commits:
new 05ee934d651 Capture coherent native broker planning inputs
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
(pinot) 01/01: Capture coherent native broker planning inputs
Posted by xi...@apache.org.
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 <xi...@apache.org>
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: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org