This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch xiangfu0/mse-proto-segment-list
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to
refs/heads/xiangfu0/mse-proto-segment-list by this push:
new 06e0b02e21c Make the segment list encoding mode a live cluster config
06e0b02e21c is described below
commit 06e0b02e21c49f04789ba05d35e8547288849150
Author: Xiang Fu <[email protected]>
AuthorDate: Tue Sep 22 16:47:54 2026 -0700
Make the segment list encoding mode a live cluster config
The per-query `protoSegmentList` option was the only way to change the
encoding without restarting a broker, but asking clients to change their
queries is harder than restarting brokers, so it was the wrong escape
hatch. Read the mode from cluster config instead, and drop the option.
- `ProtoSegmentListPredicate` also listens on cluster config for
`pinot.broker.mse.proto.segment.list`. Precedence is cluster config,
then the static broker config, then SAFE; clearing the cluster-config
key restores the static broker config, as MultiStageQueryThrottler does
for its live config. A value that is not a mode is ignored with a
warning rather than moving the cluster off the operator's chosen mode.
- The server versions are watched whatever the static mode is, since
cluster config can select SAFE at runtime.
- Remove the `protoSegmentList` query option, `QueryOptionsUtils
.isProtoSegmentList` and its test. Nothing overrides the mode per query
any more, so all servers of a query always agree on the encoding.
- The integration tests now switch the encoding through cluster config
rather than a query option, which also covers the live reload path: the
logical-table test goes through the controller's /cluster/configs
endpoint and restores SAFE afterwards, since that cluster is shared.
---
.../broker/broker/helix/BaseBrokerStarter.java | 8 ++-
.../common/utils/config/QueryOptionsUtils.java | 5 --
.../common/utils/config/QueryOptionsUtilsTest.java | 8 ---
.../tests/MultiStageEngineIntegrationTest.java | 43 +++++++++---
.../BaseLogicalTableIntegrationTest.java | 39 +++++++++--
.../pinot/query/routing/QueryPlanSerDeUtils.java | 2 +-
.../dispatch/ProtoSegmentListPredicate.java | 77 ++++++++++++++++------
.../query/service/dispatch/QueryDispatcher.java | 15 ++---
.../dispatch/ProtoSegmentListPredicateTest.java | 55 ++++++++++++++--
.../apache/pinot/spi/utils/CommonConstants.java | 20 +++---
10 files changed, 197 insertions(+), 75 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
index afaf1c0645d..0fff6b0e71e 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
@@ -108,6 +108,7 @@ import
org.apache.pinot.materializedview.handler.MaterializedViewHandler;
import org.apache.pinot.query.routing.WorkerManager;
import
org.apache.pinot.query.runtime.operator.factory.DefaultQueryOperatorFactoryProvider;
import
org.apache.pinot.query.runtime.operator.factory.QueryOperatorFactoryProvider;
+import org.apache.pinot.query.service.dispatch.ProtoSegmentListPredicate;
import org.apache.pinot.segment.spi.partition.PartitionFunctionFactory;
import org.apache.pinot.spi.accounting.ThreadAccountant;
import org.apache.pinot.spi.accounting.ThreadAccountantUtils;
@@ -586,8 +587,11 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
MultiStageBrokerRequestHandler finalHandler =
multiStageBrokerRequestHandler;
_routingManager.setServerReenableCallback(
serverInstance ->
finalHandler.getQueryDispatcher().resetClientConnectionBackoff(serverInstance));
- // Watches the server versions that pick the default segment list
encoding; a no-op unless the mode is SAFE.
-
multiStageBrokerRequestHandler.getProtoSegmentListPredicate().watchInstanceConfigs(_spectatorHelixManager);
+ // The segment list encoding follows the server versions in SAFE mode,
and its mode follows cluster config.
+ ProtoSegmentListPredicate protoSegmentListPredicate =
+ multiStageBrokerRequestHandler.getProtoSegmentListPredicate();
+ protoSegmentListPredicate.watchInstanceConfigs(_spectatorHelixManager);
+
_clusterConfigChangeHandler.registerClusterConfigChangeListener(protoSegmentListPredicate);
}
TimeSeriesRequestHandler timeSeriesRequestHandler = null;
if
(StringUtils.isNotBlank(_brokerConf.getProperty(PinotTimeSeriesConfiguration.getEnabledLanguagesConfigKey())))
{
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
index c0266c5963d..f93a4597230 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
@@ -770,11 +770,6 @@ public class QueryOptionsUtils {
return option != null ? Boolean.parseBoolean(option) : defaultValue;
}
- public static boolean isProtoSegmentList(Map<String, String> queryOptions,
boolean defaultValue) {
- String option = queryOptions.get(QueryOptionKey.PROTO_SEGMENT_LIST);
- return option != null ? Boolean.parseBoolean(option) : defaultValue;
- }
-
public static boolean isMultiClusterRoutingEnabled(Map<String, String>
queryOptions, boolean defaultValue) {
String option =
queryOptions.get(QueryOptionKey.ENABLE_MULTI_CLUSTER_ROUTING);
return option != null ? Boolean.parseBoolean(option) : defaultValue;
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/utils/config/QueryOptionsUtilsTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/utils/config/QueryOptionsUtilsTest.java
index 18711317b96..0b8155403aa 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/utils/config/QueryOptionsUtilsTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/utils/config/QueryOptionsUtilsTest.java
@@ -123,14 +123,6 @@ public class QueryOptionsUtilsTest {
QueryOptionsUtils.getInPredicatePruningThreshold(Map.of(IN_PREDICATE_PRUNING_THRESHOLD,
"invalid"));
}
- @Test
- public void testIsProtoSegmentList() {
- assertFalse(QueryOptionsUtils.isProtoSegmentList(Map.of(), false));
- assertTrue(QueryOptionsUtils.isProtoSegmentList(Map.of(), true));
- assertTrue(QueryOptionsUtils.isProtoSegmentList(Map.of(PROTO_SEGMENT_LIST,
"true"), false));
-
assertFalse(QueryOptionsUtils.isProtoSegmentList(Map.of(PROTO_SEGMENT_LIST,
"false"), true));
- }
-
@Test
public void testSkipIndexesParsing() {
String skipIndexesStr = "col1=inverted,range&col2=sorted";
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java
index a2be2c51961..efe24ae4a10 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java
@@ -296,8 +296,9 @@ public class MultiStageEngineIntegrationTest extends
BaseClusterIntegrationTestS
}
/// Every server of this cluster runs the broker's own build, so the default
SAFE mode has to switch the proto
- /// segment list encoding on by itself, through the server versions
published in the Helix instance configs. Both
- /// encodings must then return identical results for queries with one and
with several leaf stages.
+ /// segment list encoding on by itself, through the server versions
published in the Helix instance configs. Setting
+ /// the mode to NEVER in cluster config has to switch it back off without
restarting the broker, and both encodings
+ /// must return identical results for queries with one and with several leaf
stages.
@Test
public void testProtoSegmentListEncodingIsTransparent()
throws Exception {
@@ -310,19 +311,41 @@ public class MultiStageEngineIntegrationTest extends
BaseClusterIntegrationTestS
assertFalse(predicate.isEnabled(true), "Multi-cluster queries must keep
the legacy encoding");
String table = getTableName();
- String[] queries = {
+ List<String> queries = List.of(
"SELECT COUNT(*) FROM " + table,
"SELECT Carrier, COUNT(*), MAX(ArrDelay) FROM " + table + " WHERE
DaysSinceEpoch > 16312 "
+ "GROUP BY Carrier ORDER BY Carrier",
"SELECT COUNT(*) FROM " + table + " a JOIN (SELECT DISTINCT Carrier
FROM " + table + ") b "
- + "ON a.Carrier = b.Carrier"
- };
+ + "ON a.Carrier = b.Carrier");
+
+ Map<String, JsonNode> protoRows = new HashMap<>();
for (String query : queries) {
- JsonNode legacy = postQuery("SET protoSegmentList = false; " + query);
- assertTrue(legacy.get("exceptions").isEmpty(), "Unexpected exceptions
with the legacy encoding: " + legacy);
- JsonNode proto = postQuery("SET protoSegmentList = true; " + query);
- assertTrue(proto.get("exceptions").isEmpty(), "Unexpected exceptions
with the proto encoding: " + proto);
- assertEquals(proto.get("resultTable").get("rows"),
legacy.get("resultTable").get("rows"), query);
+ JsonNode response = postQuery(query);
+ assertTrue(response.get("exceptions").isEmpty(), "Unexpected exceptions
with the proto encoding: " + response);
+ protoRows.put(query, response.get("resultTable").get("rows"));
+ }
+
+ HelixConfigScope scope =
+ new
HelixConfigScopeBuilder(HelixConfigScope.ConfigScopeProperty.CLUSTER).forCluster(getHelixClusterName())
+ .build();
+ try {
+ // The kill switch: a mode set in cluster config reaches the broker
without a restart.
+ _helixManager.getConfigAccessor()
+ .set(scope, CommonConstants.Broker.CONFIG_OF_MSE_PROTO_SEGMENT_LIST,
"NEVER");
+ TestUtils.waitForCondition(aVoid -> !predicate.isEnabled(false), 10_000L,
+ "Setting the mode to NEVER in cluster config did not reach the
broker");
+
+ for (String query : queries) {
+ JsonNode response = postQuery(query);
+ assertTrue(response.get("exceptions").isEmpty(),
+ "Unexpected exceptions with the legacy encoding: " + response);
+ assertEquals(response.get("resultTable").get("rows"),
protoRows.get(query),
+ "The segment list encoding changed the result of: " + query);
+ }
+ } finally {
+ _helixManager.getConfigAccessor().set(scope,
CommonConstants.Broker.CONFIG_OF_MSE_PROTO_SEGMENT_LIST, "SAFE");
+ TestUtils.waitForCondition(aVoid -> predicate.isEnabled(false), 10_000L,
+ "Restoring SAFE in cluster config did not reach the broker");
}
}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/logicaltable/BaseLogicalTableIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/logicaltable/BaseLogicalTableIntegrationTest.java
index d6206c3bd03..1b7f3c5ec88 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/logicaltable/BaseLogicalTableIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/logicaltable/BaseLogicalTableIntegrationTest.java
@@ -29,10 +29,12 @@ import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.commons.io.FileUtils;
+import org.apache.pinot.broker.requesthandler.BrokerRequestHandlerDelegate;
import org.apache.pinot.integration.tests.BaseClusterIntegrationTestSet;
import org.apache.pinot.integration.tests.ClusterIntegrationTestUtils;
import org.apache.pinot.integration.tests.QueryAssert;
import org.apache.pinot.integration.tests.QueryGenerator;
+import org.apache.pinot.query.service.dispatch.ProtoSegmentListPredicate;
import org.apache.pinot.spi.config.table.QueryConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
@@ -41,6 +43,7 @@ import org.apache.pinot.spi.data.PhysicalTableConfig;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.TimeBoundaryConfig;
import org.apache.pinot.spi.exception.QueryErrorCode;
+import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.builder.LogicalTableConfigBuilder;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
@@ -482,20 +485,44 @@ public abstract class BaseLogicalTableIntegrationTest
extends BaseClusterIntegra
/// Both leaf-stage segment list encodings must return the same result for a
logical table, whose leaf workers carry
/// `logicalTableSegmentsMap`, keyed by physical table name, rather than the
table-type keyed `tableSegmentsMap`.
+ /// The encoding is switched through cluster config, the way an operator
would, and restored afterwards because the
+ /// cluster is shared with the other logical-table test classes.
@Test
public void testProtoSegmentListPreservesLogicalTableResults()
throws Exception {
setUseMultiStageQueryEngine(true);
String query = "SELECT Carrier, COUNT(*) FROM " + getLogicalTableName() +
" WHERE DaysSinceEpoch > 16312 "
+ "GROUP BY Carrier ORDER BY Carrier LIMIT 100";
-
- JsonNode legacy = postQuery("SET protoSegmentList = false; " + query);
- assertTrue(legacy.get("exceptions").isEmpty(), "Unexpected exceptions with
the legacy encoding: " + legacy);
- JsonNode proto = postQuery("SET protoSegmentList = true; " + query);
+ // The cluster was started by the shared suite instance, which is the one
holding the broker starter.
+ ProtoSegmentListPredicate predicate =
+ ((BrokerRequestHandlerDelegate)
_sharedClusterTestSuite._brokerStarters.get(0).getBrokerRequestHandler())
+
.getMultiStageBrokerRequestHandler().getProtoSegmentListPredicate();
+ TestUtils.waitForCondition(aVoid -> predicate.isEnabled(false), 10_000L,
+ "SAFE mode did not enable the proto segment list encoding although
every server runs the same version");
+
+ JsonNode proto = postQuery(query);
assertTrue(proto.get("exceptions").isEmpty(), "Unexpected exceptions with
the proto encoding: " + proto);
- assertEquals(proto.get("resultTable").get("rows"),
legacy.get("resultTable").get("rows"),
- "The segment list encoding changed the result of a logical table
query");
+ try {
+ setProtoSegmentListMode("NEVER");
+ TestUtils.waitForCondition(aVoid -> !predicate.isEnabled(false), 10_000L,
+ "Setting the mode to NEVER in cluster config did not reach the
broker");
+
+ JsonNode legacy = postQuery(query);
+ assertTrue(legacy.get("exceptions").isEmpty(), "Unexpected exceptions
with the legacy encoding: " + legacy);
+ assertEquals(legacy.get("resultTable").get("rows"),
proto.get("resultTable").get("rows"),
+ "The segment list encoding changed the result of a logical table
query");
+ } finally {
+ setProtoSegmentListMode("SAFE");
+ TestUtils.waitForCondition(aVoid -> predicate.isEnabled(false), 10_000L,
+ "Restoring SAFE in cluster config did not reach the broker");
+ }
+ }
+
+ private void setProtoSegmentListMode(String mode)
+ throws Exception {
+ sendPostRequest(_controllerRequestURLBuilder.forClusterConfigs(),
+
JsonUtils.objectToString(Map.of(CommonConstants.Broker.CONFIG_OF_MSE_PROTO_SEGMENT_LIST,
mode)));
}
@Test
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/query/routing/QueryPlanSerDeUtils.java
b/pinot-query-planner/src/main/java/org/apache/pinot/query/routing/QueryPlanSerDeUtils.java
index 303d3ca4084..5aaea21f4f7 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/query/routing/QueryPlanSerDeUtils.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/query/routing/QueryPlanSerDeUtils.java
@@ -48,7 +48,7 @@ import org.apache.pinot.spi.utils.JsonUtils;
/// fields understand.
///
/// Decoding accepts both, so a server always understands every broker; the
broker enables the proto encoding only
-/// when every server does (see the `protoSegmentList` query option).
+/// when every server does (see `ProtoSegmentListPredicate`).
public class QueryPlanSerDeUtils {
private static final TypeReference<Map<String, List<String>>>
SEGMENTS_MAP_TYPE = new TypeReference<>() {
};
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/ProtoSegmentListPredicate.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/ProtoSegmentListPredicate.java
index da2df6fbf86..9b56d47c612 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/ProtoSegmentListPredicate.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/ProtoSegmentListPredicate.java
@@ -23,6 +23,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
@@ -35,6 +36,7 @@ import org.apache.helix.api.listeners.PreFetch;
import org.apache.helix.model.InstanceConfig;
import org.apache.pinot.common.version.PinotVersion;
import org.apache.pinot.spi.config.instance.InstanceType;
+import org.apache.pinot.spi.config.provider.PinotClusterConfigChangeListener;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.InstanceTypeUtils;
@@ -42,9 +44,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/// Decides whether a multi-stage query that does not set the
-/// [CommonConstants.Broker.Request.QueryOptionKey#PROTO_SEGMENT_LIST] query
option ships its leaf-stage segment lists
-/// as native protobuf fields of the worker metadata, or as the legacy JSON
custom property.
+/// Decides whether a multi-stage query ships its leaf-stage segment lists as
native protobuf fields of the worker
+/// metadata, or as the legacy JSON custom property.
///
/// Only servers decode those fields, and a server that predates them finds no
segments, treats the worker as an
/// intermediate-stage worker and fails the leaf stage. The mode, set by
@@ -62,21 +63,27 @@ import org.slf4j.LoggerFactory;
/// under a different version string, and tests.
/// - [Mode#NEVER]: legacy JSON unconditionally. The kill switch.
///
-/// The per-query option overrides every mode. Modeled on
[org.apache.pinot.query.runtime.SendStatsPredicate], which
-/// gates the MSE stats on the same signal, with two deliberate differences:
only servers are checked, since brokers
-/// never decode the fields, and an unreadable instance config counts as
outdated rather than current, since the
-/// cost of a wrong answer here is a failed query rather than missing stats.
+/// The mode is read from cluster config, falling back to the static broker
config and then to
+/// [CommonConstants.Broker#DEFAULT_MSE_PROTO_SEGMENT_LIST]. Cluster config
wins and is applied to the next query, so
+/// an operator can switch the encoding off without restarting the brokers;
clearing the key restores the static
+/// broker config. A value that is not a mode is ignored with a warning,
leaving the current mode in place.
+///
+/// Modeled on [org.apache.pinot.query.runtime.SendStatsPredicate], which
gates the MSE stats on the same signal, with
+/// two deliberate differences: only servers are checked, since brokers never
decode the fields, and an unreadable
+/// instance config counts as outdated rather than current, since the cost of
a wrong answer here is a failed query
+/// rather than missing stats.
///
/// Every instance that is not a controller, broker or minion counts as a
server, so an instance config left behind
/// by a decommissioned old server keeps [Mode#SAFE] on the legacy encoding
until it is removed; the outdated servers
/// are logged whenever the encoding switches.
///
-/// Thread-safety: [#isEnabled] reads a single `volatile` flag and is
lock-free on the request path. Instance-config
-/// deliveries are serialized on `this`, which also publishes the Helix
handles set by [#watchInstanceConfigs].
+/// Thread-safety: [#isEnabled] reads two `volatile` fields and is lock-free
on the request path. Instance-config and
+/// cluster-config deliveries are serialized on `this`, which also publishes
the Helix handles set by
+/// [#watchInstanceConfigs].
@ThreadSafe
@BatchMode(enabled = false)
@PreFetch(enabled = false)
-public class ProtoSegmentListPredicate implements InstanceConfigChangeListener
{
+public class ProtoSegmentListPredicate implements
InstanceConfigChangeListener, PinotClusterConfigChangeListener {
private static final Logger LOGGER =
LoggerFactory.getLogger(ProtoSegmentListPredicate.class);
private static final String KEY =
CommonConstants.Broker.CONFIG_OF_MSE_PROTO_SEGMENT_LIST;
/// Cap on the outdated servers named in one log line, so that a large
rolling upgrade does not log the whole fleet.
@@ -86,7 +93,11 @@ public class ProtoSegmentListPredicate implements
InstanceConfigChangeListener {
NEVER, SAFE, ALWAYS
}
- private final Mode _mode;
+ /// The mode of the static broker config, used until cluster config says
otherwise and again when the cluster-config
+ /// key is cleared.
+ private final Mode _staticMode;
+ /// The mode in force, which cluster config can change at runtime.
+ private volatile Mode _mode;
private final String _currentVersion;
/// SAFE only: servers whose version is not [#_currentVersion], mapped to
the version they report. Guarded by `this`.
private final Map<String, String> _outdatedServers = new HashMap<>();
@@ -104,6 +115,7 @@ public class ProtoSegmentListPredicate implements
InstanceConfigChangeListener {
@VisibleForTesting
ProtoSegmentListPredicate(Mode mode, String currentVersion) {
+ _staticMode = mode;
_mode = mode;
_currentVersion = currentVersion;
}
@@ -126,8 +138,8 @@ public class ProtoSegmentListPredicate implements
InstanceConfigChangeListener {
return _mode;
}
- /// Whether a query that does not set the query option uses the proto
encoding. `multiClusterQuery` is whether the
- /// query routes to other clusters, whose server versions this predicate
cannot see.
+ /// Whether a query uses the proto encoding. `multiClusterQuery` is whether
the query routes to other clusters,
+ /// whose server versions this predicate cannot see.
public boolean isEnabled(boolean multiClusterQuery) {
switch (_mode) {
case ALWAYS:
@@ -139,13 +151,11 @@ public class ProtoSegmentListPredicate implements
InstanceConfigChangeListener {
}
}
- /// Starts watching the server versions of the cluster when the mode needs
it, and is a no-op otherwise.
- /// `helixManager` must already be connected. A registration failure is
logged rather than thrown: it leaves
- /// [Mode#SAFE] on the legacy encoding, which is always correct, and an
optimization must not fail broker startup.
+ /// Starts watching the server versions of the cluster. Registered whatever
the current mode is, because cluster
+ /// config can switch the mode to [Mode#SAFE] at runtime. `helixManager`
must already be connected. A registration
+ /// failure is logged rather than thrown: it leaves [Mode#SAFE] on the
legacy encoding, which is always correct, and
+ /// an optimization must not fail broker startup.
public void watchInstanceConfigs(HelixManager helixManager) {
- if (_mode != Mode.SAFE) {
- return;
- }
try {
// Published under the monitor that deliveries synchronize on, but
registered outside it: Helix may deliver the
// initial notification on another thread while registration is still in
progress.
@@ -160,6 +170,35 @@ public class ProtoSegmentListPredicate implements
InstanceConfigChangeListener {
}
}
+ /// Applies a mode set in cluster config, which wins over the static broker
config and takes effect on the next
+ /// query. Clearing the key restores the static broker config.
+ @Override
+ public synchronized void onChange(Set<String> changedConfigs, Map<String,
String> clusterConfigs) {
+ if (!changedConfigs.contains(KEY)) {
+ return;
+ }
+ String value = clusterConfigs.get(KEY);
+ Mode mode;
+ if (value == null || value.isBlank()) {
+ mode = _staticMode;
+ } else {
+ try {
+ mode = Mode.valueOf(value.trim().toUpperCase(Locale.ENGLISH));
+ } catch (IllegalArgumentException e) {
+ // Keep the mode in force: a typo must not silently move a cluster off
the encoding an operator chose.
+ LOGGER.warn("Ignoring invalid value '{}' for {}, expected one of
NEVER, SAFE, ALWAYS, staying on: {}", value,
+ KEY, _mode);
+ return;
+ }
+ }
+ Mode previous = _mode;
+ if (mode == previous) {
+ return;
+ }
+ _mode = mode;
+ LOGGER.info("Updated {} from: {} to: {}", KEY, previous, mode);
+ }
+
@Override
public synchronized void onInstanceConfigChange(List<InstanceConfig>
instanceConfigs, NotificationContext context) {
NotificationContext.Type type = context.getType();
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
index d0b88515214..8653dae86a9 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
@@ -133,9 +133,9 @@ public class QueryDispatcher {
/// Cluster-level default for stream-stats mode. Used as the fallback in
[#submitAndReduce] when the query
/// does not carry an explicit [QueryOptionKey#STREAM_STATS] override.
private final boolean _streamStatsDefault;
- /// Picks the leaf-stage segment list encoding of a query that does not
carry an explicit
- /// [QueryOptionKey#PROTO_SEGMENT_LIST] override. Read once per request,
since in its default mode it follows the
- /// server versions of the cluster; see [ProtoSegmentListPredicate].
+ /// Picks the leaf-stage segment list encoding of a query. Read once per
request, since operators can change the
+ /// mode through cluster config and its default mode follows the server
versions of the cluster; see
+ /// [ProtoSegmentListPredicate].
private final ProtoSegmentListPredicate _protoSegmentList;
public QueryDispatcher(MailboxService mailboxService, FailureDetector
failureDetector, @Nullable TlsConfig tlsConfig,
@@ -717,12 +717,11 @@ public class QueryDispatcher {
}
}
- /// Whether this query ships its leaf-stage segment lists in the proto
encoding: the query option if set, otherwise
- /// the predicate, which has to use the legacy encoding for a multi-cluster
query because it cannot see the server
- /// versions of the other clusters. Resolved once per query so that all of
its servers get the same encoding.
+ /// Whether this query ships its leaf-stage segment lists in the proto
encoding. A multi-cluster query keeps the
+ /// legacy encoding, because the predicate cannot see the server versions of
the other clusters. Resolved once per
+ /// query so that all of its servers get the same encoding even if the mode
changes mid-dispatch.
private boolean useProtoSegmentList(Map<String, String> queryOptions) {
- return QueryOptionsUtils.isProtoSegmentList(queryOptions,
-
_protoSegmentList.isEnabled(QueryOptionsUtils.isMultiClusterRoutingEnabled(queryOptions,
false)));
+ return
_protoSegmentList.isEnabled(QueryOptionsUtils.isMultiClusterRoutingEnabled(queryOptions,
false));
}
/// Builds the request for one server: the plans of the stages it takes part
in, with only its own workers'
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/service/dispatch/ProtoSegmentListPredicateTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/service/dispatch/ProtoSegmentListPredicateTest.java
index f111360d33e..1edf806bbca 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/service/dispatch/ProtoSegmentListPredicateTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/service/dispatch/ProtoSegmentListPredicateTest.java
@@ -21,6 +21,7 @@ package org.apache.pinot.query.service.dispatch;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixManager;
import org.apache.helix.NotificationContext;
@@ -187,13 +188,59 @@ public class ProtoSegmentListPredicateTest {
assertFalse(predicate.isEnabled(false));
}
+ /// The kill switch: a mode set in cluster config wins over the static
broker config and takes effect without a
+ /// broker restart.
@Test
- public void testWatchIsANoOpOutsideSafeMode()
+ public void testClusterConfigOverridesTheStaticMode() {
+ ProtoSegmentListPredicate predicate = new
ProtoSegmentListPredicate(Mode.SAFE, CURRENT);
+ predicate.refreshAllServers(Map.of(SERVER_1, CURRENT));
+ assertTrue(predicate.isEnabled(false));
+
+ predicate.onChange(Set.of(KEY), Map.of(KEY, "NEVER"));
+ assertEquals(predicate.getMode(), Mode.NEVER);
+ assertFalse(predicate.isEnabled(false), "NEVER from cluster config must
switch the encoding off");
+
+ predicate.onChange(Set.of(KEY), Map.of(KEY, "always"));
+ assertEquals(predicate.getMode(), Mode.ALWAYS);
+ assertTrue(predicate.isEnabled(true), "ALWAYS from cluster config must
apply even to multi-cluster queries");
+ }
+
+ /// Clearing the cluster-config key restores the static broker config, as
the other live broker configs do.
+ @Test
+ public void testClearingClusterConfigRestoresTheStaticMode() {
+ ProtoSegmentListPredicate predicate = new
ProtoSegmentListPredicate(Mode.ALWAYS, CURRENT);
+ predicate.onChange(Set.of(KEY), Map.of(KEY, "NEVER"));
+ assertEquals(predicate.getMode(), Mode.NEVER);
+
+ predicate.onChange(Set.of(KEY), Map.of());
+ assertEquals(predicate.getMode(), Mode.ALWAYS);
+ assertTrue(predicate.isEnabled(false));
+ }
+
+ /// A typo must not silently move the cluster off the encoding an operator
chose, so the mode in force stays.
+ @Test
+ public void testInvalidClusterConfigValueIsIgnored() {
+ ProtoSegmentListPredicate predicate = new
ProtoSegmentListPredicate(Mode.ALWAYS, CURRENT);
+ predicate.onChange(Set.of(KEY), Map.of(KEY, "true"));
+ assertEquals(predicate.getMode(), Mode.ALWAYS);
+ assertTrue(predicate.isEnabled(false));
+ }
+
+ @Test
+ public void testClusterConfigChangeThatDoesNotTouchTheKeyIsIgnored() {
+ ProtoSegmentListPredicate predicate = new
ProtoSegmentListPredicate(Mode.ALWAYS, CURRENT);
+ predicate.onChange(Set.of("some.other.key"), Map.of(KEY, "NEVER"));
+ assertEquals(predicate.getMode(), Mode.ALWAYS);
+ }
+
+ /// SAFE can be selected at runtime, so the server versions are watched
whatever the static mode is.
+ @Test
+ public void testWatchesServerVersionsWhateverTheStaticModeIs()
throws Exception {
HelixManager helixManager = helixManager(mock(HelixAdmin.class));
- new ProtoSegmentListPredicate(Mode.ALWAYS,
CURRENT).watchInstanceConfigs(helixManager);
- new ProtoSegmentListPredicate(Mode.NEVER,
CURRENT).watchInstanceConfigs(helixManager);
- verify(helixManager, never()).addInstanceConfigChangeListener(any());
+ ProtoSegmentListPredicate predicate = new
ProtoSegmentListPredicate(Mode.NEVER, CURRENT);
+ predicate.watchInstanceConfigs(helixManager);
+ verify(helixManager).addInstanceConfigChangeListener(predicate);
}
/// A failed registration must not fail broker startup; it just leaves SAFE
on the legacy encoding.
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index 25f6d5757cb..5a6c3b88f4f 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -668,15 +668,19 @@ public class CommonConstants {
public static final String CONFIG_OF_STREAM_STATS_DRAIN_MS =
"pinot.broker.mse.stream.stats.drain.ms";
public static final long DEFAULT_STREAM_STATS_DRAIN_MS = 50L;
- /// How a multi-stage query that does not set the
[Request.QueryOptionKey#PROTO_SEGMENT_LIST] query option ships
- /// its leaf-stage segment lists: as native protobuf fields of the worker
metadata, which skips a JSON encode per
- /// leaf-stage worker on the broker and a JSON parse per worker on the
server, or as the legacy JSON string custom
- /// property. A server that predates the proto fields fails the leaf
stages it receives in the proto encoding.
+ /// How a multi-stage query ships its leaf-stage segment lists: as native
protobuf fields of the worker metadata,
+ /// which skips a JSON encode per leaf-stage worker on the broker and a
JSON parse per worker on the server, or as
+ /// the legacy JSON string custom property. A server that predates the
proto fields fails the leaf stages it
+ /// receives in the proto encoding.
/// - `SAFE` (default): proto only while every server of the cluster
reports this broker's Pinot version, so the
/// encoding switches itself on when a rolling upgrade completes;
multi-cluster queries always use the legacy
/// encoding.
/// - `ALWAYS`: proto unconditionally.
/// - `NEVER`: legacy unconditionally.
+ ///
+ /// Also read from cluster config under the same key, which wins over the
static broker config and takes effect on
+ /// the next query, so the encoding can be switched off without restarting
the brokers; clearing the cluster-config
+ /// key restores the static broker config.
public static final String CONFIG_OF_MSE_PROTO_SEGMENT_LIST =
"pinot.broker.mse.proto.segment.list";
public static final String DEFAULT_MSE_PROTO_SEGMENT_LIST = "SAFE";
@@ -953,14 +957,6 @@ public class CommonConstants {
/// other transport error) during dispatch, the broker cancels the
query and surfaces the error to the
/// client.
public static final String STREAM_STATS = "streamStats";
- /// When set, overrides `pinot.broker.mse.proto.segment.list` for one
query: `true` ships the leaf-stage
- /// segment lists as native protobuf fields of the worker metadata,
`false` as the legacy JSON string custom
- /// property.
- ///
- /// **Mixed-version note.** `true` requires every server the query
reaches to understand the proto encoding: an
- /// older server fails the leaf stage. The broker default (`SAFE`)
already picks the proto encoding whenever
- /// that holds, so this option is mostly useful to force the legacy
encoding.
- public static final String PROTO_SEGMENT_LIST = "protoSegmentList";
/// If set, changes the explain behavior in multi-stage engine.
///
/// `true` means to ask servers for the physical plan while false
means to just use logical plan.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
