This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new ce2dc49cb0f Add SQL query option policies: legacy OPTION() cluster
config and per-request sqlOptionsMode (#19570)
ce2dc49cb0f is described below
commit ce2dc49cb0fbe0148b387c1abec9fb87ad8eb1d5
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Wed Sep 16 17:54:04 2026 -0700
Add SQL query option policies: legacy OPTION() cluster config and
per-request sqlOptionsMode (#19570)
- Cluster config `pinot.query.legacy.option.syntax.mode` (`ALLOW` default /
`IGNORE` / `REJECT`) controls the
legacy PQL-style `OPTION(k=v)` suffix; applied at broker and controller
startup
- Query option `sqlOptionsMode` (`ALLOW` default / `IGNORE` / `REJECT`),
honored only from the request payload,
controls SQL-embedded `SET` / `OPTION(...)` options so a gateway can
guarantee its request options win
- Controller `/sql` now parses once with the exact payload it forwards, so
engine and database routing use the
same options the broker applies (also aligns `database` precedence with
the broker: SQL over request)
- Broker parse-catch sites and the `IN_SUBQUERY` parse and execution paths
preserve `QueryException` codes and
messages instead of forcing `SQL_PARSING` / `QUERY_EXECUTION`
- Add `QueryErrorCode.fromThrowable(Throwable, QueryErrorCode)` and use it
wherever a site only needs the code
Co-authored-by: Claude Fable 5.1 <[email protected]>
---
.../broker/api/resources/PinotClientRequest.java | 3 +-
.../broker/broker/helix/BaseBrokerStarter.java | 8 +-
.../apache/pinot/broker/grpc/BrokerGrpcServer.java | 3 +-
.../requesthandler/BaseBrokerRequestHandler.java | 13 +-
.../BaseSingleStageBrokerRequestHandler.java | 14 ++-
.../BrokerRequestHandlerDelegate.java | 5 +-
.../BaseSingleStageBrokerRequestHandlerTest.java | 61 +++++++++
.../utils/config/QueryOptionConfigListener.java | 78 ++++++++++++
.../common/utils/config/QueryOptionsUtils.java | 47 ++++++-
.../pinot/common/utils/request/RequestUtils.java | 32 +++--
.../apache/pinot/sql/parsers/CalciteSqlParser.java | 15 ++-
.../config/QueryOptionConfigListenerTest.java | 71 +++++++++++
.../pinot/sql/parsers/SqlOptionsModeTest.java | 136 +++++++++++++++++++++
.../api/resources/PinotQueryResource.java | 54 +++-----
.../api/resources/PinotQueryResourceTest.java | 40 ++++++
.../core/operator/combine/BaseCombineOperator.java | 8 +-
.../streaming/BaseStreamingCombineOperator.java | 9 +-
.../runtime/executor/OpChainSchedulerService.java | 21 ++--
.../apache/pinot/spi/exception/QueryErrorCode.java | 6 +
.../apache/pinot/spi/utils/CommonConstants.java | 23 +++-
20 files changed, 546 insertions(+), 101 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/api/resources/PinotClientRequest.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/api/resources/PinotClientRequest.java
index 729055c9c4e..7b681e815d5 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/api/resources/PinotClientRequest.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/api/resources/PinotClientRequest.java
@@ -667,7 +667,8 @@ public class PinotClientRequest {
try {
sqlNodeAndOptions =
RequestUtils.parseQuery(sqlRequestJson.get(Request.SQL).asText(),
sqlRequestJson);
} catch (Exception e) {
- return new BrokerResponseNative(QueryErrorCode.SQL_PARSING,
e.getMessage());
+ QueryErrorCode errorCode = QueryErrorCode.fromThrowable(e,
QueryErrorCode.SQL_PARSING);
+ return new BrokerResponseNative(errorCode, e.getMessage());
}
if (forceUseMultiStage) {
sqlNodeAndOptions.setExtraOptions(Map.of(Request.QueryOptionKey.USE_MULTISTAGE_ENGINE,
"true"));
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 ebdad1eaf06..a4080f8d960 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
@@ -86,6 +86,7 @@ import org.apache.pinot.common.metrics.MseMetrics;
import org.apache.pinot.common.utils.PinotAppConfigs;
import org.apache.pinot.common.utils.ServiceStartableUtils;
import org.apache.pinot.common.utils.ServiceStatus;
+import org.apache.pinot.common.utils.config.QueryOptionConfigListener;
import org.apache.pinot.common.utils.config.QueryWorkloadConfigUtils;
import org.apache.pinot.common.utils.config.TagNameUtils;
import org.apache.pinot.common.utils.helix.HelixHelper;
@@ -672,6 +673,10 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
LOGGER.info("Wiring up cluster config change handler with helix");
_spectatorHelixManager.addClusterfigChangeListener(_clusterConfigChangeHandler);
+ // Registered before the query endpoints start so that the cluster configs
are applied before the first query
+
_clusterConfigChangeHandler.registerClusterConfigChangeListener(ContinuousJfrStarter.INSTANCE);
+
_clusterConfigChangeHandler.registerClusterConfigChangeListener(_serverRoutingStatsManager);
+ _clusterConfigChangeHandler.registerClusterConfigChangeListener(new
QueryOptionConfigListener());
LOGGER.info("Starting broker admin application on: {}",
ListenerConfigUtil.toString(_listenerConfigs));
_brokerAdminApplication = createBrokerAdminApp();
@@ -742,9 +747,6 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
_brokerMetrics.addTimedValue(BrokerTimer.STARTUP_SUCCESS_DURATION_MS,
System.currentTimeMillis() - startTimeMs, TimeUnit.MILLISECONDS);
-
_clusterConfigChangeHandler.registerClusterConfigChangeListener(ContinuousJfrStarter.INSTANCE);
-
_clusterConfigChangeHandler.registerClusterConfigChangeListener(_serverRoutingStatsManager);
-
NettyInspector.registerMetrics(_brokerMetrics);
LOGGER.info("Finish starting Pinot broker");
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/grpc/BrokerGrpcServer.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/grpc/BrokerGrpcServer.java
index 6d6f4fce789..da8e696c489 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/grpc/BrokerGrpcServer.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/grpc/BrokerGrpcServer.java
@@ -228,10 +228,11 @@ public class BrokerGrpcServer extends
PinotQueryBrokerGrpc.PinotQueryBrokerImplB
try {
sqlNodeAndOptions = RequestUtils.parseQuery(query, requestJsonNode);
} catch (Exception e) {
+ QueryErrorCode errorCode = QueryErrorCode.fromThrowable(e,
QueryErrorCode.SQL_PARSING);
BrokerResponse brokerResponse;
Broker.BrokerResponse errorBlock;
try {
- brokerResponse = new BrokerResponseNative(QueryErrorCode.SQL_PARSING,
e.getMessage());
+ brokerResponse = new BrokerResponseNative(errorCode, e.getMessage());
errorBlock =
Broker.BrokerResponse.newBuilder().setPayload(ByteString.copyFrom(
brokerResponse.toJsonString().getBytes())).build();
} catch (IOException ex) {
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 742cf545d02..fcc86726946 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
@@ -24,7 +24,6 @@ import com.google.common.collect.Maps;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.Locale;
import java.util.Map;
import java.util.OptionalLong;
import java.util.Set;
@@ -50,8 +49,6 @@ import org.apache.pinot.common.metrics.BrokerQueryPhase;
import org.apache.pinot.common.response.BrokerResponse;
import org.apache.pinot.common.response.broker.BrokerResponseNative;
import org.apache.pinot.common.response.broker.QueryProcessingException;
-import org.apache.pinot.common.utils.config.QueryOptionsUtils;
-import
org.apache.pinot.common.utils.config.QueryOptionsUtils.SqlQueryOptionValidationMode;
import org.apache.pinot.common.utils.request.RequestUtils;
import org.apache.pinot.core.auth.Actions;
import org.apache.pinot.core.auth.TargetType;
@@ -137,11 +134,6 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
Broker.DEFAULT_BROKER_ENABLE_QUERY_CANCELLATION);
_enableAutoRewriteAggregationType =
config.getProperty(Broker.CONFIG_OF_BROKER_QUERY_ENABLE_AUTO_REWRITE_AGGREGATION_TYPE);
- // Process-wide static because the SQL parser has no access to the broker
config. Set here rather
- // than parsed per query; a broker restart is needed to pick up a config
change.
-
QueryOptionsUtils.setSqlQueryOptionValidationMode(SqlQueryOptionValidationMode.valueOf(
-
config.getProperty(Broker.CONFIG_OF_BROKER_QUERY_OPTION_VALIDATION_MODE,
-
Broker.DEFAULT_BROKER_QUERY_OPTION_VALIDATION_MODE).trim().toUpperCase(Locale.ROOT)));
if (_enableQueryCancellation) {
_queriesById = new ConcurrentHashMap<>();
_clientQueryIds = new ConcurrentHashMap<>();
@@ -197,8 +189,9 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
sqlNodeAndOptions = RequestUtils.parseQuery(query, request);
} catch (Exception e) {
// Do not log or emit metric here because it is pure user error
- requestContext.setErrorCode(QueryErrorCode.SQL_PARSING);
- return new BrokerResponseNative(QueryErrorCode.SQL_PARSING,
e.getMessage());
+ QueryErrorCode errorCode = QueryErrorCode.fromThrowable(e,
QueryErrorCode.SQL_PARSING);
+ requestContext.setErrorCode(errorCode);
+ return new BrokerResponseNative(errorCode, e.getMessage());
}
}
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 a099861f951..d93ffc1ff70 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
@@ -1183,9 +1183,9 @@ public abstract class BaseSingleStageBrokerRequestHandler
extends BaseBrokerRequ
} catch (Exception e) {
LOGGER.info("Caught exception while handling the subquery in request {}:
{}, {}", requestId,
_queryLogger.redactQuery(query,
requestContext.getQueryFingerprint()), e.getMessage());
- requestContext.setErrorCode(QueryErrorCode.QUERY_EXECUTION);
- return new CompileResult(
- new BrokerResponseNative(QueryErrorCode.QUERY_EXECUTION,
e.getMessage()));
+ QueryErrorCode errorCode = QueryErrorCode.fromThrowable(e,
QueryErrorCode.QUERY_EXECUTION);
+ requestContext.setErrorCode(errorCode);
+ return new CompileResult(new BrokerResponseNative(errorCode,
e.getMessage()));
}
boolean ignoreCase = _tableCache.isIgnoreCase();
@@ -1640,8 +1640,8 @@ public abstract class BaseSingleStageBrokerRequestHandler
extends BaseBrokerRequ
sqlNodeAndOptions = RequestUtils.parseQuery(subquery, jsonRequest);
} catch (Exception e) {
// Do not log or emit metric here because it is pure user error
- requestContext.setErrorCode(QueryErrorCode.SQL_PARSING);
- throw new RuntimeException("Failed to parse subquery: " + subquery, e);
+ QueryErrorCode errorCode = QueryErrorCode.fromThrowable(e,
QueryErrorCode.SQL_PARSING);
+ throw new QueryException(errorCode, "Failed to parse subquery: " +
e.getMessage(), e);
}
// Add null handling option from broker config only if there is no
override in the query
@@ -1665,7 +1665,9 @@ public abstract class BaseSingleStageBrokerRequestHandler
extends BaseBrokerRequ
doHandleRequest(requestId, subquery, sqlNodeAndOptions, jsonRequest,
requesterIdentity, requestContext,
httpHeaders, accessControl, false);
if (response.getExceptionsSize() != 0) {
- throw new RuntimeException("Caught exception while executing subquery:
" + subquery);
+ QueryProcessingException exception = response.getExceptions().get(0);
+ throw new
QueryException(QueryErrorCode.fromErrorCode(exception.getErrorCode()),
+ "Caught exception while executing subquery: " + subquery + ": " +
exception.getMessage());
}
String serializedIdSet = (String)
response.getResultTable().getRows().get(0)[0];
function.setOperator(IN_ID_SET);
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BrokerRequestHandlerDelegate.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BrokerRequestHandlerDelegate.java
index ee0586f2815..07a987ae801 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BrokerRequestHandlerDelegate.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BrokerRequestHandlerDelegate.java
@@ -119,8 +119,9 @@ public class BrokerRequestHandlerDelegate implements
BrokerRequestHandler {
sqlNodeAndOptions =
RequestUtils.parseQuery(request.get(Request.SQL).asText(), request);
} catch (Exception e) {
// Do not log or emit metric here because it is pure user error
- requestContext.setErrorCode(QueryErrorCode.SQL_PARSING);
- return new BrokerResponseNative(QueryErrorCode.SQL_PARSING,
e.getMessage());
+ QueryErrorCode errorCode = QueryErrorCode.fromThrowable(e,
QueryErrorCode.SQL_PARSING);
+ requestContext.setErrorCode(errorCode);
+ return new BrokerResponseNative(errorCode, e.getMessage());
}
}
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 ae661c0107a..6c26207b9ae 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
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.broker.requesthandler;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import com.sun.net.httpserver.HttpServer;
import java.net.InetSocketAddress;
import java.util.HashMap;
@@ -31,6 +32,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
+import javax.annotation.Nullable;
import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
import org.apache.helix.model.InstanceConfig;
import org.apache.pinot.broker.api.AccessControl;
@@ -46,6 +48,7 @@ import org.apache.pinot.common.request.Function;
import org.apache.pinot.common.request.PinotQuery;
import org.apache.pinot.common.response.BrokerResponse;
import org.apache.pinot.common.response.broker.BrokerResponseNative;
+import org.apache.pinot.common.response.broker.QueryProcessingException;
import org.apache.pinot.core.routing.RoutingTable;
import org.apache.pinot.core.routing.SegmentsToQuery;
import org.apache.pinot.core.routing.TableRouteInfo;
@@ -73,11 +76,15 @@ 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.exception.BadQueryRequestException;
+import org.apache.pinot.spi.exception.QueryErrorCode;
import org.apache.pinot.spi.trace.LoggerConstants;
import org.apache.pinot.spi.trace.RequestContext;
+import org.apache.pinot.spi.trace.RequestScope;
+import org.apache.pinot.spi.trace.Tracing;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.CommonConstants.Broker;
import org.apache.pinot.spi.utils.CommonConstants.Query.Range;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.sql.FilterKind;
import org.apache.pinot.sql.parsers.CalciteSqlParser;
@@ -548,6 +555,60 @@ public class BaseSingleStageBrokerRequestHandlerTest {
Assert.assertEquals(operands.get(1).getLiteral().getStringValue(),
expectedRange);
}
+ @Test
+ public void testRejectedSqlOptionsInSubqueryPreserveErrorCode()
+ throws Exception {
+ BaseSingleStageBrokerRequestHandler handler =
createHybridHandlerWithTimeBoundary(new AtomicReference<>());
+
+ BrokerResponseNative response = handleRequest(handler, "SELECT * FROM
myTable WHERE IN_SUBQUERY(created_15min, "
+ + "'SET timeoutMs = ''1''; SELECT ID_SET(created_15min) FROM myTable')
= 1", "sqlOptionsMode=reject");
+ assertSingleException(response, QueryErrorCode.QUERY_VALIDATION, "Query
options are not allowed in the SQL");
+ }
+
+ @Test
+ public void testRejectedSqlOptionsInNestedSubqueryPreserveErrorCode()
+ throws Exception {
+ BaseSingleStageBrokerRequestHandler handler =
createHybridHandlerWithTimeBoundary(new AtomicReference<>());
+
+ // The SET sits two IN_SUBQUERY levels down, so the rejection has to
survive the inner subquery's response
+ BrokerResponseNative response = handleRequest(handler,
+ "SELECT * FROM myTable WHERE IN_SUBQUERY(created_15min, 'SELECT
ID_SET(created_15min) FROM myTable WHERE "
+ + "IN_SUBQUERY(created_15min, ''SET timeoutMs = 1; SELECT
ID_SET(created_15min) FROM myTable'') "
+ + "= 1') = 1", "sqlOptionsMode=reject");
+ assertSingleException(response, QueryErrorCode.QUERY_VALIDATION, "Query
options are not allowed in the SQL");
+ }
+
+ @Test
+ public void testSubqueryParseFailurePreservesErrorCode()
+ throws Exception {
+ BaseSingleStageBrokerRequestHandler handler =
createHybridHandlerWithTimeBoundary(new AtomicReference<>());
+
+ BrokerResponseNative response = handleRequest(handler,
+ "SELECT * FROM myTable WHERE IN_SUBQUERY(created_15min, 'SELECT
ID_SET(created_15min) FROM') = 1", null);
+ assertSingleException(response, QueryErrorCode.SQL_PARSING, "Failed to
parse subquery");
+ }
+
+ private static BrokerResponseNative
handleRequest(BaseSingleStageBrokerRequestHandler handler, String sql,
+ @Nullable String queryOptions)
+ throws Exception {
+ ObjectNode request = JsonUtils.newObjectNode().put(Broker.Request.SQL,
sql);
+ if (queryOptions != null) {
+ request.put(Broker.Request.QUERY_OPTIONS, queryOptions);
+ }
+ try (RequestScope requestContext =
Tracing.getTracer().createRequestScope()) {
+ requestContext.setRequestArrivalTimeMillis(System.currentTimeMillis());
+ return (BrokerResponseNative) handler.handleRequest(request, null, null,
requestContext, null);
+ }
+ }
+
+ private static void assertSingleException(BrokerResponseNative response,
QueryErrorCode errorCode,
+ String messagePart) {
+ Assert.assertEquals(response.getExceptions().size(), 1,
response.toString());
+ QueryProcessingException exception = response.getExceptions().get(0);
+ Assert.assertEquals(exception.getErrorCode(), errorCode.getId());
+ Assert.assertTrue(exception.getMessage().contains(messagePart),
exception.getMessage());
+ }
+
@Test
public void testTimeBoundaryMergesWithBetween()
throws Exception {
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionConfigListener.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionConfigListener.java
new file mode 100644
index 00000000000..8203a64d8ec
--- /dev/null
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionConfigListener.java
@@ -0,0 +1,78 @@
+/**
+ * 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.common.utils.config;
+
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Consumer;
+import org.apache.pinot.common.utils.config.QueryOptionsUtils.SqlOptionsMode;
+import
org.apache.pinot.common.utils.config.QueryOptionsUtils.SqlQueryOptionValidationMode;
+import org.apache.pinot.spi.config.provider.PinotClusterConfigChangeListener;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Applies the `pinot.broker.query.option.*` cluster configs to the
process-wide SQL query option policies held by
+/// [QueryOptionsUtils]:
[Broker#CONFIG_OF_BROKER_QUERY_OPTION_VALIDATION_MODE] and
+/// [Broker#CONFIG_OF_BROKER_QUERY_OPTION_LEGACY_SYNTAX_MODE]. Changes apply
live, and a removed key restores the
+/// default; the broker instance config is not consulted. An invalid value is
logged and the current mode kept, so
+/// that a typo can neither break query parsing nor the other listeners.
+public class QueryOptionConfigListener implements
PinotClusterConfigChangeListener {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(QueryOptionConfigListener.class);
+
+ @Override
+ public void onChange(Set<String> changedConfigs, Map<String, String>
clusterConfigs) {
+ if
(changedConfigs.contains(Broker.CONFIG_OF_BROKER_QUERY_OPTION_VALIDATION_MODE))
{
+ applyValidationMode(clusterConfigs);
+ }
+ if
(changedConfigs.contains(Broker.CONFIG_OF_BROKER_QUERY_OPTION_LEGACY_SYNTAX_MODE))
{
+ applyLegacySyntaxMode(clusterConfigs);
+ }
+ }
+
+ private static void applyValidationMode(Map<String, String> clusterConfigs) {
+ apply(clusterConfigs, Broker.CONFIG_OF_BROKER_QUERY_OPTION_VALIDATION_MODE,
+ Broker.DEFAULT_BROKER_QUERY_OPTION_VALIDATION_MODE,
SqlQueryOptionValidationMode.class,
+ QueryOptionsUtils.getSqlQueryOptionValidationMode(),
QueryOptionsUtils::setSqlQueryOptionValidationMode);
+ }
+
+ private static void applyLegacySyntaxMode(Map<String, String>
clusterConfigs) {
+ apply(clusterConfigs,
Broker.CONFIG_OF_BROKER_QUERY_OPTION_LEGACY_SYNTAX_MODE,
+ Broker.DEFAULT_BROKER_QUERY_OPTION_LEGACY_SYNTAX_MODE,
SqlOptionsMode.class,
+ QueryOptionsUtils.getLegacyOptionSyntaxMode(),
QueryOptionsUtils::setLegacyOptionSyntaxMode);
+ }
+
+ private static <E extends Enum<E>> void apply(Map<String, String>
clusterConfigs, String key, String defaultValue,
+ Class<E> enumClass, E currentValue, Consumer<E> setter) {
+ String value = clusterConfigs.getOrDefault(key, defaultValue);
+ E newValue;
+ try {
+ newValue = Enum.valueOf(enumClass,
value.trim().toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException e) {
+ LOGGER.error("Ignoring invalid value '{}' for cluster config: {},
keeping: {}", value, key, currentValue);
+ return;
+ }
+ if (newValue != currentValue) {
+ setter.accept(newValue);
+ LOGGER.info("Updated cluster config: {} from {} to {}", key,
currentValue, newValue);
+ }
+ }
+}
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 8a9bdb8639b..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
@@ -21,9 +21,11 @@ package org.apache.pinot.common.utils.config;
import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -31,6 +33,7 @@ import java.util.concurrent.ConcurrentHashMap;
import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.spi.config.table.FieldConfig;
+import org.apache.pinot.spi.exception.QueryErrorCode;
import org.apache.pinot.spi.utils.CommonConstants;
import
org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
import
org.apache.pinot.spi.utils.CommonConstants.MultiStageQueryRunner.JoinOverFlowMode;
@@ -56,6 +59,19 @@ public class QueryOptionsUtils {
REJECT
}
+ /// How query options embedded in the SQL text are handled, as opposed to
options passed through the request
+ /// payload. Shared by the brokers' legacy `OPTION(...)` syntax policy
+ ///
([CommonConstants.Broker#CONFIG_OF_BROKER_QUERY_OPTION_LEGACY_SYNTAX_MODE]) and
the per-request
+ /// [QueryOptionKey#SQL_OPTIONS_MODE].
+ public enum SqlOptionsMode {
+ /// The options are applied. Default.
+ ALLOW,
+ /// The options are dropped.
+ IGNORE,
+ /// The statement fails.
+ REJECT
+ }
+
private static final Logger LOGGER =
LoggerFactory.getLogger(QueryOptionsUtils.class);
private static final Map<String, String> CONFIG_RESOLVER;
@@ -83,6 +99,7 @@ public class QueryOptionsUtils {
private static volatile SqlQueryOptionValidationMode
_sqlQueryOptionValidationMode =
SqlQueryOptionValidationMode.NONE;
+ private static volatile SqlOptionsMode _legacyOptionSyntaxMode =
SqlOptionsMode.ALLOW;
static {
// this is a bit hacky, but lots of the code depends directly on usage of
@@ -140,13 +157,37 @@ public class QueryOptionsUtils {
return _sqlQueryOptionValidationMode;
}
- /// Sets the validation mode applied to SQL-supplied query option keys.
Called once per process at
- /// broker startup from
[CommonConstants.Broker#CONFIG_OF_BROKER_QUERY_OPTION_VALIDATION_MODE], and
- /// by tests to restore [SqlQueryOptionValidationMode#NONE].
+ /// Sets the validation mode applied to SQL-supplied query option keys, see
+ /// [CommonConstants.Broker#CONFIG_OF_BROKER_QUERY_OPTION_VALIDATION_MODE].
public static void
setSqlQueryOptionValidationMode(SqlQueryOptionValidationMode mode) {
_sqlQueryOptionValidationMode = mode;
}
+ public static SqlOptionsMode getLegacyOptionSyntaxMode() {
+ return _legacyOptionSyntaxMode;
+ }
+
+ /// Sets how the legacy `OPTION(...)` query option suffix is handled, see
+ ///
[CommonConstants.Broker#CONFIG_OF_BROKER_QUERY_OPTION_LEGACY_SYNTAX_MODE].
+ public static void setLegacyOptionSyntaxMode(SqlOptionsMode mode) {
+ _legacyOptionSyntaxMode = mode;
+ }
+
+ /// Returns the per-request [QueryOptionKey#SQL_OPTIONS_MODE], `ALLOW` when
absent. Fails with
+ /// [QueryErrorCode#QUERY_VALIDATION] when the value is not a
[SqlOptionsMode].
+ public static SqlOptionsMode getSqlOptionsMode(Map<String, String>
queryOptions) {
+ String mode = queryOptions.get(QueryOptionKey.SQL_OPTIONS_MODE);
+ if (mode == null) {
+ return SqlOptionsMode.ALLOW;
+ }
+ try {
+ return SqlOptionsMode.valueOf(mode.trim().toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException e) {
+ throw QueryErrorCode.QUERY_VALIDATION.asException("Invalid value '" +
mode + "' for query option '"
+ + QueryOptionKey.SQL_OPTIONS_MODE + "', must be one of " +
Arrays.toString(SqlOptionsMode.values()));
+ }
+ }
+
/// Registers an option key that [#validateSqlQueryOptions] accepts in
addition to the keys
/// declared on [QueryOptionKey]. Meant for plugins that read custom options
off the
/// `BrokerRequest`; call it from plugin init. Case-insensitive, thread safe
and idempotent.
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/request/RequestUtils.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/request/RequestUtils.java
index 0f5a3b4d0c6..a120893c867 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/request/RequestUtils.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/request/RequestUtils.java
@@ -55,6 +55,9 @@ import org.apache.pinot.common.request.Identifier;
import org.apache.pinot.common.request.Literal;
import org.apache.pinot.common.request.PinotQuery;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.apache.pinot.common.utils.config.QueryOptionsUtils;
+import org.apache.pinot.common.utils.config.QueryOptionsUtils.SqlOptionsMode;
+import org.apache.pinot.spi.exception.QueryErrorCode;
import org.apache.pinot.spi.utils.BigDecimalUtils;
import org.apache.pinot.spi.utils.BytesUtils;
import org.apache.pinot.spi.utils.CommonConstants;
@@ -97,21 +100,34 @@ public class RequestUtils {
return sqlNodeAndOptions;
}
- /// Sets extra options for the given query.
+ /// Merges the request payload options (`queryOptions` and `trace`) into the
options parsed from the SQL. The SQL
+ /// options take precedence, unless the request sets
[Request.QueryOptionKey#SQL_OPTIONS_MODE] to `IGNORE` (the SQL
+ /// options are dropped) or `REJECT` (the query fails when it carries any).
@VisibleForTesting
public static void setOptions(SqlNodeAndOptions sqlNodeAndOptions, JsonNode
jsonRequest) {
- Map<String, String> queryOptions = new HashMap<>();
+ Map<String, String> requestOptions = new HashMap<>();
if (jsonRequest.has(Request.QUERY_OPTIONS)) {
-
queryOptions.putAll(getOptionsFromString(jsonRequest.get(Request.QUERY_OPTIONS).asText()));
+
requestOptions.putAll(getOptionsFromString(jsonRequest.get(Request.QUERY_OPTIONS).asText()));
}
if (jsonRequest.has(Request.TRACE) &&
jsonRequest.get(Request.TRACE).asBoolean()) {
- queryOptions.put(Request.TRACE, "true");
+ requestOptions.put(Request.TRACE, "true");
}
- if (!queryOptions.isEmpty()) {
- LOGGER.debug("Query options are set to: {}", queryOptions);
+ if (requestOptions.isEmpty()) {
+ return;
+ }
+ LOGGER.debug("Query options are set to: {}", requestOptions);
+ requestOptions =
QueryOptionsUtils.resolveCaseInsensitiveOptions(requestOptions);
+ SqlOptionsMode sqlOptionsMode =
QueryOptionsUtils.getSqlOptionsMode(requestOptions);
+ Map<String, String> sqlOptions = sqlNodeAndOptions.getOptions();
+ if (sqlOptionsMode != SqlOptionsMode.ALLOW && !sqlOptions.isEmpty()) {
+ if (sqlOptionsMode == SqlOptionsMode.REJECT) {
+ throw QueryErrorCode.QUERY_VALIDATION.asException(
+ "Query options are not allowed in the SQL for this request, found:
" + sqlOptions.keySet());
+ }
+ sqlOptions.clear();
}
- // Setting all query options back into SqlNodeAndOptions. The above
ordering matters due to priority overwrite rule
- sqlNodeAndOptions.setExtraOptions(queryOptions);
+ // SQL options take precedence over request options
+ requestOptions.forEach(sqlOptions::putIfAbsent);
}
public static Expression getIdentifierExpression(String identifier) {
diff --git
a/pinot-common/src/main/java/org/apache/pinot/sql/parsers/CalciteSqlParser.java
b/pinot-common/src/main/java/org/apache/pinot/sql/parsers/CalciteSqlParser.java
index d82c1a96922..ed16a9d85ad 100644
---
a/pinot-common/src/main/java/org/apache/pinot/sql/parsers/CalciteSqlParser.java
+++
b/pinot-common/src/main/java/org/apache/pinot/sql/parsers/CalciteSqlParser.java
@@ -68,6 +68,7 @@ import org.apache.pinot.common.request.JoinType;
import org.apache.pinot.common.request.PinotQuery;
import org.apache.pinot.common.request.context.GroupingSets;
import org.apache.pinot.common.utils.config.QueryOptionsUtils;
+import org.apache.pinot.common.utils.config.QueryOptionsUtils.SqlOptionsMode;
import org.apache.pinot.common.utils.request.RequestUtils;
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.sql.FilterKind;
@@ -121,9 +122,19 @@ public class CalciteSqlParser {
sql = ParserUtils.sanitizeSql(sql);
// extract and remove OPTIONS string
- List<String> options = extractOptionsFromSql(sql);
- if (!options.isEmpty()) {
+ List<String> options = List.of();
+ SqlOptionsMode legacyOptionSyntaxMode =
QueryOptionsUtils.getLegacyOptionSyntaxMode();
+ if (legacyOptionSyntaxMode == SqlOptionsMode.IGNORE) {
sql = removeOptionsFromSql(sql);
+ } else {
+ options = extractOptionsFromSql(sql);
+ if (!options.isEmpty()) {
+ if (legacyOptionSyntaxMode == SqlOptionsMode.REJECT) {
+ throw new SqlCompilationException("Legacy OPTION(...) query options
are not allowed on this cluster, use "
+ + "'SET <key> = <value>;' statements instead: " + options);
+ }
+ sql = removeOptionsFromSql(sql);
+ }
}
try (StringReader inStream = new StringReader(sql)) {
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/utils/config/QueryOptionConfigListenerTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/utils/config/QueryOptionConfigListenerTest.java
new file mode 100644
index 00000000000..73cae4fea65
--- /dev/null
+++
b/pinot-common/src/test/java/org/apache/pinot/common/utils/config/QueryOptionConfigListenerTest.java
@@ -0,0 +1,71 @@
+/**
+ * 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.common.utils.config;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.pinot.common.utils.config.QueryOptionsUtils.SqlOptionsMode;
+import
org.apache.pinot.common.utils.config.QueryOptionsUtils.SqlQueryOptionValidationMode;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+
+
+public class QueryOptionConfigListenerTest {
+ private static final String VALIDATION_MODE_KEY =
Broker.CONFIG_OF_BROKER_QUERY_OPTION_VALIDATION_MODE;
+ private static final String LEGACY_SYNTAX_MODE_KEY =
Broker.CONFIG_OF_BROKER_QUERY_OPTION_LEGACY_SYNTAX_MODE;
+
+ @BeforeMethod
+ @AfterMethod
+ public void resetModes() {
+
QueryOptionsUtils.setSqlQueryOptionValidationMode(SqlQueryOptionValidationMode.NONE);
+ QueryOptionsUtils.setLegacyOptionSyntaxMode(SqlOptionsMode.ALLOW);
+ }
+
+ @Test
+ public void testAppliesClusterConfigChanges() {
+ QueryOptionConfigListener listener = new QueryOptionConfigListener();
+
+ listener.onChange(Set.of(VALIDATION_MODE_KEY, LEGACY_SYNTAX_MODE_KEY),
+ Map.of(VALIDATION_MODE_KEY, "warn", LEGACY_SYNTAX_MODE_KEY, " reject
"));
+ assertEquals(QueryOptionsUtils.getSqlQueryOptionValidationMode(),
SqlQueryOptionValidationMode.WARN);
+ assertEquals(QueryOptionsUtils.getLegacyOptionSyntaxMode(),
SqlOptionsMode.REJECT);
+
+ // Unrelated changes are ignored
+ listener.onChange(Set.of("otherKey"), Map.of("otherKey", "value"));
+ assertEquals(QueryOptionsUtils.getSqlQueryOptionValidationMode(),
SqlQueryOptionValidationMode.WARN);
+ assertEquals(QueryOptionsUtils.getLegacyOptionSyntaxMode(),
SqlOptionsMode.REJECT);
+
+ // Removing the keys restores the defaults
+ listener.onChange(Set.of(VALIDATION_MODE_KEY, LEGACY_SYNTAX_MODE_KEY),
Map.of());
+ assertEquals(QueryOptionsUtils.getSqlQueryOptionValidationMode(),
SqlQueryOptionValidationMode.NONE);
+ assertEquals(QueryOptionsUtils.getLegacyOptionSyntaxMode(),
SqlOptionsMode.ALLOW);
+ }
+
+ @Test
+ public void testInvalidValueKeepsCurrentMode() {
+ QueryOptionConfigListener listener = new QueryOptionConfigListener();
+ listener.onChange(Set.of(LEGACY_SYNTAX_MODE_KEY),
Map.of(LEGACY_SYNTAX_MODE_KEY, "reject"));
+ listener.onChange(Set.of(LEGACY_SYNTAX_MODE_KEY),
Map.of(LEGACY_SYNTAX_MODE_KEY, "bogus"));
+ assertEquals(QueryOptionsUtils.getLegacyOptionSyntaxMode(),
SqlOptionsMode.REJECT);
+ }
+}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/sql/parsers/SqlOptionsModeTest.java
b/pinot-common/src/test/java/org/apache/pinot/sql/parsers/SqlOptionsModeTest.java
new file mode 100644
index 00000000000..1fd061b69e8
--- /dev/null
+++
b/pinot-common/src/test/java/org/apache/pinot/sql/parsers/SqlOptionsModeTest.java
@@ -0,0 +1,136 @@
+/**
+ * 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.sql.parsers;
+
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.utils.config.QueryOptionsUtils;
+import org.apache.pinot.common.utils.config.QueryOptionsUtils.SqlOptionsMode;
+import org.apache.pinot.common.utils.request.RequestUtils;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+import org.apache.pinot.spi.exception.QueryException;
+import org.apache.pinot.spi.utils.CommonConstants.Broker.Request;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+
+
+/// Covers the two policies for query options embedded in the SQL text: the
legacy `OPTION(...)` syntax mode and the
+/// per-request [Request.QueryOptionKey#SQL_OPTIONS_MODE].
+public class SqlOptionsModeTest {
+
+ @AfterMethod
+ public void resetLegacyOptionSyntaxMode() {
+ QueryOptionsUtils.setLegacyOptionSyntaxMode(SqlOptionsMode.ALLOW);
+ }
+
+ @Test
+ public void testLegacyOptionSyntaxAllowedByDefault() {
+ assertEquals(QueryOptionsUtils.getLegacyOptionSyntaxMode(),
SqlOptionsMode.ALLOW);
+ assertEquals(sqlOptionsOf("select * from vegetables
OPTION(timeoutMs=1000)"), Map.of("timeoutMs", "1000"));
+ }
+
+ @Test
+ public void testIgnoredLegacyOptionSyntaxIsStrippedAndDropped() {
+ QueryOptionsUtils.setLegacyOptionSyntaxMode(SqlOptionsMode.IGNORE);
+ assertEquals(sqlOptionsOf("select * from vegetables
OPTION(timeoutMs=1000)"), Map.of());
+ // SET statements are unaffected
+ assertEquals(sqlOptionsOf("SET timeoutMs='2000'; select * from vegetables
OPTION(timeoutMs=1000, skipUpsert=true)"),
+ Map.of("timeoutMs", "2000"));
+ }
+
+ @Test
+ public void testRejectedLegacyOptionSyntaxFailsEveryStatementType() {
+ QueryOptionsUtils.setLegacyOptionSyntaxMode(SqlOptionsMode.REJECT);
+ for (String sql : List.of("select * from vegetables
OPTION(timeoutMs=1000)",
+ "INSERT INTO db.tbl FROM FILE 'file:///tmp/file1'
OPTION(taskName=myTask-1)")) {
+ SqlCompilationException e = expectThrows(SqlCompilationException.class,
() -> sqlOptionsOf(sql));
+ assertTrue(e.getMessage().contains("OPTION(...)"), e.getMessage());
+ assertTrue(e.getMessage().contains("SET"), e.getMessage());
+ }
+ // SET statements are unaffected
+ assertEquals(sqlOptionsOf("SET timeoutMs='2000'; select * from
vegetables"), Map.of("timeoutMs", "2000"));
+ assertEquals(sqlOptionsOf("SET taskName='myTask-1'; INSERT INTO db.tbl
FROM FILE 'file:///tmp/file1'"),
+ Map.of("taskName", "myTask-1"));
+ }
+
+ @Test
+ public void testSqlOptionsAllowedByDefaultWithPrecedenceOverRequestOptions()
{
+ assertEquals(parse("SET timeoutMs='1000'; select * from vegetables",
"timeoutMs=2000;maxExecutionThreads=4"),
+ Map.of("timeoutMs", "1000", "maxExecutionThreads", "4"));
+ }
+
+ @Test
+ public void testIgnoredSqlOptionsAreDropped() {
+ assertEquals(parse("SET timeoutMs='1000'; select * from vegetables
OPTION(skipUpsert=true)",
+ "timeoutMs=2000;sqlOptionsMode=ignore"), Map.of("timeoutMs", "2000",
"sqlOptionsMode", "ignore"));
+ }
+
+ @Test
+ public void testRejectedSqlOptionsFailTheQuery() {
+ QueryException e = expectThrows(QueryException.class,
+ () -> parse("SET timeoutMs='1000'; select * from vegetables
OPTION(skipUpsert=true)", "sqlOptionsMode=reject"));
+ assertEquals(e.getErrorCode(), QueryErrorCode.QUERY_VALIDATION);
+ assertTrue(e.getMessage().contains("timeoutMs"), e.getMessage());
+ assertTrue(e.getMessage().contains("skipUpsert"), e.getMessage());
+ // A query without SQL options is unaffected
+ assertEquals(parse("select * from vegetables",
"timeoutMs=2000;sqlOptionsMode=reject"),
+ Map.of("timeoutMs", "2000", "sqlOptionsMode", "reject"));
+ }
+
+ @Test
+ public void testSqlOptionsModeIsCaseInsensitive() {
+ QueryException e = expectThrows(QueryException.class,
+ () -> parse("SET timeoutMs='1000'; select * from vegetables",
"SQLOPTIONSMODE=Reject"));
+ assertEquals(e.getErrorCode(), QueryErrorCode.QUERY_VALIDATION);
+ }
+
+ @Test
+ public void testSqlOptionsModeInSqlIsNotHonored() {
+ // Only the request payload can restrict SQL options; the SQL has no say
over its own options
+ assertEquals(parse("SET sqlOptionsMode='reject'; SET timeoutMs='1000';
select * from vegetables", null),
+ Map.of("sqlOptionsMode", "reject", "timeoutMs", "1000"));
+ }
+
+ @Test
+ public void testInvalidSqlOptionsModeFailsEvenWithoutSqlOptions() {
+ QueryException e =
+ expectThrows(QueryException.class, () -> parse("select * from
vegetables", "sqlOptionsMode=bogus"));
+ assertEquals(e.getErrorCode(), QueryErrorCode.QUERY_VALIDATION);
+ assertTrue(e.getMessage().contains("bogus"), e.getMessage());
+ }
+
+ private static Map<String, String> sqlOptionsOf(String sql) {
+ return CalciteSqlParser.compileToSqlNodeAndOptions(sql).getOptions();
+ }
+
+ private static Map<String, String> parse(String sql, @Nullable String
queryOptions) {
+ ObjectNode request = JsonUtils.newObjectNode().put(Request.SQL, sql);
+ if (queryOptions != null) {
+ request.put(Request.QUERY_OPTIONS, queryOptions);
+ }
+ return RequestUtils.parseQuery(sql, request).getOptions();
+ }
+}
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotQueryResource.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotQueryResource.java
index ee2a6bd5a1c..7c3aaafa9d8 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotQueryResource.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotQueryResource.java
@@ -57,7 +57,6 @@ import javax.ws.rs.core.HttpHeaders;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.StreamingOutput;
-import org.apache.calcite.sql.SqlNode;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.io.IOUtils;
import org.apache.hc.core5.net.URIBuilder;
@@ -91,7 +90,6 @@ import org.apache.pinot.spi.utils.CommonConstants;
import
org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
-import org.apache.pinot.sql.parsers.CalciteSqlCompiler;
import org.apache.pinot.sql.parsers.CalciteSqlParser;
import org.apache.pinot.sql.parsers.PinotSqlType;
import org.apache.pinot.sql.parsers.SqlNodeAndOptions;
@@ -389,13 +387,11 @@ public class PinotQueryResource {
@Nullable String queryOptions)
throws Exception {
LOGGER.debug("Trace: {}, Running query: {}", traceEnabled, sqlQuery);
- SqlNodeAndOptions sqlNodeAndOptions;
- sqlNodeAndOptions = CalciteSqlParser.compileToSqlNodeAndOptions(sqlQuery);
+ // Parse with the exact payload forwarded to the broker, so that the
options used to route the query (engine,
+ // database) are the ones the broker will apply
+ ObjectNode requestJson = getRequestJson(sqlQuery, traceEnabled,
queryOptions);
+ SqlNodeAndOptions sqlNodeAndOptions = RequestUtils.parseQuery(sqlQuery,
requestJson);
Map<String, String> options = sqlNodeAndOptions.getOptions();
- if (queryOptions != null) {
- Map<String, String> optionsFromString =
RequestUtils.getOptionsFromString(queryOptions);
- sqlNodeAndOptions.setExtraOptions(optionsFromString);
- }
PinotSqlType sqlType = sqlNodeAndOptions.getSqlType();
if (sqlType == PinotSqlType.DDL) {
throw QueryErrorCode.QUERY_VALIDATION.asException(
@@ -414,8 +410,8 @@ public class PinotQueryResource {
switch (sqlType) {
case DQL:
return isMse
- ? getMultiStageQueryResponse(sqlQuery, queryOptions, httpHeaders,
traceEnabled)
- : getQueryResponse(sqlQuery, sqlNodeAndOptions.getSqlNode(),
traceEnabled, queryOptions, httpHeaders);
+ ? getMultiStageQueryResponse(sqlQuery, sqlNodeAndOptions,
requestJson, httpHeaders)
+ : getQueryResponse(sqlQuery, sqlNodeAndOptions, requestJson,
httpHeaders);
case DML:
Map<String, String> headers = extractHeaders(httpHeaders);
return output -> {
@@ -428,8 +424,8 @@ public class PinotQueryResource {
}
}
- private StreamingOutput getMultiStageQueryResponse(String query, String
queryOptions, HttpHeaders httpHeaders,
- String traceEnabled) {
+ private StreamingOutput getMultiStageQueryResponse(String query,
SqlNodeAndOptions sqlNodeAndOptions,
+ ObjectNode requestJson, HttpHeaders httpHeaders) {
// Validate data access
// we don't have a cross table access control rule so only ADMIN can make
request to multi-stage engine.
@@ -438,23 +434,19 @@ public class PinotQueryResource {
throw new WebApplicationException("Permission denied",
Response.Status.FORBIDDEN);
}
- Map<String, String> queryOptionsMap =
RequestUtils.parseQuery(query).getOptions();
- if (queryOptions != null) {
- queryOptionsMap.putAll(RequestUtils.getOptionsFromString(queryOptions));
- }
- String database =
DatabaseUtils.extractDatabaseFromQueryRequest(queryOptionsMap, httpHeaders);
- List<String> tableNames = getTableNames(query, database);
+ String database =
DatabaseUtils.extractDatabaseFromQueryRequest(sqlNodeAndOptions.getOptions(),
httpHeaders);
+ List<String> tableNames = getTableNames(query, sqlNodeAndOptions,
database);
List<String> instanceIds = getInstanceIds(query, tableNames, database);
String instanceId = selectRandomInstanceId(instanceIds);
- return sendRequestToBroker(query, instanceId, traceEnabled, queryOptions,
httpHeaders);
+ return sendRequestToBroker(query, instanceId, requestJson, httpHeaders);
}
- private List<String> getTableNames(String query, String database) {
+ private List<String> getTableNames(String query, SqlNodeAndOptions
sqlNodeAndOptions, String database) {
QueryEnvironment queryEnvironment =
new QueryEnvironment(database,
_pinotHelixResourceManager.getTableCache(), null);
List<String> tableNames;
- try (QueryEnvironment.CompiledQuery compiledQuery =
queryEnvironment.compile(query)) {
+ try (QueryEnvironment.CompiledQuery compiledQuery =
queryEnvironment.compile(query, sqlNodeAndOptions)) {
tableNames = new ArrayList<>(compiledQuery.getTableNames());
} catch (QueryException e) {
if (e.getErrorCode() != QueryErrorCode.UNKNOWN) {
@@ -493,24 +485,19 @@ public class PinotQueryResource {
return instanceIds;
}
- private StreamingOutput getQueryResponse(String query, @Nullable SqlNode
sqlNode, String traceEnabled,
- String queryOptions, HttpHeaders httpHeaders) {
+ private StreamingOutput getQueryResponse(String query, SqlNodeAndOptions
sqlNodeAndOptions, ObjectNode requestJson,
+ HttpHeaders httpHeaders) {
// Get resource table name.
String tableName;
- Map<String, String> queryOptionsMap =
RequestUtils.parseQuery(query).getOptions();
- if (queryOptions != null) {
- queryOptionsMap.putAll(RequestUtils.getOptionsFromString(queryOptions));
- }
String database;
try {
- database =
DatabaseUtils.extractDatabaseFromQueryRequest(queryOptionsMap, httpHeaders);
+ database =
DatabaseUtils.extractDatabaseFromQueryRequest(sqlNodeAndOptions.getOptions(),
httpHeaders);
} catch (DatabaseConflictException e) {
throw QueryErrorCode.QUERY_VALIDATION.asException(e);
}
try {
- String inputTableName =
- sqlNode != null ?
RequestUtils.getTableNames(CalciteSqlParser.compileSqlNodeToPinotQuery(sqlNode)).iterator()
- .next() :
CalciteSqlCompiler.compileToBrokerRequest(query).getQuerySource().getTableName();
+ String inputTableName = RequestUtils.getTableNames(
+
CalciteSqlParser.compileSqlNodeToPinotQuery(sqlNodeAndOptions.getSqlNode())).iterator().next();
tableName =
_pinotHelixResourceManager.getActualTableName(inputTableName, database);
} catch (Exception e) {
LOGGER.error("Caught exception while compiling query: {}", query, e);
@@ -536,7 +523,7 @@ public class PinotQueryResource {
// Get brokers for the resource table.
List<String> instanceIds =
_pinotHelixResourceManager.getBrokerInstancesFor(rawTableName);
String instanceId = selectRandomInstanceId(instanceIds);
- return sendRequestToBroker(query, instanceId, traceEnabled, queryOptions,
httpHeaders);
+ return sendRequestToBroker(query, instanceId, requestJson, httpHeaders);
}
// given a list of tables, returns the list of tableConfigs
@@ -602,14 +589,13 @@ public class PinotQueryResource {
return
brokerInstanceConfigs.map(InstanceConfig::getInstanceName).collect(Collectors.toList());
}
- private StreamingOutput sendRequestToBroker(String query, String instanceId,
String traceEnabled, String queryOptions,
+ private StreamingOutput sendRequestToBroker(String query, String instanceId,
ObjectNode requestJson,
HttpHeaders httpHeaders) {
InstanceConfig instanceConfig = getInstanceConfig(instanceId);
String hostName = getHost(instanceConfig);
String protocol = _controllerConf.getControllerBrokerProtocol();
int port = getPort(instanceConfig);
String url = getQueryURL(protocol, hostName, port);
- ObjectNode requestJson = getRequestJson(query, traceEnabled, queryOptions);
// Forward client-supplied headers
Map<String, String> headers = extractHeaders(httpHeaders);
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotQueryResourceTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotQueryResourceTest.java
index 70f7f02bce6..c516c58bd1a 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotQueryResourceTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotQueryResourceTest.java
@@ -22,7 +22,9 @@ import java.io.ByteArrayOutputStream;
import javax.ws.rs.core.StreamingOutput;
import org.apache.pinot.common.config.provider.TableCache;
import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.api.access.AccessControl;
import org.apache.pinot.controller.api.access.AccessControlFactory;
+import org.apache.pinot.controller.api.access.AccessType;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.exception.QueryErrorCode;
@@ -35,6 +37,8 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -108,6 +112,42 @@ public class PinotQueryResourceTest {
Assert.assertTrue(response.contains("/sql/ddl"));
}
+ @Test
+ public void testSqlOptionsDecideTheEngine() {
+ // The mocked controller conf reports the multi-stage engine as disabled,
so routing to it fails with INTERNAL
+ String response = streamingOutputToString(
+ _pinotQueryResource.handleGetSql("SET useMultistageEngine = 'true';
SELECT * FROM a", null, null, null));
+
Assert.assertTrue(response.contains(String.valueOf(QueryErrorCode.INTERNAL.getId())),
response);
+ Assert.assertTrue(response.contains("Multi-Stage query engine not
enabled"), response);
+ }
+
+ @Test
+ public void testIgnoredSqlOptionsDoNotDecideTheEngine() {
+ mockSingleStageBrokerSelection();
+ String response = streamingOutputToString(
+ _pinotQueryResource.handleGetSql("SET useMultistageEngine = 'true';
SELECT * FROM a", null,
+ "sqlOptionsMode=ignore", null));
+
Assert.assertTrue(response.contains(String.valueOf(QueryErrorCode.BROKER_RESOURCE_MISSING.getId())),
response);
+ }
+
+ @Test
+ public void testRejectedSqlOptionsFailBeforeRouting() {
+ String response = streamingOutputToString(
+ _pinotQueryResource.handleGetSql("SET useMultistageEngine = 'true';
SELECT * FROM a", null,
+ "sqlOptionsMode=reject", null));
+
Assert.assertTrue(response.contains(String.valueOf(QueryErrorCode.QUERY_VALIDATION.getId())),
response);
+ Assert.assertTrue(response.contains("useMultistageEngine"), response);
+ }
+
+ /// Lets a single-stage query get as far as broker selection, which fails
with BROKER_RESOURCE_MISSING because no
+ /// broker serves the table. Reaching that point proves the query was routed
to the single-stage engine.
+ private void mockSingleStageBrokerSelection() {
+ when(_resourceManager.getActualTableName(any(),
any())).then(AdditionalAnswers.returnsFirstArg());
+ AccessControl accessControl = mock(AccessControl.class);
+ when(accessControl.hasAccess(any(), eq(AccessType.READ), any(),
any())).thenReturn(true);
+ when(_accessControlFactory.create()).thenReturn(accessControl);
+ }
+
public static String streamingOutputToString(StreamingOutput
streamingOutput) {
try (ByteArrayOutputStream byteArrayOutputStream = new
ByteArrayOutputStream()) {
streamingOutput.write(byteArrayOutputStream);
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/BaseCombineOperator.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/BaseCombineOperator.java
index 1c56a538d82..f4e4f67f6c0 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/BaseCombineOperator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/BaseCombineOperator.java
@@ -286,13 +286,7 @@ public abstract class BaseCombineOperator<T extends
BaseResultsBlock> extends Ba
+ " on segment " + operator.getIndexSegment().getSegmentName()
: "Caught exception while doing operator: " + operator.getClass();
- QueryErrorCode errorCode;
- if (e instanceof QueryException) {
- QueryException queryException = (QueryException) e;
- errorCode = queryException.getErrorCode();
- } else {
- errorCode = QueryErrorCode.QUERY_EXECUTION;
- }
+ QueryErrorCode errorCode = QueryErrorCode.fromThrowable(e,
QueryErrorCode.QUERY_EXECUTION);
// TODO: Only include exception message if it is a QueryException.
Otherwise, it might expose sensitive information
throw errorCode.asException(errorMessage + ": " + e.getMessage(), e);
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/streaming/BaseStreamingCombineOperator.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/streaming/BaseStreamingCombineOperator.java
index e49b4fb1272..ca1b5930bbe 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/streaming/BaseStreamingCombineOperator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/streaming/BaseStreamingCombineOperator.java
@@ -35,7 +35,6 @@ import
org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.spi.exception.EarlyTerminationException;
import org.apache.pinot.spi.exception.QueryErrorCode;
import org.apache.pinot.spi.exception.QueryErrorMessage;
-import org.apache.pinot.spi.exception.QueryException;
import org.apache.pinot.spi.query.QueryThreadContext;
import org.apache.pinot.spi.utils.CommonConstants;
import org.slf4j.Logger;
@@ -172,12 +171,8 @@ public abstract class BaseStreamingCombineOperator<T
extends BaseResultsBlock> e
_processingException.compareAndSet(null, t);
// Clear the blocking queue and add the exception results block to
terminate the main thread
_blockingQueue.clear();
- QueryErrorMessage errorMsg;
- if (t instanceof QueryException) {
- errorMsg = QueryErrorMessage.safeMsg(((QueryException)
t).getErrorCode(), t.getMessage());
- } else {
- errorMsg = QueryErrorMessage.safeMsg(QueryErrorCode.QUERY_EXECUTION,
t.getMessage());
- }
+ QueryErrorMessage errorMsg =
+ QueryErrorMessage.safeMsg(QueryErrorCode.fromThrowable(t,
QueryErrorCode.QUERY_EXECUTION), t.getMessage());
_blockingQueue.offer(new ExceptionResultsBlock(errorMsg));
}
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java
index cc20b3d8408..24cbdef09ec 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java
@@ -48,7 +48,6 @@ import
org.apache.pinot.query.runtime.plan.MultiStageQueryStats;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.exception.QueryCancelledException;
import org.apache.pinot.spi.exception.QueryErrorCode;
-import org.apache.pinot.spi.exception.QueryException;
import org.apache.pinot.spi.exception.TerminationException;
import org.apache.pinot.spi.metrics.PinotMeter;
import org.apache.pinot.spi.query.QueryExecutionContext;
@@ -213,18 +212,14 @@ public class OpChainSchedulerService {
public void onFailure(Throwable t) {
String logMsg = "Failed to execute operator chain: " + t.getMessage();
_metrics.onOpChainFinished(rootOperator);
- if (t instanceof QueryException) {
- switch (((QueryException) t).getErrorCode()) {
- case UNKNOWN:
- case INTERNAL:
- LOGGER.error(logMsg, t);
- break;
- default:
- LOGGER.warn(logMsg);
- break;
- }
- } else {
- LOGGER.error(logMsg, t);
+ switch (QueryErrorCode.fromThrowable(t, QueryErrorCode.UNKNOWN)) {
+ case UNKNOWN:
+ case INTERNAL:
+ LOGGER.error(logMsg, t);
+ break;
+ default:
+ LOGGER.warn(logMsg);
+ break;
}
decrementActiveOpChains(requestId);
notifyCompletionListener(opChainId, operatorChain, statsRef.get(), t);
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/exception/QueryErrorCode.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/exception/QueryErrorCode.java
index cfee7557e4c..78286d4f44f 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/exception/QueryErrorCode.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/exception/QueryErrorCode.java
@@ -142,6 +142,12 @@ public enum QueryErrorCode {
return queryErrorCode;
}
+ /// Returns the error code carried by the given throwable when it is a
[QueryException], or the given default
+ /// otherwise.
+ public static QueryErrorCode fromThrowable(Throwable t, QueryErrorCode
defaultErrorCode) {
+ return t instanceof QueryException ? ((QueryException) t).getErrorCode() :
defaultErrorCode;
+ }
+
public static <T> Map<QueryErrorCode, T> fromKeyMap(Map<Integer, T>
originalMap) {
EnumMap<QueryErrorCode, T> newMap = new EnumMap<>(QueryErrorCode.class);
for (Map.Entry<Integer, T> entry : originalMap.entrySet()) {
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 7fbf935c687..e0de1210a65 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
@@ -384,14 +384,22 @@ public class CommonConstants {
public static final String DEFAULT_BROKER_QUERY_LOG_SQL_REDACTION = "none";
public static final String CONFIG_OF_BROKER_QUERY_ENABLE_NULL_HANDLING =
"pinot.broker.query.enable.null.handling";
/// How query option keys supplied through SQL `SET` / `OPTION(...)` on
DQL queries are validated.
- /// Broker config key: `pinot.broker.query.option.validationMode`.
- /// One of `QueryOptionsUtils.SqlQueryOptionValidationMode`: `NONE`
(default, unknown keys are
- /// preserved silently, as they always have been), `WARN` (preserved,
logged once per distinct
- /// unknown key) or `REJECT` (query fails). Plugins can allowlist their
own keys for `REJECT` via
+ /// Cluster config key: `pinot.broker.query.option.validationMode`,
applied live and not read from the broker
+ /// instance config. One of
`QueryOptionsUtils.SqlQueryOptionValidationMode`: `NONE` (default, unknown keys
are
+ /// preserved silently, as they always have been), `WARN` (preserved,
logged once per distinct unknown key) or
+ /// `REJECT` (query fails). Plugins can allowlist their own keys for
`REJECT` via
/// `QueryOptionsUtils.registerSqlQueryOptionKey`.
public static final String CONFIG_OF_BROKER_QUERY_OPTION_VALIDATION_MODE =
"pinot.broker.query.option.validationMode";
public static final String DEFAULT_BROKER_QUERY_OPTION_VALIDATION_MODE =
"NONE";
+ /// How the legacy PQL-style `OPTION(key=value)` query option suffix is
handled.
+ /// Cluster config key: `pinot.broker.query.option.legacySyntaxMode`,
applied live and not read from the broker
+ /// instance config. One of `QueryOptionsUtils.SqlOptionsMode`: `ALLOW`
(default) applies the options as always,
+ /// `IGNORE` strips the suffix and drops its options, `REJECT` fails the
statement with an error pointing at
+ /// `SET`. Applies to every statement type, since `SET` covers them all.
+ public static final String
CONFIG_OF_BROKER_QUERY_OPTION_LEGACY_SYNTAX_MODE =
+ "pinot.broker.query.option.legacySyntaxMode";
+ public static final String DEFAULT_BROKER_QUERY_OPTION_LEGACY_SYNTAX_MODE
= "ALLOW";
/// When true, the broker initializes the materialized view metadata cache
and query rewrite
/// engine. When false (default), MV rewrite is disabled regardless of
per-MV
/// `rewriteEnabled` setting.
@@ -902,6 +910,13 @@ public class CommonConstants {
public static final String USE_FIXED_REPLICA = "useFixedReplica";
public static final String EXPLAIN_PLAN_VERBOSE = "explainPlanVerbose";
public static final String USE_MULTISTAGE_ENGINE =
"useMultistageEngine";
+ /// How query options embedded in the SQL text (`SET` statements and
the legacy `OPTION(...)` suffix) are
+ /// handled for this request, one of
`QueryOptionsUtils.SqlOptionsMode`: `ALLOW` (default) merges them with
+ /// precedence over the request options, as always; `IGNORE` drops
them so that only the request options
+ /// apply; `REJECT` fails the query when it carries any. Only honored
from the request payload
+ /// (`queryOptions`, gRPC metadata), never from the SQL itself, so a
gateway that sets request options on
+ /// behalf of its users can guarantee the query text cannot override
them.
+ public static final String SQL_OPTIONS_MODE = "sqlOptionsMode";
public static final String INFER_PARTITION_HINT = "inferPartitionHint";
public static final String ENABLE_NULL_HANDLING = "enableNullHandling";
public static final String APPLICATION_NAME = "applicationName";
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
