This is an automated email from the ASF dual-hosted git repository.
jackie 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 cb065e4 Refactor QueryOptions into QueryOptionsUtils to avoid
redundant parsing (#7601)
cb065e4 is described below
commit cb065e4f7ab74d4e8d4e485667bba8fb80aa307d
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Wed Oct 20 11:42:33 2021 -0700
Refactor QueryOptions into QueryOptionsUtils to avoid redundant parsing
(#7601)
---
.../requesthandler/BaseBrokerRequestHandler.java | 79 +++++++++++-----------
.../broker/requesthandler/QueryValidationTest.java | 4 +-
.../core/common/datatable/DataTableUtils.java | 4 +-
.../combine/GroupByOrderByCombineOperator.java | 12 ++--
.../plan/AggregationGroupByOrderByPlanNode.java | 11 +--
.../apache/pinot/core/plan/CombinePlanNode.java | 6 +-
.../org/apache/pinot/core/plan/FilterPlanNode.java | 24 +++----
.../core/plan/maker/InstancePlanMakerImplV2.java | 23 +++----
.../query/executor/ServerQueryExecutorV1Impl.java | 26 ++++---
.../query/reduce/AggregationDataTableReducer.java | 12 ++--
.../query/reduce/DistinctDataTableReducer.java | 8 +--
.../core/query/reduce/GroupByDataTableReducer.java | 10 +--
.../query/reduce/SelectionDataTableReducer.java | 17 ++---
.../core/query/request/context/QueryContext.java | 8 ++-
.../{QueryOptions.java => QueryOptionsUtils.java} | 58 ++++------------
15 files changed, 128 insertions(+), 174 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java
index b767fc0..2ac2f2a 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
@@ -80,7 +80,7 @@ import org.apache.pinot.core.query.optimizer.QueryOptimizer;
import org.apache.pinot.core.query.reduce.BrokerReduceService;
import org.apache.pinot.core.requesthandler.PinotQueryParserFactory;
import org.apache.pinot.core.transport.ServerInstance;
-import org.apache.pinot.core.util.QueryOptions;
+import org.apache.pinot.core.util.QueryOptionsUtils;
import org.apache.pinot.pql.parsers.pql2.ast.FilterKind;
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.spi.config.table.TableConfig;
@@ -158,9 +158,9 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
_numDroppedLogRateLimiter = RateLimiter.create(1.0);
_brokerReduceService = new BrokerReduceService(_config);
- LOGGER
- .info("Broker Id: {}, timeout: {}ms, query response limit: {}, query
log length: {}, query log max rate: {}qps",
- _brokerId, _brokerTimeoutMs, _queryResponseLimit, _queryLogLength,
_queryLogRateLimiter.getRate());
+ LOGGER.info(
+ "Broker Id: {}, timeout: {}ms, query response limit: {}, query log
length: {}, query log max rate: {}qps",
+ _brokerId, _brokerTimeoutMs, _queryResponseLimit, _queryLogLength,
_queryLogRateLimiter.getRate());
}
private String getDefaultBrokerId() {
@@ -226,17 +226,16 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
return processLiteralOnlyQuery(pinotQuery, compilationStartTimeNs,
requestStatistics);
} catch (Exception e) {
// TODO: refine the exceptions here to early termination the queries
won't requires to send to servers.
- LOGGER
- .warn("Unable to execute literal request {}: {} at broker,
fallback to server query. {}", requestId, query,
- e.getMessage());
+ LOGGER.warn("Unable to execute literal request {}: {} at broker,
fallback to server query. {}", requestId,
+ query, e.getMessage());
}
}
try {
handleSubquery(pinotQuery, requestId, request, requesterIdentity,
requestStatistics);
} catch (Exception e) {
- LOGGER
- .info("Caught exception while handling the subquery in request {}:
{}, {}", requestId, query, e.getMessage());
+ LOGGER.info("Caught exception while handling the subquery in request {}:
{}, {}", requestId, query,
+ e.getMessage());
requestStatistics.setErrorCode(QueryException.QUERY_EXECUTION_ERROR_CODE);
return new
BrokerResponseNative(QueryException.getException(QueryException.QUERY_EXECUTION_ERROR,
e));
}
@@ -249,8 +248,8 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
try {
updateColumnNames(rawTableName, pinotQuery);
} catch (Exception e) {
- LOGGER
- .warn("Caught exception while updating column names in request {}:
{}, {}", requestId, query, e.getMessage());
+ LOGGER.warn("Caught exception while updating column names in request {}:
{}, {}", requestId, query,
+ e.getMessage());
}
if (_defaultHllLog2m > 0) {
handleHLLLog2mOverride(pinotQuery, _defaultHllLog2m);
@@ -274,8 +273,8 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
requestStatistics.setErrorCode(QueryException.ACCESS_DENIED_ERROR_CODE);
return new BrokerResponseNative(QueryException.ACCESS_DENIED_ERROR);
}
- _brokerMetrics
- .addPhaseTiming(rawTableName, BrokerQueryPhase.AUTHORIZATION,
System.nanoTime() - compilationEndTimeNs);
+ _brokerMetrics.addPhaseTiming(rawTableName, BrokerQueryPhase.AUTHORIZATION,
+ System.nanoTime() - compilationEndTimeNs);
// Get the tables hit by the request
String offlineTableName = null;
@@ -346,11 +345,11 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
if (offlineTableName != null && realtimeTableName != null) {
// Hybrid
offlineBrokerRequest = getOfflineBrokerRequest(brokerRequest);
- _queryOptimizer
- .optimize(offlineBrokerRequest.getPinotQuery(),
_tableCache.getTableConfig(offlineTableName), schema);
+ _queryOptimizer.optimize(offlineBrokerRequest.getPinotQuery(),
_tableCache.getTableConfig(offlineTableName),
+ schema);
realtimeBrokerRequest = getRealtimeBrokerRequest(brokerRequest);
- _queryOptimizer
- .optimize(realtimeBrokerRequest.getPinotQuery(),
_tableCache.getTableConfig(realtimeTableName), schema);
+ _queryOptimizer.optimize(realtimeBrokerRequest.getPinotQuery(),
_tableCache.getTableConfig(realtimeTableName),
+ schema);
requestStatistics.setFanoutType(RequestStatistics.FanoutType.HYBRID);
requestStatistics.setOfflineServerTenant(getServerTenant(offlineTableName));
requestStatistics.setRealtimeServerTenant(getServerTenant(realtimeTableName));
@@ -481,8 +480,8 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
processBrokerRequest(requestId, brokerRequest, offlineBrokerRequest,
offlineRoutingTable, realtimeBrokerRequest,
realtimeRoutingTable, remainingTimeMs, serverStats,
requestStatistics);
long executionEndTimeNs = System.nanoTime();
- _brokerMetrics
- .addPhaseTiming(rawTableName, BrokerQueryPhase.QUERY_EXECUTION,
executionEndTimeNs - routingEndTimeNs);
+ _brokerMetrics.addPhaseTiming(rawTableName,
BrokerQueryPhase.QUERY_EXECUTION,
+ executionEndTimeNs - routingEndTimeNs);
// Track number of queries with number of groups limit reached
if (brokerResponse.isNumGroupsLimitReached()) {
@@ -589,8 +588,8 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
try {
handleSubquery(brokerRequest, requestId, request, requesterIdentity,
requestStatistics);
} catch (Exception e) {
- LOGGER
- .info("Caught exception while handling the subquery in request {}:
{}, {}", requestId, query, e.getMessage());
+ LOGGER.info("Caught exception while handling the subquery in request {}:
{}, {}", requestId, query,
+ e.getMessage());
requestStatistics.setErrorCode(QueryException.QUERY_EXECUTION_ERROR_CODE);
return new
BrokerResponseNative(QueryException.getException(QueryException.QUERY_EXECUTION_ERROR,
e));
}
@@ -603,8 +602,8 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
try {
updateColumnNames(rawTableName, brokerRequest);
} catch (Exception e) {
- LOGGER
- .warn("Caught exception while updating column names in request {}:
{}, {}", requestId, query, e.getMessage());
+ LOGGER.warn("Caught exception while updating column names in request {}:
{}, {}", requestId, query,
+ e.getMessage());
}
if (_defaultHllLog2m > 0) {
handleHyperloglogLog2mOverride(brokerRequest, _defaultHllLog2m);
@@ -628,8 +627,8 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
requestStatistics.setErrorCode(QueryException.ACCESS_DENIED_ERROR_CODE);
return new BrokerResponseNative(QueryException.ACCESS_DENIED_ERROR);
}
- _brokerMetrics
- .addPhaseTiming(rawTableName, BrokerQueryPhase.AUTHORIZATION,
System.nanoTime() - compilationEndTimeNs);
+ _brokerMetrics.addPhaseTiming(rawTableName, BrokerQueryPhase.AUTHORIZATION,
+ System.nanoTime() - compilationEndTimeNs);
// Get the tables hit by the request
String offlineTableName = null;
@@ -794,8 +793,8 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
processBrokerRequest(requestId, brokerRequest, offlineBrokerRequest,
offlineRoutingTable, realtimeBrokerRequest,
realtimeRoutingTable, remainingTimeMs, serverStats,
requestStatistics);
long executionEndTimeNs = System.nanoTime();
- _brokerMetrics
- .addPhaseTiming(rawTableName, BrokerQueryPhase.QUERY_EXECUTION,
executionEndTimeNs - routingEndTimeNs);
+ _brokerMetrics.addPhaseTiming(rawTableName,
BrokerQueryPhase.QUERY_EXECUTION,
+ executionEndTimeNs - routingEndTimeNs);
// Track number of queries with number of groups limit reached
if (brokerResponse.isNumGroupsLimitReached()) {
@@ -902,8 +901,8 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
String serializedIdSet = (String)
response.getAggregationResults().get(0).getValue();
expression.setValue(TransformFunctionType.INIDSET.name());
- children
- .set(1, new
TransformExpressionTree(TransformExpressionTree.ExpressionType.LITERAL,
serializedIdSet, null));
+ children.set(1,
+ new
TransformExpressionTree(TransformExpressionTree.ExpressionType.LITERAL,
serializedIdSet, null));
} else {
for (TransformExpressionTree child : children) {
handleSubquery(child, requestId, jsonRequest, requesterIdentity,
requestStatistics);
@@ -1535,7 +1534,7 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
private long setQueryTimeout(String tableNameWithType, Map<String, String>
queryOptions, long timeSpentMs)
throws TimeoutException {
long queryTimeoutMs;
- Long queryLevelTimeoutMs = QueryOptions.getTimeoutMs(queryOptions);
+ Long queryLevelTimeoutMs = QueryOptionsUtils.getTimeoutMs(queryOptions);
if (queryLevelTimeoutMs != null) {
// Use query-level timeout if exists
queryTimeoutMs = queryLevelTimeoutMs;
@@ -1552,8 +1551,8 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
long remainingTimeMs = queryTimeoutMs - timeSpentMs;
if (remainingTimeMs <= 0) {
- String errorMessage = String
- .format("Query timed out (time spent: %dms, timeout: %dms) for
table: %s before scattering the request",
+ String errorMessage =
+ String.format("Query timed out (time spent: %dms, timeout: %dms) for
table: %s before scattering the request",
timeSpentMs, queryTimeoutMs, tableNameWithType);
throw new TimeoutException(errorMessage);
}
@@ -1651,10 +1650,10 @@ public abstract class BaseBrokerRequestHandler
implements BrokerRequestHandler {
// The results of PQL groupByMode (if numAggregations > 1) cannot be
returned in SQL responseFormat, as the
// results are non-tabular
// Checking for this upfront, to avoid executing the query and wasting
resources
- QueryOptions queryOptions = new
QueryOptions(brokerRequest.getQueryOptions());
if (brokerRequest.isSetAggregationsInfo() && brokerRequest.getGroupBy() !=
null) {
- if (brokerRequest.getAggregationsInfoSize() > 1 &&
queryOptions.isResponseFormatSQL() && !queryOptions
- .isGroupByModeSQL()) {
+ Map<String, String> queryOptions = brokerRequest.getQueryOptions();
+ if (brokerRequest.getAggregationsInfoSize() > 1 &&
!QueryOptionsUtils.isGroupByModeSQL(queryOptions)
+ && QueryOptionsUtils.isResponseFormatSQL(queryOptions)) {
throw new UnsupportedOperationException(
"The results of a GROUP BY query with multiple aggregations in PQL
is not tabular, and cannot be returned"
+ " in SQL responseFormat");
@@ -1716,8 +1715,9 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
// SQL query should always have response format and group-by mode set to
SQL
// TODO: Remove these 2 options after deprecating PQL
- QueryOptions queryOptions = new QueryOptions(pinotQuery.getQueryOptions());
- if (!queryOptions.isGroupByModeSQL() ||
!queryOptions.isResponseFormatSQL()) {
+ Map<String, String> queryOptions = pinotQuery.getQueryOptions();
+ if (queryOptions == null ||
!QueryOptionsUtils.isGroupByModeSQL(queryOptions)
+ || !QueryOptionsUtils.isResponseFormatSQL(queryOptions)) {
throw new IllegalStateException("SQL query should always have response
format and group-by mode set to SQL");
}
}
@@ -1767,8 +1767,9 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
Expression timeFilterExpression = RequestUtils.getFunctionExpression(
isOfflineRequest ? FilterKind.LESS_THAN_OR_EQUAL.name() :
FilterKind.GREATER_THAN.name());
- timeFilterExpression.getFunctionCall().setOperands(Arrays
- .asList(RequestUtils.createIdentifierExpression(timeColumn),
RequestUtils.getLiteralExpression(timeValue)));
+ timeFilterExpression.getFunctionCall().setOperands(
+ Arrays.asList(RequestUtils.createIdentifierExpression(timeColumn),
+ RequestUtils.getLiteralExpression(timeValue)));
Expression filterExpression = pinotQuery.getFilterExpression();
if (filterExpression != null) {
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/QueryValidationTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/QueryValidationTest.java
index a5a7dd0..96e7768 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/QueryValidationTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/QueryValidationTest.java
@@ -126,7 +126,7 @@ public class QueryValidationTest {
BaseBrokerRequestHandler.validateRequest(brokerRequest, 1000);
Assert.fail("Query should have failed");
} catch (Exception e) {
- Assert.assertEquals(errorMessage, e.getMessage());
+ Assert.assertEquals(e.getMessage(), errorMessage);
}
}
@@ -136,7 +136,7 @@ public class QueryValidationTest {
BaseBrokerRequestHandler.validateRequest(pinotQuery, 1000);
Assert.fail("Query should have failed");
} catch (Exception e) {
- Assert.assertEquals(errorMessage, e.getMessage());
+ Assert.assertEquals(e.getMessage(), errorMessage);
}
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableUtils.java
b/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableUtils.java
index 1ff7df2..05b6e3b 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableUtils.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableUtils.java
@@ -36,7 +36,7 @@ import
org.apache.pinot.core.query.aggregation.function.DistinctAggregationFunct
import org.apache.pinot.core.query.distinct.DistinctTable;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.core.query.request.context.utils.QueryContextUtils;
-import org.apache.pinot.core.util.QueryOptions;
+import org.apache.pinot.core.util.QueryOptionsUtils;
/**
@@ -135,7 +135,7 @@ public class DataTableUtils {
if (groupByExpressions != null) {
// Aggregation group-by query
- if (new QueryOptions(queryContext.getQueryOptions()).isGroupByModeSQL())
{
+ if (QueryOptionsUtils.isGroupByModeSQL(queryContext.getQueryOptions())) {
// SQL format
int numColumns = groupByExpressions.size() + numAggregations;
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByOrderByCombineOperator.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByOrderByCombineOperator.java
index e4203bc..a218da2 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByOrderByCombineOperator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByOrderByCombineOperator.java
@@ -23,7 +23,6 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
@@ -46,7 +45,7 @@ import
org.apache.pinot.core.query.aggregation.groupby.AggregationGroupByResult;
import org.apache.pinot.core.query.aggregation.groupby.GroupKeyGenerator;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.core.util.GroupByUtils;
-import org.apache.pinot.core.util.QueryOptions;
+import org.apache.pinot.core.util.QueryOptionsUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -79,12 +78,9 @@ public class GroupByOrderByCombineOperator extends
BaseCombineOperator {
super(operators, queryContext, executorService, endTimeMs,
maxExecutionThreads > 0 ? maxExecutionThreads : operators.size());
- Map<String, String> queryOptions = queryContext.getQueryOptions();
- if (queryOptions != null) {
- Integer minTrimSizeOption =
QueryOptions.getMinServerGroupTrimSize(queryOptions);
- if (minTrimSizeOption != null) {
- minTrimSize = minTrimSizeOption;
- }
+ Integer minTrimSizeOption =
QueryOptionsUtils.getMinServerGroupTrimSize(queryContext.getQueryOptions());
+ if (minTrimSizeOption != null) {
+ minTrimSize = minTrimSizeOption;
}
if (minTrimSize > 0) {
int limit = queryContext.getLimit();
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationGroupByOrderByPlanNode.java
b/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationGroupByOrderByPlanNode.java
index 6cc42ca..7fb5f1f 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationGroupByOrderByPlanNode.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationGroupByOrderByPlanNode.java
@@ -30,7 +30,7 @@ import
org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.core.startree.CompositePredicateEvaluator;
import org.apache.pinot.core.startree.StarTreeUtils;
import org.apache.pinot.core.startree.plan.StarTreeTransformPlanNode;
-import org.apache.pinot.core.util.QueryOptions;
+import org.apache.pinot.core.util.QueryOptionsUtils;
import org.apache.pinot.segment.spi.IndexSegment;
import
org.apache.pinot.segment.spi.index.startree.AggregationFunctionColumnPair;
import org.apache.pinot.segment.spi.index.startree.StarTreeV2;
@@ -54,13 +54,8 @@ public class AggregationGroupByOrderByPlanNode implements
PlanNode {
_queryContext = queryContext;
_maxInitialResultHolderCapacity = maxInitialResultHolderCapacity;
_numGroupsLimit = numGroupsLimit;
- Map<String, String> queryOptions = _queryContext.getQueryOptions();
- if (queryOptions != null) {
- Integer minSegmentGroupTrimSize =
QueryOptions.getMinSegmentGroupTrimSize(queryOptions);
- _minGroupTrimSize = minSegmentGroupTrimSize != null ?
minSegmentGroupTrimSize : minGroupTrimSize;
- } else {
- _minGroupTrimSize = minGroupTrimSize;
- }
+ Integer minSegmentGroupTrimSize =
QueryOptionsUtils.getMinSegmentGroupTrimSize(_queryContext.getQueryOptions());
+ _minGroupTrimSize = minSegmentGroupTrimSize != null ?
minSegmentGroupTrimSize : minGroupTrimSize;
}
@Override
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java
b/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java
index c9ac5f7..09125ba 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java
@@ -22,7 +22,6 @@ import io.grpc.stub.StreamObserver;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.Phaser;
@@ -41,7 +40,7 @@ import
org.apache.pinot.core.operator.combine.SelectionOrderByCombineOperator;
import
org.apache.pinot.core.operator.streaming.StreamingSelectionOnlyCombineOperator;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.core.query.request.context.utils.QueryContextUtils;
-import org.apache.pinot.core.util.QueryOptions;
+import org.apache.pinot.core.util.QueryOptionsUtils;
import org.apache.pinot.core.util.trace.TraceCallable;
import org.apache.pinot.spi.exception.BadQueryRequestException;
@@ -187,8 +186,7 @@ public class CombinePlanNode implements PlanNode {
_maxExecutionThreads);
} else {
// Aggregation group-by
- Map<String, String> queryOptions = _queryContext.getQueryOptions();
- if (queryOptions != null &&
QueryOptions.isGroupByModeSQL(queryOptions)) {
+ if
(QueryOptionsUtils.isGroupByModeSQL(_queryContext.getQueryOptions())) {
return new GroupByOrderByCombineOperator(operators, _queryContext,
_executorService, _endTimeMs,
_maxExecutionThreads, _minGroupTrimSize, _groupTrimThreshold);
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/plan/FilterPlanNode.java
b/pinot-core/src/main/java/org/apache/pinot/core/plan/FilterPlanNode.java
index 4418e1d..ce06d79 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/plan/FilterPlanNode.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/FilterPlanNode.java
@@ -45,7 +45,7 @@ import
org.apache.pinot.core.operator.filter.predicate.FSTBasedRegexpPredicateEv
import org.apache.pinot.core.operator.filter.predicate.PredicateEvaluator;
import
org.apache.pinot.core.operator.filter.predicate.PredicateEvaluatorProvider;
import org.apache.pinot.core.query.request.context.QueryContext;
-import org.apache.pinot.core.util.QueryOptions;
+import org.apache.pinot.core.util.QueryOptionsUtils;
import
org.apache.pinot.segment.local.segment.index.datasource.MutableDataSource;
import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.segment.spi.datasource.DataSource;
@@ -72,10 +72,7 @@ public class FilterPlanNode implements PlanNode {
public BaseFilterOperator run() {
FilterContext filter = _queryContext.getFilter();
ThreadSafeMutableRoaringBitmap validDocIds =
_indexSegment.getValidDocIds();
- boolean upsertSkipped = false;
- if (_queryContext.getQueryOptions() != null) {
- upsertSkipped = new
QueryOptions(_queryContext.getQueryOptions()).isSkipUpsert();
- }
+ boolean upsertSkipped =
QueryOptionsUtils.isSkipUpsert(_queryContext.getQueryOptions());
if (filter != null) {
BaseFilterOperator filterOperator = constructPhysicalOperator(filter,
_queryContext.getDebugOptions());
if (validDocIds != null && !upsertSkipped) {
@@ -188,12 +185,11 @@ public class FilterPlanNode implements PlanNode {
// similar to that of FSTBasedEvaluator, else use regular flow
of getting predicate evaluator.
PredicateEvaluator evaluator;
if (dataSource.getFSTIndex() != null) {
- evaluator = FSTBasedRegexpPredicateEvaluatorFactory
- .newFSTBasedEvaluator(dataSource.getFSTIndex(),
dataSource.getDictionary(),
- ((RegexpLikePredicate) predicate).getValue());
+ evaluator =
FSTBasedRegexpPredicateEvaluatorFactory.newFSTBasedEvaluator(dataSource.getFSTIndex(),
+ dataSource.getDictionary(), ((RegexpLikePredicate)
predicate).getValue());
} else if (dataSource instanceof MutableDataSource &&
((MutableDataSource) dataSource).isFSTEnabled()) {
- evaluator = FSTBasedRegexpPredicateEvaluatorFactory
- .newAutomatonBasedEvaluator(dataSource.getDictionary(),
+ evaluator =
+
FSTBasedRegexpPredicateEvaluatorFactory.newAutomatonBasedEvaluator(dataSource.getDictionary(),
((RegexpLikePredicate) predicate).getValue());
} else {
evaluator =
PredicateEvaluatorProvider.getPredicateEvaluator(predicate,
dataSource.getDictionary(),
@@ -202,8 +198,8 @@ public class FilterPlanNode implements PlanNode {
return FilterOperatorUtils.getLeafFilterOperator(evaluator,
dataSource, _numDocs);
case JSON_MATCH:
JsonIndexReader jsonIndex = dataSource.getJsonIndex();
- Preconditions
- .checkState(jsonIndex != null, "Cannot apply JSON_MATCH on
column: %s without json index", column);
+ Preconditions.checkState(jsonIndex != null, "Cannot apply
JSON_MATCH on column: %s without json index",
+ column);
return new JsonMatchFilterOperator(jsonIndex,
((JsonMatchPredicate) predicate).getValue(), _numDocs);
case IS_NULL:
NullValueVectorReader nullValueVector =
dataSource.getNullValueVector();
@@ -220,8 +216,8 @@ public class FilterPlanNode implements PlanNode {
return new MatchAllFilterOperator(_numDocs);
}
default:
- PredicateEvaluator predicateEvaluator =
PredicateEvaluatorProvider
- .getPredicateEvaluator(predicate, dataSource.getDictionary(),
+ PredicateEvaluator predicateEvaluator =
+ PredicateEvaluatorProvider.getPredicateEvaluator(predicate,
dataSource.getDictionary(),
dataSource.getDataSourceMetadata().getDataType());
return
FilterOperatorUtils.getLeafFilterOperator(predicateEvaluator, dataSource,
_numDocs);
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java
b/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java
index a54be41..90eee3c 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java
@@ -24,7 +24,6 @@ import io.grpc.stub.StreamObserver;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
@@ -46,7 +45,7 @@ import org.apache.pinot.core.query.config.QueryExecutorConfig;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.core.query.request.context.utils.QueryContextUtils;
import org.apache.pinot.core.util.GroupByUtils;
-import org.apache.pinot.core.util.QueryOptions;
+import org.apache.pinot.core.util.QueryOptionsUtils;
import org.apache.pinot.segment.spi.FetchContext;
import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.spi.env.PinotConfiguration;
@@ -185,16 +184,13 @@ public class InstancePlanMakerImplV2 implements PlanMaker
{
}
private int getMaxExecutionThreads(QueryContext queryContext) {
- Map<String, String> queryOptions = queryContext.getQueryOptions();
- if (queryOptions != null) {
- Integer maxExecutionThreadsFromQuery =
QueryOptions.getMaxExecutionThreads(queryOptions);
- if (maxExecutionThreadsFromQuery != null && maxExecutionThreadsFromQuery
> 0) {
- // Do not allow query to override the execution threads over the
instance-level limit
- if (_maxExecutionThreads > 0) {
- return Math.min(_maxExecutionThreads, maxExecutionThreadsFromQuery);
- } else {
- return maxExecutionThreadsFromQuery;
- }
+ Integer maxExecutionThreadsFromQuery =
QueryOptionsUtils.getMaxExecutionThreads(queryContext.getQueryOptions());
+ if (maxExecutionThreadsFromQuery != null && maxExecutionThreadsFromQuery >
0) {
+ // Do not allow query to override the execution threads over the
instance-level limit
+ if (_maxExecutionThreads > 0) {
+ return Math.min(_maxExecutionThreads, maxExecutionThreadsFromQuery);
+ } else {
+ return maxExecutionThreadsFromQuery;
}
}
return _maxExecutionThreads;
@@ -206,8 +202,7 @@ public class InstancePlanMakerImplV2 implements PlanMaker {
List<ExpressionContext> groupByExpressions =
queryContext.getGroupByExpressions();
if (groupByExpressions != null) {
// Aggregation group-by query
- Map<String, String> queryOptions = queryContext.getQueryOptions();
- if (queryOptions != null &&
QueryOptions.isGroupByModeSQL(queryOptions)) {
+ if
(QueryOptionsUtils.isGroupByModeSQL(queryContext.getQueryOptions())) {
return new AggregationGroupByOrderByPlanNode(indexSegment,
queryContext, _maxInitialResultHolderCapacity,
_numGroupsLimit, _minSegmentGroupTrimSize);
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/executor/ServerQueryExecutorV1Impl.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/executor/ServerQueryExecutorV1Impl.java
index 28e3db8..4af6693 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/executor/ServerQueryExecutorV1Impl.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/executor/ServerQueryExecutorV1Impl.java
@@ -53,7 +53,7 @@ import
org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.core.query.request.context.TimerContext;
import
org.apache.pinot.core.query.request.context.utils.QueryContextConverterUtils;
import org.apache.pinot.core.query.utils.idset.IdSet;
-import org.apache.pinot.core.util.QueryOptions;
+import org.apache.pinot.core.util.QueryOptionsUtils;
import org.apache.pinot.core.util.trace.TraceContext;
import org.apache.pinot.segment.local.data.manager.SegmentDataManager;
import org.apache.pinot.segment.local.data.manager.TableDataManager;
@@ -125,19 +125,16 @@ public class ServerQueryExecutorV1Impl implements
QueryExecutor {
LOGGER.debug("Incoming request Id: {}, query: {}", requestId,
queryContext);
// Use the timeout passed from the request if exists, or the
instance-level timeout
long queryTimeoutMs = _defaultTimeOutMs;
- Map<String, String> queryOptions = queryContext.getQueryOptions();
- if (queryOptions != null) {
- Long timeoutFromQueryOptions = QueryOptions.getTimeoutMs(queryOptions);
- if (timeoutFromQueryOptions != null) {
- queryTimeoutMs = timeoutFromQueryOptions;
- }
+ Long timeoutFromQueryOptions =
QueryOptionsUtils.getTimeoutMs(queryContext.getQueryOptions());
+ if (timeoutFromQueryOptions != null) {
+ queryTimeoutMs = timeoutFromQueryOptions;
}
// Query scheduler wait time already exceeds query timeout, directly return
if (querySchedulingTimeMs >= queryTimeoutMs) {
_serverMetrics.addMeteredTableValue(tableNameWithType,
ServerMeter.SCHEDULING_TIMEOUT_EXCEPTIONS, 1);
- String errorMessage = String
- .format("Query scheduling took %dms (longer than query timeout of
%dms)", querySchedulingTimeMs,
+ String errorMessage =
+ String.format("Query scheduling took %dms (longer than query timeout
of %dms)", querySchedulingTimeMs,
queryTimeoutMs);
DataTable dataTable = DataTableBuilder.getEmptyDataTable();
dataTable.addException(QueryException.getException(QueryException.QUERY_SCHEDULING_TIMEOUT_ERROR,
errorMessage));
@@ -281,9 +278,10 @@ public class ServerQueryExecutorV1Impl implements
QueryExecutor {
return dataTable;
} else {
TimerContext.Timer planBuildTimer =
timerContext.startNewPhaseTimer(ServerQueryPhase.BUILD_QUERY_PLAN);
- Plan queryPlan = enableStreaming ? _planMaker
- .makeStreamingInstancePlan(selectedSegments, queryContext,
executorService, responseObserver, endTimeMs)
- : _planMaker.makeInstancePlan(selectedSegments, queryContext,
executorService, endTimeMs);
+ Plan queryPlan =
+ enableStreaming ?
_planMaker.makeStreamingInstancePlan(selectedSegments, queryContext,
executorService,
+ responseObserver, endTimeMs)
+ : _planMaker.makeInstancePlan(selectedSegments, queryContext,
executorService, endTimeMs);
planBuildTimer.stopAndRecord();
TimerContext.Timer planExecTimer =
timerContext.startNewPhaseTimer(ServerQueryPhase.QUERY_PLAN_EXECUTION);
@@ -343,8 +341,8 @@ public class ServerQueryExecutorV1Impl implements
QueryExecutor {
}
List<ExpressionContext> arguments = function.getArguments();
if (StringUtils.remove(function.getFunctionName(),
'_').equalsIgnoreCase(IN_PARTITIONED_SUBQUERY)) {
- Preconditions
- .checkState(arguments.size() == 2, "IN_PARTITIONED_SUBQUERY requires
2 arguments: expression, subquery");
+ Preconditions.checkState(arguments.size() == 2,
+ "IN_PARTITIONED_SUBQUERY requires 2 arguments: expression,
subquery");
ExpressionContext subqueryExpression = arguments.get(1);
Preconditions.checkState(subqueryExpression.getType() ==
ExpressionContext.Type.LITERAL,
"Second argument of IN_PARTITIONED_SUBQUERY must be a literal
(subquery)");
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java
index 47ed2d8..8cccacc 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java
@@ -34,7 +34,7 @@ import
org.apache.pinot.core.query.aggregation.function.AggregationFunction;
import
org.apache.pinot.core.query.aggregation.function.AggregationFunctionUtils;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.core.transport.ServerRoutingInstance;
-import org.apache.pinot.core.util.QueryOptions;
+import org.apache.pinot.core.util.QueryOptionsUtils;
/**
@@ -50,9 +50,9 @@ public class AggregationDataTableReducer implements
DataTableReducer {
AggregationDataTableReducer(QueryContext queryContext) {
_queryContext = queryContext;
_aggregationFunctions = queryContext.getAggregationFunctions();
- QueryOptions queryOptions = new
QueryOptions(queryContext.getQueryOptions());
- _preserveType = queryOptions.isPreserveType();
- _responseFormatSql = queryOptions.isResponseFormatSQL();
+ Map<String, String> queryOptions = queryContext.getQueryOptions();
+ _preserveType = QueryOptionsUtils.isPreserveType(queryOptions);
+ _responseFormatSql = QueryOptionsUtils.isResponseFormatSQL(queryOptions);
}
/**
@@ -145,8 +145,8 @@ public class AggregationDataTableReducer implements
DataTableReducer {
} else {
// Format the values into strings
for (int i = 0; i < numAggregationFunctions; i++) {
- aggregationResults.add(new AggregationResult(columnNames[i],
AggregationFunctionUtils
-
.formatValue(_aggregationFunctions[i].getFinalResultColumnType().format(finalResults[i]))));
+ aggregationResults.add(new AggregationResult(columnNames[i],
AggregationFunctionUtils.formatValue(
+
_aggregationFunctions[i].getFinalResultColumnType().format(finalResults[i]))));
}
}
return aggregationResults;
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/DistinctDataTableReducer.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/DistinctDataTableReducer.java
index b9afbe1..a28ff00 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/DistinctDataTableReducer.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/DistinctDataTableReducer.java
@@ -37,7 +37,7 @@ import
org.apache.pinot.core.query.aggregation.function.DistinctAggregationFunct
import org.apache.pinot.core.query.distinct.DistinctTable;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.core.transport.ServerRoutingInstance;
-import org.apache.pinot.core.util.QueryOptions;
+import org.apache.pinot.core.util.QueryOptionsUtils;
/**
@@ -50,7 +50,7 @@ public class DistinctDataTableReducer implements
DataTableReducer {
// TODO: queryOptions.isPreserveType() is ignored for DISTINCT queries.
DistinctDataTableReducer(QueryContext queryContext,
DistinctAggregationFunction distinctAggregationFunction) {
_distinctAggregationFunction = distinctAggregationFunction;
- _responseFormatSql = new
QueryOptions(queryContext.getQueryOptions()).isResponseFormatSQL();
+ _responseFormatSql =
QueryOptionsUtils.isResponseFormatSQL(queryContext.getQueryOptions());
}
/**
@@ -88,8 +88,8 @@ public class DistinctDataTableReducer implements
DataTableReducer {
int numColumns = columns.length;
ColumnDataType[] columnDataTypes = new ColumnDataType[numColumns];
Arrays.fill(columnDataTypes, ColumnDataType.STRING);
- brokerResponseNative
- .setResultTable(new ResultTable(new DataSchema(columns,
columnDataTypes), Collections.emptyList()));
+ brokerResponseNative.setResultTable(
+ new ResultTable(new DataSchema(columns, columnDataTypes),
Collections.emptyList()));
} else {
brokerResponseNative.setSelectionResults(new
SelectionResults(Arrays.asList(columns), Collections.emptyList()));
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/GroupByDataTableReducer.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/GroupByDataTableReducer.java
index 5363758..fd668c7 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/GroupByDataTableReducer.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/GroupByDataTableReducer.java
@@ -55,7 +55,7 @@ import
org.apache.pinot.core.query.aggregation.groupby.AggregationGroupByTrimmin
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.core.transport.ServerRoutingInstance;
import org.apache.pinot.core.util.GroupByUtils;
-import org.apache.pinot.core.util.QueryOptions;
+import org.apache.pinot.core.util.QueryOptionsUtils;
import org.apache.pinot.core.util.trace.TraceRunnable;
@@ -86,10 +86,10 @@ public class GroupByDataTableReducer implements
DataTableReducer {
assert _groupByExpressions != null;
_numGroupByExpressions = _groupByExpressions.size();
_numColumns = _numAggregationFunctions + _numGroupByExpressions;
- QueryOptions queryOptions = new
QueryOptions(queryContext.getQueryOptions());
- _preserveType = queryOptions.isPreserveType();
- _groupByModeSql = queryOptions.isGroupByModeSQL();
- _responseFormatSql = queryOptions.isResponseFormatSQL();
+ Map<String, String> queryOptions = queryContext.getQueryOptions();
+ _preserveType = QueryOptionsUtils.isPreserveType(queryOptions);
+ _groupByModeSql = QueryOptionsUtils.isGroupByModeSQL(queryOptions);
+ _responseFormatSql = QueryOptionsUtils.isResponseFormatSQL(queryOptions);
_sqlQuery = queryContext.getBrokerRequest().getPinotQuery() != null;
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/SelectionDataTableReducer.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/SelectionDataTableReducer.java
index 51c573d..fe09051 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/SelectionDataTableReducer.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/SelectionDataTableReducer.java
@@ -36,7 +36,7 @@ import
org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.core.query.selection.SelectionOperatorService;
import org.apache.pinot.core.query.selection.SelectionOperatorUtils;
import org.apache.pinot.core.transport.ServerRoutingInstance;
-import org.apache.pinot.core.util.QueryOptions;
+import org.apache.pinot.core.util.QueryOptionsUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,9 +54,9 @@ public class SelectionDataTableReducer implements
DataTableReducer {
SelectionDataTableReducer(QueryContext queryContext) {
_queryContext = queryContext;
- QueryOptions queryOptions = new
QueryOptions(queryContext.getQueryOptions());
- _preserveType = queryOptions.isPreserveType();
- _responseFormatSql = queryOptions.isResponseFormatSQL();
+ Map<String, String> queryOptions = queryContext.getQueryOptions();
+ _preserveType = QueryOptionsUtils.isPreserveType(queryOptions);
+ _responseFormatSql = QueryOptionsUtils.isResponseFormatSQL(queryOptions);
}
/**
@@ -89,8 +89,8 @@ public class SelectionDataTableReducer implements
DataTableReducer {
brokerMetrics.addMeteredTableValue(TableNameBuilder.extractRawTableName(tableName),
BrokerMeter.RESPONSE_MERGE_EXCEPTIONS, 1L);
}
- brokerResponseNative
- .addToExceptions(new
QueryProcessingException(QueryException.MERGE_RESPONSE_ERROR_CODE,
errorMessage));
+ brokerResponseNative.addToExceptions(
+ new
QueryProcessingException(QueryException.MERGE_RESPONSE_ERROR_CODE,
errorMessage));
}
}
@@ -112,8 +112,9 @@ public class SelectionDataTableReducer implements
DataTableReducer {
brokerResponseNative.setResultTable(
SelectionOperatorUtils.renderResultTableWithoutOrdering(reducedRows,
dataSchema, selectionColumns));
} else {
- brokerResponseNative.setSelectionResults(SelectionOperatorUtils
- .renderSelectionResultsWithoutOrdering(reducedRows, dataSchema,
selectionColumns, _preserveType));
+ brokerResponseNative.setSelectionResults(
+
SelectionOperatorUtils.renderSelectionResultsWithoutOrdering(reducedRows,
dataSchema, selectionColumns,
+ _preserveType));
}
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java
index 2b044e1..d81abaa 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java
@@ -87,7 +87,7 @@ public class QueryContext {
private QueryContext(String tableName, List<ExpressionContext>
selectExpressions, List<String> aliasList,
@Nullable FilterContext filter, @Nullable List<ExpressionContext>
groupByExpressions,
@Nullable FilterContext havingFilter, @Nullable
List<OrderByExpressionContext> orderByExpressions, int limit,
- int offset, @Nullable Map<String, String> queryOptions, @Nullable
Map<String, String> debugOptions,
+ int offset, Map<String, String> queryOptions, @Nullable Map<String,
String> debugOptions,
BrokerRequest brokerRequest) {
_tableName = tableName;
_selectExpressions = selectExpressions;
@@ -171,9 +171,8 @@ public class QueryContext {
}
/**
- * Returns the query options of the query, or {@code null} if not exist.
+ * Returns the query options of the query.
*/
- @Nullable
public Map<String, String> getQueryOptions() {
return _queryOptions;
}
@@ -306,6 +305,9 @@ public class QueryContext {
public QueryContext build() {
// TODO: Add validation logic here
+ if (_queryOptions == null) {
+ _queryOptions = Collections.emptyMap();
+ }
QueryContext queryContext =
new QueryContext(_tableName, _selectExpressions, _aliasList,
_filter, _groupByExpressions, _havingFilter,
_orderByExpressions, _limit, _offset, _queryOptions,
_debugOptions, _brokerRequest);
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/util/QueryOptions.java
b/pinot-core/src/main/java/org/apache/pinot/core/util/QueryOptionsUtils.java
similarity index 67%
rename from
pinot-core/src/main/java/org/apache/pinot/core/util/QueryOptions.java
rename to
pinot-core/src/main/java/org/apache/pinot/core/util/QueryOptionsUtils.java
index 5e2c960..dff1326 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/util/QueryOptions.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/util/QueryOptionsUtils.java
@@ -25,50 +25,10 @@ import
org.apache.pinot.spi.utils.CommonConstants.Broker.Request;
/**
- * Wrapper class to read query options
+ * Utils to parse query options.
*/
-public class QueryOptions {
- private final Long _timeoutMs;
- private final boolean _groupByModeSQL;
- private final boolean _responseFormatSQL;
- private final boolean _preserveType;
- private final boolean _skipUpsert;
-
- public QueryOptions(@Nullable Map<String, String> queryOptions) {
- if (queryOptions != null) {
- _timeoutMs = getTimeoutMs(queryOptions);
- _groupByModeSQL = isGroupByModeSQL(queryOptions);
- _responseFormatSQL =
Request.SQL.equalsIgnoreCase(queryOptions.get(Request.QueryOptionKey.RESPONSE_FORMAT));
- _preserveType =
Boolean.parseBoolean(queryOptions.get(Request.QueryOptionKey.PRESERVE_TYPE));
- _skipUpsert =
Boolean.parseBoolean(queryOptions.get(Request.QueryOptionKey.SKIP_UPSERT));
- } else {
- _timeoutMs = null;
- _groupByModeSQL = false;
- _responseFormatSQL = false;
- _preserveType = false;
- _skipUpsert = false;
- }
- }
-
- @Nullable
- public Long getTimeoutMs() {
- return _timeoutMs;
- }
-
- public boolean isGroupByModeSQL() {
- return _groupByModeSQL;
- }
-
- public boolean isResponseFormatSQL() {
- return _responseFormatSQL;
- }
-
- public boolean isPreserveType() {
- return _preserveType;
- }
-
- public boolean isSkipUpsert() {
- return _skipUpsert;
+public class QueryOptionsUtils {
+ private QueryOptionsUtils() {
}
@Nullable
@@ -87,6 +47,18 @@ public class QueryOptions {
return
Request.SQL.equalsIgnoreCase(queryOptions.get(Request.QueryOptionKey.GROUP_BY_MODE));
}
+ public static boolean isResponseFormatSQL(Map<String, String> queryOptions) {
+ return
Request.SQL.equalsIgnoreCase(queryOptions.get(Request.QueryOptionKey.RESPONSE_FORMAT));
+ }
+
+ public static boolean isPreserveType(Map<String, String> queryOptions) {
+ return
Boolean.parseBoolean(queryOptions.get(Request.QueryOptionKey.PRESERVE_TYPE));
+ }
+
+ public static boolean isSkipUpsert(Map<String, String> queryOptions) {
+ return
Boolean.parseBoolean(queryOptions.get(Request.QueryOptionKey.SKIP_UPSERT));
+ }
+
@Nullable
public static Integer getMaxExecutionThreads(Map<String, String>
queryOptions) {
String maxExecutionThreadsString =
queryOptions.get(Request.QueryOptionKey.MAX_EXECUTION_THREADS);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
