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 8228534299a Reuse compiled constant string casts during planning
(#19514)
8228534299a is described below
commit 8228534299a796107b6f9f4e1e39ebc94f4f866e
Author: Xiang Fu <[email protected]>
AuthorDate: Tue Sep 15 03:00:22 2026 -0700
Reuse compiled constant string casts during planning (#19514)
---
.../pinot/perf/BenchmarkConstantCastPlanning.java | 145 ++++++
.../pinot/perf/BenchmarkScalarCastPlanning.java | 151 ++++++
.../apache/pinot/calcite/rex/PinotRexExecutor.java | 130 +++++
.../org/apache/pinot/query/QueryEnvironment.java | 12 +
.../pinot/calcite/rex/PinotRexExecutorTest.java | 532 +++++++++++++++++++++
.../pinot/query/ConstantCastPlanningTest.java | 186 +++++++
6 files changed, 1156 insertions(+)
diff --git
a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkConstantCastPlanning.java
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkConstantCastPlanning.java
new file mode 100644
index 00000000000..df2703f6eaa
--- /dev/null
+++
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkConstantCastPlanning.java
@@ -0,0 +1,145 @@
+/**
+ * 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.perf;
+
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.config.provider.StaticTableCache;
+import org.apache.pinot.query.QueryEnvironment;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+
+/// Measures SQL parsing, validation and logical optimization of timestamp
range filters, with and without a window
+/// selecting the latest version of each event. CAST, EPOCH_STRING_CAST and
TIMESTAMP_LITERAL use equivalent constants;
+/// EPOCH_MILLIS uses a LONG time column as a control without timestamp casts.
No cluster or query execution is needed.
+/// Each benchmark thread owns its environment and cycles through prebuilt
queries with different account IDs, keeping
+/// query-string construction outside the measured operation.
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@Fork(2)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@State(Scope.Thread)
+public class BenchmarkConstantCastPlanning {
+ private static final String TABLE_NAME = "events";
+ private static final int NUM_QUERIES = 1024;
+
+ @Param({"FILTER", "WINDOW"})
+ private String _queryShape;
+
+ @Param({"CAST", "EPOCH_STRING_CAST", "TIMESTAMP_LITERAL", "EPOCH_MILLIS"})
+ private String _literalForm;
+
+ private QueryEnvironment _queryEnvironment;
+ private String[] _queries;
+ private int _nextQuery;
+
+ @Setup
+ public void setUp() {
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addSingleValueDimension("account_id", DataType.LONG)
+ .addSingleValueDimension("version", DataType.LONG)
+ .addSingleValueDimension("sequence_id", DataType.LONG)
+ .addMetric("reading", DataType.DOUBLE)
+ .addDateTime("event_time", DataType.TIMESTAMP, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .addDateTime("event_time_ms", DataType.LONG, "1:MILLISECONDS:EPOCH",
"1:MILLISECONDS")
+ .build();
+ schema.setEnableColumnBasedNullHandling(true);
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
+ StaticTableCache tableCache = new StaticTableCache(List.of(tableConfig),
List.of(schema), List.of(), false);
+ _queryEnvironment = new QueryEnvironment(QueryEnvironment.configBuilder()
+ .requestId(0L)
+ .database(CommonConstants.DEFAULT_DATABASE)
+ .tableCache(tableCache)
+ .isNullHandlingEnabled(true)
+ .defaultUsePhysicalOptimizer(false)
+ .build());
+
+ String timeColumn = "event_time";
+ String lowerBound;
+ String upperBound;
+ switch (_literalForm) {
+ case "CAST":
+ lowerBound = "CAST('2024-01-01 00:00:00' AS TIMESTAMP)";
+ upperBound = "CAST('2024-02-01 00:00:00' AS TIMESTAMP)";
+ break;
+ case "EPOCH_STRING_CAST":
+ lowerBound = "CAST('1704067200000' AS TIMESTAMP)";
+ upperBound = "CAST('1706745600000' AS TIMESTAMP)";
+ break;
+ case "TIMESTAMP_LITERAL":
+ lowerBound = "TIMESTAMP '2024-01-01 00:00:00'";
+ upperBound = "TIMESTAMP '2024-02-01 00:00:00'";
+ break;
+ case "EPOCH_MILLIS":
+ timeColumn = "event_time_ms";
+ lowerBound = "1704067200000";
+ upperBound = "1706745600000";
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown literal form: " +
_literalForm);
+ }
+
+ _queries = new String[NUM_QUERIES];
+ for (int i = 0; i < NUM_QUERIES; i++) {
+ String filter = " FROM " + TABLE_NAME + " WHERE account_id = " +
(1000000L + i)
+ + " AND " + timeColumn + " >= " + lowerBound + " AND " + timeColumn
+ " < " + upperBound;
+ switch (_queryShape) {
+ case "FILTER":
+ _queries[i] = "SELECT " + timeColumn + ", reading" + filter + "
ORDER BY " + timeColumn + " LIMIT 1000";
+ break;
+ case "WINDOW":
+ _queries[i] = "SELECT " + timeColumn + ", reading FROM (SELECT " +
timeColumn
+ + ", reading, ROW_NUMBER() OVER (PARTITION BY account_id, " +
timeColumn
+ + " ORDER BY version DESC, sequence_id DESC) AS row_num" + filter
+ + ") WHERE row_num = 1 ORDER BY " + timeColumn + " LIMIT 1000";
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown query shape: " +
_queryShape);
+ }
+ }
+ _nextQuery = 0;
+ }
+
+ @Benchmark
+ public Set<String> compile() {
+ String query = _queries[_nextQuery];
+ _nextQuery = (_nextQuery + 1) & (NUM_QUERIES - 1);
+ try (QueryEnvironment.CompiledQuery compiledQuery =
_queryEnvironment.compile(query)) {
+ return compiledQuery.getTableNames();
+ }
+ }
+}
diff --git
a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkScalarCastPlanning.java
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkScalarCastPlanning.java
new file mode 100644
index 00000000000..a3c7890d21a
--- /dev/null
+++
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkScalarCastPlanning.java
@@ -0,0 +1,151 @@
+/**
+ * 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.perf;
+
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.config.provider.StaticTableCache;
+import org.apache.pinot.query.QueryEnvironment;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+
+/// Measures SQL parsing, validation and logical optimization of scalar
filters using constant string casts or
+/// equivalent numeric/boolean literals. Predicates compare each constant with
a matching column type. DECIMAL uses
+/// the schema's default precision and scale. No cluster or query execution is
needed.
+/// Each thread owns its environment and cycles through 1,024 prebuilt
queries. Numeric cast values vary with each
+/// query; boolean values alternate, with a varying account filter keeping the
queries distinct. Query construction
+/// stays outside the measured operation, and warmup populates reusable
conversion templates.
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@Fork(2)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@State(Scope.Thread)
+public class BenchmarkScalarCastPlanning {
+ private static final String TABLE_NAME = "scalar_events";
+ private static final int NUM_QUERIES = 1024;
+
+ @Param({"INTEGER", "BIGINT", "DOUBLE", "DECIMAL", "BOOLEAN"})
+ private String _dataType;
+
+ @Param({"CAST", "LITERAL"})
+ private String _literalForm;
+
+ private QueryEnvironment _queryEnvironment;
+ private String[] _queries;
+ private int _nextQuery;
+
+ @Setup
+ public void setUp() {
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addSingleValueDimension("account_id", DataType.LONG)
+ .addSingleValueDimension("int_value", DataType.INT)
+ .addSingleValueDimension("long_value", DataType.LONG)
+ .addSingleValueDimension("double_value", DataType.DOUBLE)
+ .addSingleValueDimension("decimal_value", DataType.BIG_DECIMAL)
+ .addSingleValueDimension("boolean_value", DataType.BOOLEAN)
+ .build();
+ schema.setEnableColumnBasedNullHandling(true);
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build();
+ StaticTableCache tableCache = new StaticTableCache(List.of(tableConfig),
List.of(schema), List.of(), false);
+ _queryEnvironment = new QueryEnvironment(QueryEnvironment.configBuilder()
+ .requestId(0L)
+ .database(CommonConstants.DEFAULT_DATABASE)
+ .tableCache(tableCache)
+ .isNullHandlingEnabled(true)
+ .defaultUsePhysicalOptimizer(false)
+ .build());
+
+ _queries = new String[NUM_QUERIES];
+ for (int i = 0; i < NUM_QUERIES; i++) {
+ String column;
+ String value;
+ String literal;
+ switch (_dataType) {
+ case "INTEGER":
+ column = "int_value";
+ value = Integer.toString(1000000 + i);
+ literal = value;
+ break;
+ case "BIGINT":
+ column = "long_value";
+ value = Long.toString(5000000000L + i);
+ literal = value;
+ break;
+ case "DOUBLE":
+ column = "double_value";
+ value = (1000000 + i) + ".25";
+ literal = value + "E0";
+ break;
+ case "DECIMAL":
+ column = "decimal_value";
+ value = (1000000 + i) + ".25";
+ literal = value;
+ break;
+ case "BOOLEAN":
+ column = "boolean_value";
+ value = (i & 1) == 0 ? "true" : "false";
+ literal = value;
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown data type: " +
_dataType);
+ }
+ String constant;
+ switch (_literalForm) {
+ case "CAST":
+ constant = "CAST('" + value + "' AS " + _dataType + ")";
+ break;
+ case "LITERAL":
+ constant = literal;
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown literal form: " +
_literalForm);
+ }
+ _queries[i] = "SELECT account_id, " + column + " FROM " + TABLE_NAME + "
WHERE " + column + " = " + constant
+ + " AND account_id >= " + (1000000L + i) + " ORDER BY account_id
LIMIT 1000";
+ }
+ _nextQuery = 0;
+ }
+
+ @Benchmark
+ public Set<String> compile() {
+ String query = _queries[_nextQuery];
+ _nextQuery = (_nextQuery + 1) & (NUM_QUERIES - 1);
+ try (QueryEnvironment.CompiledQuery compiledQuery =
_queryEnvironment.compile(query)) {
+ return compiledQuery.getTableNames();
+ }
+ }
+}
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rex/PinotRexExecutor.java
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rex/PinotRexExecutor.java
new file mode 100644
index 00000000000..5d4757f0ec3
--- /dev/null
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rex/PinotRexExecutor.java
@@ -0,0 +1,130 @@
+/**
+ * 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.calcite.rex;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.calcite.DataContext;
+import org.apache.calcite.DataContexts;
+import org.apache.calcite.linq4j.function.Function1;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeSystem;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexExecutor;
+import org.apache.calcite.rex.RexExecutorImpl;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.type.SqlTypeUtil;
+
+
+/// Reduces string literal casts using bounded, reusable Calcite executables.
Each source/target type pair is compiled
+/// with an input reference instead of a literal, so different values share
Calcite's conversion code. Other expressions
+/// use the fallback executor. Cached functions are immutable; input values
and results stay local to each call, making
+/// this executor thread-safe without sharing RexExecutable's mutable data
context.
+public final class PinotRexExecutor implements RexExecutor {
+ public static final PinotRexExecutor INSTANCE = new
PinotRexExecutor(RexUtil.EXECUTOR);
+
+ private final RexExecutor _fallback;
+ private final Cache<CastSignature, Function1<DataContext, Object[]>> _casts;
+
+ @VisibleForTesting
+ PinotRexExecutor(RexExecutor fallback) {
+ this(fallback, 256);
+ }
+
+ @VisibleForTesting
+ PinotRexExecutor(RexExecutor fallback, int cacheSize) {
+ _fallback = fallback;
+ _casts = CacheBuilder.newBuilder().maximumSize(cacheSize).build();
+ }
+
+ @Override
+ public void reduce(RexBuilder rexBuilder, List<RexNode> constExps,
List<RexNode> reducedValues) {
+ // Calcite treats reduction as all-or-nothing. Delegate the entire batch
if any expression is unsupported, so a
+ // failure in the fallback cannot leave the batch partially reduced.
+ for (RexNode expression : constExps) {
+ if (!isSupportedStringCast(expression)) {
+ _fallback.reduce(rexBuilder, constExps, reducedValues);
+ return;
+ }
+ }
+ List<RexNode> literals = new ArrayList<>(constExps.size());
+ try {
+ for (RexNode expression : constExps) {
+ RexLiteral operand = (RexLiteral) ((RexCall)
expression).getOperands().get(0);
+ Object value = null;
+ if (!operand.isNull()) {
+ Function1<DataContext, Object[]> cast = getCast(rexBuilder,
operand.getType(), expression.getType());
+ DataContext context = DataContexts.of(Map.of("inputRecord", new
Object[]{RexLiteral.stringValue(operand)}));
+ value = cast.apply(context)[0];
+ }
+ literals.add(rexBuilder.makeLiteral(value, expression.getType(),
true));
+ }
+ } catch (RuntimeException | ExecutionException e) {
+ // Like Calcite, retain the entire batch if conversion, compilation or
literal construction fails. In particular,
+ // invalid epoch strings must remain available for Pinot's later
timestamp conversion.
+ reducedValues.addAll(constExps);
+ return;
+ }
+ reducedValues.addAll(literals);
+ }
+
+ @VisibleForTesting
+ Function1<DataContext, Object[]> getCast(RexBuilder builder, RelDataType
source, RelDataType target)
+ throws ExecutionException {
+ CastSignature signature = new CastSignature(source, target,
builder.getTypeFactory().getTypeSystem());
+ return _casts.get(signature, () -> {
+ RelDataType rowType = builder.getTypeFactory().builder().add("value",
source).build();
+ RexNode cast = builder.makeAbstractCast(target,
builder.makeInputRef(source, 0), false);
+ return RexExecutorImpl.getExecutable(builder, List.of(cast),
rowType).getFunction();
+ });
+ }
+
+ private static boolean isSupportedStringCast(RexNode expression) {
+ if (!(expression instanceof RexCall)) {
+ return false;
+ }
+ RexCall call = (RexCall) expression;
+ if (call.getOperator() != SqlStdOperatorTable.CAST ||
call.getOperands().size() != 1) {
+ return false;
+ }
+ // Keep timezone-dependent and structured conversions in the fallback's
own data context.
+ RelDataType target = call.getType();
+ if (!SqlTypeUtil.isAtomic(target) ||
SqlTypeName.TZ_TYPES.contains(target.getSqlTypeName())
+ || target.getSqlTypeName() == SqlTypeName.VARIANT) {
+ return false;
+ }
+ RexNode operand = call.getOperands().get(0);
+ return operand instanceof RexLiteral &&
SqlTypeUtil.isCharacter(operand.getType());
+ }
+
+ /// Full types retain precision, scale, nullability, charset and collation;
the type system supplies rounding rules.
+ /// Literal values never enter the key or the generated code.
+ private record CastSignature(RelDataType source, RelDataType target,
RelDataTypeSystem typeSystem) {
+ }
+}
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java
b/pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java
index bfa24cac481..3f30e35547a 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java
@@ -44,6 +44,7 @@ import org.apache.calcite.prepare.CalciteCatalogReader;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.RelRoot;
import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexExecutor;
import org.apache.calcite.runtime.CalciteContextException;
import org.apache.calcite.sql.SqlExplain;
import org.apache.calcite.sql.SqlExplainFormat;
@@ -62,6 +63,7 @@ import
org.apache.pinot.calcite.rel.rules.PinotJoinToDynamicBroadcastRule;
import org.apache.pinot.calcite.rel.rules.PinotRelDistributionTraitRule;
import org.apache.pinot.calcite.rel.rules.PinotRuleUtils;
import org.apache.pinot.calcite.rel.rules.PinotSortExchangeCopyRule;
+import org.apache.pinot.calcite.rex.PinotRexExecutor;
import org.apache.pinot.calcite.sql.fun.PinotOperatorTable;
import org.apache.pinot.calcite.sql2rel.PinotConvertletTable;
import org.apache.pinot.calcite.sql2rel.PinotRelDecorrelator;
@@ -446,6 +448,13 @@ public class QueryEnvironment {
///
/// It is important to notice that the returned tree is not yet
[optimized][#optimize(RelRoot, PlannerContext)].
private RelRoot toRelation(SqlNode sqlNode, PlannerContext plannerContext) {
+ RelOptPlanner planner = plannerContext.getRelOptPlanner();
+ RexExecutor originalExecutor = planner.getExecutor();
+ if (originalExecutor == null) {
+ // SqlToRelConverter transforms its RelBuilder, discarding executors
provided only through the builder context.
+ // Install on the per-query planner so conversion and field trimming can
reuse compiled cast templates.
+ planner.setExecutor(PinotRexExecutor.INSTANCE);
+ }
try {
RexBuilder rexBuilder = new RexBuilder(_typeFactory);
RelOptCluster cluster =
RelOptCluster.create(plannerContext.getRelOptPlanner(), rexBuilder);
@@ -481,6 +490,9 @@ public class QueryEnvironment {
} catch (Throwable e) {
throw QueryErrorCode.QUERY_PLANNING.asException(
"Error converting query to relational expression: " +
e.getMessage(), e);
+ } finally {
+ // Keep the executor policy of the subsequent optimization phase
unchanged.
+ planner.setExecutor(originalExecutor);
}
}
diff --git
a/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rex/PinotRexExecutorTest.java
b/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rex/PinotRexExecutorTest.java
new file mode 100644
index 00000000000..89351688c6b
--- /dev/null
+++
b/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rex/PinotRexExecutorTest.java
@@ -0,0 +1,532 @@
+/**
+ * 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.calcite.rex;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.TimeZone;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import org.apache.calcite.DataContext;
+import org.apache.calcite.DataContexts;
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.linq4j.function.Function1;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.volcano.VolcanoPlanner;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rel.type.RelDataTypeSystemImpl;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexExecutor;
+import org.apache.calcite.rex.RexExecutorImpl;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.runtime.Hook;
+import org.apache.calcite.sql.SqlCollation;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlSpecialOperator;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.NlsString;
+import org.apache.pinot.calcite.rel.rules.PinotRuleUtils;
+import org.apache.pinot.query.type.TypeFactory;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotSame;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+
+
+/// Compares scalar constant reduction with Calcite and verifies reuse of
immutable conversion templates.
+public class PinotRexExecutorTest {
+ private static final RexExecutor CALCITE_EXECUTOR = new
RexExecutorImpl(DataContexts.EMPTY);
+ private static final RexExecutor FAILING_FALLBACK = (builder, expressions,
results) -> {
+ throw new AssertionError("Supported casts must not use the fallback: " +
expressions);
+ };
+
+ @DataProvider
+ public Object[][] scalarLiterals() {
+ List<Object[]> cases = new ArrayList<>();
+ addScalarCases(cases, SqlTypeName.TINYINT, -1, -1, "127", "-128", "128",
"-129", " 42 ", "1.5", "invalid", null);
+ addScalarCases(cases, SqlTypeName.SMALLINT, -1, -1, "32767", "-32768",
"32768", "-32769", " 42 ", "invalid", null);
+ addScalarCases(cases, SqlTypeName.INTEGER, -1, -1, "2147483647",
"-2147483648", "2147483648", "-2147483649",
+ "\t42\n", "1.5", "invalid", null);
+ addScalarCases(cases, SqlTypeName.UTINYINT, -1, -1, "1", "255", "256",
"-1", null);
+ addScalarCases(cases, SqlTypeName.USMALLINT, -1, -1, "1", "65535",
"65536", "-1", null);
+ addScalarCases(cases, SqlTypeName.UINTEGER, -1, -1, "1", "4294967295",
"4294967296", "-1", null);
+ addScalarCases(cases, SqlTypeName.UBIGINT, -1, -1, "1",
"18446744073709551615", "18446744073709551616", "-1", null);
+ for (SqlTypeName target : List.of(SqlTypeName.FLOAT, SqlTypeName.REAL,
SqlTypeName.DOUBLE)) {
+ addScalarCases(cases, target, -1, -1, "1.25", "-1.5", "1e100", " 1.25 ",
"NaN", "invalid", null);
+ }
+ addScalarCases(cases, SqlTypeName.DECIMAL, 5, 2, "123.455", "-123.455",
"999.995", " 1.25 ", "invalid", null);
+ addScalarCases(cases, SqlTypeName.DECIMAL, 8, 3, "123.4555", "-123.4555",
"99999.9995", " 1.25 ", "invalid", null);
+ addScalarCases(cases, SqlTypeName.BOOLEAN, -1, -1, "true", "FALSE", " true
", "1", "invalid", null);
+ addScalarCases(cases, SqlTypeName.DATE, -1, -1, "2000-02-29",
"1969-12-31", " 2000-02-29 ", "2026-02-30",
+ "invalid", null);
+ for (int precision : List.of(0, 3)) {
+ addScalarCases(cases, SqlTypeName.TIME, precision, -1, "12:34:56.9876",
"23:59:59.9999", " 12:34:56 ",
+ "25:00:00", "invalid", null);
+ }
+ addScalarCases(cases, SqlTypeName.UUID, -1, -1,
"123e4567-e89b-12d3-a456-426614174000",
+ " 123e4567-e89b-12d3-a456-426614174000 ", "123", "invalid", null);
+ for (SqlTypeName target : List.of(SqlTypeName.CHAR, SqlTypeName.VARCHAR,
SqlTypeName.BINARY,
+ SqlTypeName.VARBINARY)) {
+ addScalarCases(cases, target, 3, -1, "ab", "abcdef", " a ", "", null);
+ }
+ return cases.toArray(new Object[0][]);
+ }
+
+ @Test(dataProvider = "scalarLiterals")
+ public void testScalarCastsMatchCalciteWithoutFallback(SqlTypeName
sourceType, SqlTypeName targetType,
+ int precision, int scale, String value, boolean validControl) {
+ LiteralRexBuilder builder = new LiteralRexBuilder();
+ RelDataTypeFactory factory = builder.getTypeFactory();
+ RelDataType target;
+ if (precision < 0) {
+ target = factory.createSqlType(targetType);
+ } else if (scale < 0) {
+ target = factory.createSqlType(targetType, precision);
+ } else {
+ target = factory.createSqlType(targetType, precision, scale);
+ }
+ target = factory.createTypeWithNullability(target, true);
+ RelDataType source = factory.createSqlType(sourceType, 64);
+ RexNode operand = value == null ? builder.makeNullLiteral(source) :
builder.stringLiteral(value, source);
+ RexNode expression = builder.makeAbstractCast(target, operand, false);
+ List<RexNode> input = List.of(expression);
+ List<RexNode> expected = reduce(CALCITE_EXECUTOR, builder, input);
+ if (SqlTypeName.UNSIGNED_TYPES.contains(targetType) && value != null) {
+ // Calcite can evaluate unsigned casts but cannot materialize their
non-null RexLiterals yet.
+ assertEquals(expected, input, "Unsigned casts must remain unchanged when
literal construction is unsupported");
+ } else if (validControl) {
+ assertFalse(expected.equals(input), "The valid control must reduce for "
+ target);
+ }
+ List<RexNode> actual = reduce(new PinotRexExecutor(FAILING_FALLBACK),
builder, input);
+ assertEquals(actual, expected);
+ assertEquals(actual.get(0).getType(), expected.get(0).getType());
+ if (value == null) {
+ assertTrue(RexLiteral.isNullLiteral(actual.get(0)));
+ } else if (expected.equals(input)) {
+ assertSame(actual.get(0), expression, "A failed cast must retain the
original expression");
+ }
+ }
+
+ private static void addScalarCases(List<Object[]> cases, SqlTypeName target,
int precision, int scale,
+ String... values) {
+ for (SqlTypeName source : List.of(SqlTypeName.CHAR, SqlTypeName.VARCHAR)) {
+ for (int i = 0; i < values.length; i++) {
+ cases.add(new Object[]{source, target, precision, scale, values[i], i
== 0});
+ }
+ }
+ }
+
+ @DataProvider
+ public Object[][] bigintLiterals() {
+ List<Object[]> cases = new ArrayList<>();
+ for (SqlTypeName sourceType : List.of(SqlTypeName.CHAR,
SqlTypeName.VARCHAR)) {
+ for (String value : new String[]{"0", "-0", "-1", "+1", "000123", " 123
", "\t123\n", "9007199254740993",
+ "9223372036854775807", "-9223372036854775808",
"9223372036854775808", "-9223372036854775809",
+ "1.0", "1e3", "", "invalid", null}) {
+ cases.add(new Object[]{sourceType, value});
+ }
+ }
+ return cases.toArray(new Object[0][]);
+ }
+
+ @Test(dataProvider = "bigintLiterals")
+ public void testBigintCastsMatchCalciteWithoutFallback(SqlTypeName
sourceType, String value) {
+ LiteralRexBuilder builder = new LiteralRexBuilder();
+ RelDataType source = builder.getTypeFactory().createSqlType(sourceType,
30);
+ RelDataType target = builder.getTypeFactory().createTypeWithNullability(
+ builder.getTypeFactory().createSqlType(SqlTypeName.BIGINT), true);
+ RexNode operand = value == null ? builder.makeNullLiteral(source) :
builder.stringLiteral(value, sourceType);
+ RexNode expression = builder.makeAbstractCast(target, operand, false);
+ List<RexNode> input = List.of(expression);
+ CountingFallback fallback = new CountingFallback(CALCITE_EXECUTOR);
+ List<RexNode> expected = reduce(CALCITE_EXECUTOR, builder, input);
+ List<RexNode> actual = reduce(new PinotRexExecutor(fallback), builder,
input);
+ assertEquals(actual, expected);
+ assertEquals(actual.get(0).getType(), expected.get(0).getType());
+ assertEquals(fallback._calls.get(), 0);
+ if (expected.equals(input)) {
+ assertSame(actual.get(0), expression);
+ }
+ }
+
+ @Test
+ public void testBigintMixedBatchesMatchCalcite() {
+ LiteralRexBuilder builder = new LiteralRexBuilder();
+ RelDataType bigint =
builder.getTypeFactory().createSqlType(SqlTypeName.BIGINT);
+ RexNode valid = builder.makeAbstractCast(bigint,
builder.makeLiteral("123"), false);
+ RexNode invalid = builder.makeAbstractCast(bigint,
builder.makeLiteral("invalid"), false);
+ RexNode timestamp = timestampCast(builder, SqlTypeName.CHAR, "2026-04-01
00:00:00", 3);
+ RexNode unsupported = builder.makeCall(SqlStdOperatorTable.PLUS,
builder.makeExactLiteral(BigDecimal.ONE),
+ builder.makeExactLiteral(BigDecimal.TEN));
+ for (List<RexNode> input : List.of(List.of(valid, invalid),
List.of(invalid, valid), List.of(valid, timestamp),
+ List.of(timestamp, valid), List.of(valid, unsupported))) {
+ CountingFallback fallback = new CountingFallback(CALCITE_EXECUTOR);
+ List<RexNode> actual = reduce(new PinotRexExecutor(fallback), builder,
input);
+ assertEquals(actual, reduce(CALCITE_EXECUTOR, builder, input));
+ assertEquals(fallback._calls.get(), input.contains(unsupported) ? 1 : 0);
+ }
+ }
+
+ @DataProvider
+ public Object[][] timestampLiterals() {
+ List<Object[]> cases = new ArrayList<>();
+ for (SqlTypeName sourceType : List.of(SqlTypeName.CHAR,
SqlTypeName.VARCHAR)) {
+ for (String value : List.of("2026-09-01 00:00:00", "1970-01-01
00:00:00", "1969-12-31 23:59:59.999",
+ "2000-02-29 12:34:56.123", "2026-09-01 12:34:56.1", "2026-09-01
12:34:56.123456789",
+ "2026-09-01 23:59:59.9999", "1969-12-31 23:59:59.9999")) {
+ cases.add(new Object[]{sourceType, value, 3});
+ }
+ for (int precision : List.of(0, 1, 2)) {
+ cases.add(new Object[]{sourceType, "1969-12-31 23:59:59.987654",
precision});
+ cases.add(new Object[]{sourceType, "2026-09-01 12:34:56.987654",
precision});
+ }
+ cases.add(new Object[]{sourceType, "2026-09-01 23:59:59.9999", 0});
+ cases.add(new Object[]{sourceType, "1969-12-31 23:59:59.9999", 0});
+ }
+ return cases.toArray(new Object[0][]);
+ }
+
+ @Test(dataProvider = "timestampLiterals")
+ public void testSupportedCastsDoNotUseFallback(SqlTypeName sourceType,
String value, int precision) {
+ LiteralRexBuilder builder = new LiteralRexBuilder();
+ RexNode expression = timestampCast(builder, sourceType, value, precision);
+ assertEquals(expression.getKind(), SqlKind.CAST);
+ List<RexNode> expected = reduce(CALCITE_EXECUTOR, builder,
List.of(expression));
+ assertFalse(expected.get(0).equals(expression), "The Calcite control must
reduce the test expression");
+
+ List<RexNode> actual = reduce(new PinotRexExecutor(FAILING_FALLBACK),
builder, List.of(expression));
+ assertEquals(actual, expected);
+ assertEquals(actual.get(0).getType(), expected.get(0).getType());
+ }
+
+ @Test
+ public void testNullCastsPreserveTypeAndNullabilityWithoutFallback() {
+ LiteralRexBuilder builder = new LiteralRexBuilder();
+ for (SqlTypeName sourceType : List.of(SqlTypeName.CHAR,
SqlTypeName.VARCHAR)) {
+ RelDataType type = builder.getTypeFactory().createSqlType(sourceType,
20);
+ RelDataType target = builder.getTypeFactory().createTypeWithNullability(
+ builder.getTypeFactory().createSqlType(SqlTypeName.TIMESTAMP, 3),
true);
+ RexNode expression = builder.makeAbstractCast(target,
builder.makeNullLiteral(type), false);
+ List<RexNode> expected = reduce(CALCITE_EXECUTOR, builder,
List.of(expression));
+ List<RexNode> actual = reduce(new PinotRexExecutor(FAILING_FALLBACK),
builder, List.of(expression));
+ assertEquals(actual, expected);
+ assertTrue(RexLiteral.isNullLiteral(actual.get(0)));
+ assertEquals(actual.get(0).getType(), expected.get(0).getType());
+ }
+ }
+
+ @DataProvider
+ public Object[][] timestampEdgeInputs() {
+ return new Object[][]{
+ {"2026-09-01"}, {" 2026-09-01 12:34:56.123 "}, {"2026-09-01
12:34:56.123 "},
+ {"\t2026-09-01 12:34:56.123\n"}, {""}, {"not-a-timestamp"},
{"2026-02-30 00:00:00"},
+ {"2026-09-01 25:00:00"}, {"2026-09-01T12:34:56Z"}, {"2026-09-01
12:34:56+02:00"},
+ {"1788266096123"}
+ };
+ }
+
+ @Test(dataProvider = "timestampEdgeInputs")
+ public void testWhitespaceAndInvalidInputsMatchCalcite(String value) {
+ LiteralRexBuilder builder = new LiteralRexBuilder();
+ RexNode expression = timestampCast(builder, SqlTypeName.CHAR, value, 3);
+ List<RexNode> input = List.of(expression);
+ List<RexNode> expected = reduce(CALCITE_EXECUTOR, builder, input);
+ CountingFallback fallback = new CountingFallback(CALCITE_EXECUTOR);
+
+ List<RexNode> actual = reduce(new PinotRexExecutor(fallback), builder,
input);
+ assertEquals(actual, expected);
+ assertEquals(fallback._calls.get(), 0, "Eligible casts must bypass the
fallback even when conversion fails");
+ if (expected.equals(input)) {
+ assertSame(actual.get(0), expression, "Failed casts must retain the
original expression");
+ }
+ }
+
+ @Test
+ public void testUnsupportedExpressionsDelegateTheWholeBatchInOrder() {
+ LiteralRexBuilder builder = new LiteralRexBuilder();
+ RexNode fast = timestampCast(builder, SqlTypeName.CHAR, "2026-09-01
12:34:56.123", 3);
+ RelDataType timestamp =
builder.getTypeFactory().createSqlType(SqlTypeName.TIMESTAMP, 3);
+ RexLiteral string = builder.stringLiteral("2026-09-01 12:34:56.123",
SqlTypeName.CHAR);
+ List<RexNode> unsupported = List.of(
+ builder.makeCall(SqlStdOperatorTable.PLUS,
builder.makeExactLiteral(BigDecimal.ONE),
+ builder.makeExactLiteral(BigDecimal.TEN)),
+ builder.makeAbstractCast(timestamp, string, true),
+ builder.makeAbstractCast(timestamp, string, false,
builder.makeLiteral("YYYY-MM-DD HH24:MI:SS.FF")),
+ builder.makeAbstractCast(timestamp, builder.makeAbstractCast(
+ builder.getTypeFactory().createSqlType(SqlTypeName.VARCHAR, 30),
string, false), false),
+ builder.makeCall(timestamp, new SqlSpecialOperator("CAST",
SqlKind.CAST), List.of(string)));
+
+ for (RexNode expression : unsupported) {
+ List<RexNode> input = List.of(fast, expression, fast);
+ CountingFallback fallback = new CountingFallback(CALCITE_EXECUTOR);
+ List<RexNode> actual = reduce(new PinotRexExecutor(fallback), builder,
input);
+ assertEquals(actual, reduce(CALCITE_EXECUTOR, builder, input));
+ assertEquals(fallback._calls.get(), 1);
+ assertSame(fallback._lastInput, input, "Fallback must receive the
original full batch");
+ assertSame(input.get(0), fast, "The immutable input must not be
modified");
+ }
+ }
+
+ @Test
+ public void testFailedCastDoesNotPartiallyReduceBatch() {
+ LiteralRexBuilder builder = new LiteralRexBuilder();
+ RexNode fast = timestampCast(builder, SqlTypeName.CHAR, "2026-09-01
12:34:56.123", 3);
+ RexNode invalid = timestampCast(builder, SqlTypeName.VARCHAR,
"not-a-timestamp", 3);
+ for (List<RexNode> input : List.of(List.of(fast, invalid),
List.of(invalid, fast))) {
+ CountingFallback fallback = new CountingFallback(CALCITE_EXECUTOR);
+ List<RexNode> actual = reduce(new PinotRexExecutor(fallback), builder,
input);
+ assertEquals(actual, reduce(CALCITE_EXECUTOR, builder, input));
+ assertEquals(actual, input, "Calcite keeps every expression unchanged
when one expression fails");
+ assertEquals(fallback._calls.get(), 0, "Failed eligible casts must not
retry through the fallback");
+ for (int i = 0; i < input.size(); i++) {
+ assertSame(actual.get(i), input.get(i), "The entire failed batch must
retain its original expressions");
+ }
+ }
+ }
+
+ @Test
+ public void testTimezoneDependentTargetsUseTheirFallbackContext() {
+ LiteralRexBuilder builder = new LiteralRexBuilder();
+ RexNode expression = builder.makeAbstractCast(
+
builder.getTypeFactory().createSqlType(SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE,
3),
+ builder.stringLiteral("2026-09-01 12:34:56.123", SqlTypeName.CHAR),
false);
+ List<RexNode> input = List.of(expression);
+ for (String zone : List.of("UTC", "America/Los_Angeles", "Asia/Kolkata")) {
+ RexExecutor baseline = new RexExecutorImpl(DataContexts.of(
+ Map.of(DataContext.Variable.TIME_ZONE.camelName,
TimeZone.getTimeZone(zone))));
+ CountingFallback fallback = new CountingFallback(baseline);
+ assertEquals(reduce(new PinotRexExecutor(fallback), builder, input),
reduce(baseline, builder, input), zone);
+ assertEquals(fallback._calls.get(), 1);
+ assertSame(fallback._lastInput, input);
+ }
+ }
+
+ @Test
+ public void testCastTemplateIsReusedAcrossLiteralValues() throws Exception {
+ LiteralRexBuilder builder = new LiteralRexBuilder();
+ RelDataType source =
builder.getTypeFactory().createSqlType(SqlTypeName.VARCHAR, 20);
+ RelDataType target =
builder.getTypeFactory().createSqlType(SqlTypeName.INTEGER);
+ PinotRexExecutor executor = new PinotRexExecutor(FAILING_FALLBACK);
+ Function1<DataContext, Object[]> template = executor.getCast(builder,
source, target);
+ for (String value : List.of("1", "2345", "-67")) {
+ List<RexNode> input = List.of(builder.makeAbstractCast(target,
builder.stringLiteral(value, source), false));
+ assertEquals(reduce(executor, builder, input), reduce(CALCITE_EXECUTOR,
builder, input));
+ assertSame(executor.getCast(new LiteralRexBuilder(), source, target),
template,
+ "Literal values and equivalent builders must reuse the same compiled
function");
+ }
+ }
+
+ @Test
+ public void testCastCacheDistinguishesFullTypesAndTypeSystem() throws
Exception {
+ LiteralRexBuilder builder = new LiteralRexBuilder();
+ RelDataTypeFactory factory = builder.getTypeFactory();
+ RelDataType source = factory.createSqlType(SqlTypeName.VARCHAR, 20);
+ RelDataType target = factory.createSqlType(SqlTypeName.DECIMAL, 8, 2);
+ PinotRexExecutor executor = new PinotRexExecutor(FAILING_FALLBACK);
+ Function1<DataContext, Object[]> template = executor.getCast(builder,
source, target);
+ SqlCollation latinCollation = new
SqlCollation(SqlCollation.Coercibility.IMPLICIT, Locale.US,
+ StandardCharsets.ISO_8859_1, "primary");
+ for (RelDataType otherSource :
List.of(factory.createSqlType(SqlTypeName.CHAR, 20),
+ factory.createSqlType(SqlTypeName.VARCHAR, 21),
factory.createTypeWithNullability(source, true),
+ factory.createTypeWithCharsetAndCollation(source,
StandardCharsets.ISO_8859_1, latinCollation))) {
+ assertNotSame(executor.getCast(builder, otherSource, target), template,
otherSource.getFullTypeString());
+ }
+ for (RelDataType otherTarget :
List.of(factory.createSqlType(SqlTypeName.DECIMAL, 9, 2),
+ factory.createSqlType(SqlTypeName.DECIMAL, 8, 3),
factory.createTypeWithNullability(target, true))) {
+ assertNotSame(executor.getCast(builder, source, otherTarget), template,
otherTarget.getFullTypeString());
+ }
+ RelDataType textTarget = factory.createSqlType(SqlTypeName.VARCHAR, 8);
+ RelDataType latinTarget =
factory.createTypeWithCharsetAndCollation(textTarget,
StandardCharsets.ISO_8859_1,
+ latinCollation);
+ assertNotSame(executor.getCast(builder, source, textTarget),
executor.getCast(builder, source, latinTarget));
+ LiteralRexBuilder otherBuilder = new LiteralRexBuilder(new
JavaTypeFactoryImpl(new RelDataTypeSystemImpl() {
+ @Override
+ public RoundingMode roundingMode() {
+ return RoundingMode.FLOOR;
+ }
+ }));
+ assertNotSame(executor.getCast(otherBuilder, source, target), template,
+ "Different type systems must not share compiled rounding semantics");
+ List<RexNode> input = List.of(builder.makeAbstractCast(target,
builder.stringLiteral("-1.239", source), false));
+ for (RexBuilder rexBuilder : List.of(builder, otherBuilder)) {
+ assertEquals(reduce(executor, rexBuilder, input),
reduce(CALCITE_EXECUTOR, rexBuilder, input));
+ }
+ assertSame(executor.getCast(builder, source, target), template);
+ }
+
+ @Test
+ public void testCastCacheEvictsOldTemplates() throws Exception {
+ LiteralRexBuilder builder = new LiteralRexBuilder();
+ RelDataTypeFactory factory = builder.getTypeFactory();
+ RelDataType source = factory.createSqlType(SqlTypeName.VARCHAR, 20);
+ RelDataType integer = factory.createSqlType(SqlTypeName.INTEGER);
+ RelDataType bigint = factory.createSqlType(SqlTypeName.BIGINT);
+ RelDataType decimal = factory.createSqlType(SqlTypeName.DECIMAL, 8, 2);
+ PinotRexExecutor executor = new PinotRexExecutor(FAILING_FALLBACK, 2);
+ Function1<DataContext, Object[]> first = executor.getCast(builder, source,
integer);
+ Function1<DataContext, Object[]> second = executor.getCast(builder,
source, bigint);
+ executor.getCast(builder, source, decimal);
+ assertSame(executor.getCast(builder, source, bigint), second, "A recently
compiled template should remain cached");
+ assertNotSame(executor.getCast(builder, source, integer), first, "The
oldest template must be evicted at capacity");
+ List<RexNode> input = List.of(builder.makeAbstractCast(integer,
builder.stringLiteral("123", source), false));
+ assertEquals(reduce(executor, builder, input), reduce(CALCITE_EXECUTOR,
builder, input));
+ }
+
+ @Test
+ public void testSharedExecutorHasNoCrossRequestState() throws Exception {
+ PinotRexExecutor executor = new PinotRexExecutor(FAILING_FALLBACK);
+ ExecutorService threads = Executors.newFixedThreadPool(4);
+ CountDownLatch ready = new CountDownLatch(4);
+ try {
+ List<Callable<Void>> tasks = new ArrayList<>();
+ for (int task = 0; task < 16; task++) {
+ LiteralRexBuilder localBuilder = new LiteralRexBuilder();
+ RelDataType source =
localBuilder.getTypeFactory().createSqlType(SqlTypeName.VARCHAR, 20);
+ RelDataType target =
localBuilder.getTypeFactory().createSqlType(SqlTypeName.INTEGER);
+ List<RexNode> input = List.of(localBuilder.makeAbstractCast(target,
+ localBuilder.stringLiteral(Integer.toString(100 + task), source),
false));
+ List<RexNode> expected = reduce(CALCITE_EXECUTOR, localBuilder, input);
+ tasks.add(() -> {
+ ready.countDown();
+ assertTrue(ready.await(30, TimeUnit.SECONDS), "All workers must
reach the fresh cache together");
+ for (int i = 0; i < 10; i++) {
+ assertEquals(reduce(executor, localBuilder, input), expected);
+ }
+ return null;
+ });
+ }
+ for (Future<Void> result : threads.invokeAll(tasks)) {
+ result.get();
+ }
+ } finally {
+ threads.shutdownNow();
+ }
+ }
+
+ @Test
+ @SuppressWarnings("try") // The resources scope the thread-local hooks;
their values are intentionally unused.
+ public void testRelBuilderTransformPreservesOptimizedExecutor() {
+ AtomicInteger generatedReductions = new AtomicInteger();
+ RelNode expected;
+ try (Hook.Closeable ignored = Hook.EXPRESSION_REDUCER.addThread(
+ (Consumer<Object>) event -> generatedReductions.incrementAndGet())) {
+ expected = projectTimestampCast(CALCITE_EXECUTOR);
+ }
+ assertTrue(generatedReductions.get() > 0, "The control must exercise
Calcite's generated reducer");
+
+ generatedReductions.set(0);
+ RelNode actual;
+ try (Hook.Closeable ignored = Hook.EXPRESSION_REDUCER.addThread(
+ (Consumer<Object>) event -> generatedReductions.incrementAndGet())) {
+ actual = projectTimestampCast(PinotRexExecutor.INSTANCE);
+ }
+ assertEquals(RelOptUtil.toString(actual), RelOptUtil.toString(expected));
+ assertEquals(generatedReductions.get(), 0, "RelBuilder.transform must
preserve the planner's optimized executor");
+ }
+
+ @Test
+ public void testRelBuilderTransformPreservesExplicitExecutor() {
+ CountingFallback explicitExecutor = new CountingFallback(CALCITE_EXECUTOR);
+ RelNode actual = projectTimestampCast(explicitExecutor);
+ assertTrue(explicitExecutor._calls.get() > 0, "An explicit planner
executor must retain precedence");
+ assertEquals(RelOptUtil.toString(actual),
RelOptUtil.toString(projectTimestampCast(CALCITE_EXECUTOR)));
+ }
+
+ private static RelNode projectTimestampCast(RexExecutor executor) {
+ LiteralRexBuilder builder = new LiteralRexBuilder();
+ VolcanoPlanner planner = new VolcanoPlanner();
+ planner.setExecutor(executor);
+ RelBuilder relBuilder =
PinotRuleUtils.PINOT_REL_FACTORY.create(RelOptCluster.create(planner, builder),
null)
+ .transform(config -> config.withPruneInputOfAggregate(true));
+ RelNode result = relBuilder.values(new String[]{"unused"}, 0)
+ .project(timestampCast(builder, SqlTypeName.CHAR, "2026-09-01
12:34:56.123", 3)).build();
+ assertSame(planner.getExecutor(), executor, "RelBuilder.transform must not
replace the planner executor");
+ return result;
+ }
+
+ private static RexNode timestampCast(LiteralRexBuilder builder, SqlTypeName
sourceType, String value,
+ int precision) {
+ // makeCast may fold a literal before it reaches the executor, which would
make a dispatch test vacuous.
+ return
builder.makeAbstractCast(builder.getTypeFactory().createSqlType(SqlTypeName.TIMESTAMP,
precision),
+ builder.stringLiteral(value, sourceType), false);
+ }
+
+ private static List<RexNode> reduce(RexExecutor executor, RexBuilder
builder, List<RexNode> expressions) {
+ List<RexNode> reduced = new ArrayList<>();
+ executor.reduce(builder, expressions, reduced);
+ return reduced;
+ }
+
+ private static class CountingFallback implements RexExecutor {
+ private final RexExecutor _delegate;
+ private final AtomicInteger _calls = new AtomicInteger();
+ private List<RexNode> _lastInput;
+
+ CountingFallback(RexExecutor delegate) {
+ _delegate = delegate;
+ }
+
+ @Override
+ public void reduce(RexBuilder builder, List<RexNode> expressions,
List<RexNode> reduced) {
+ _calls.incrementAndGet();
+ _lastInput = expressions;
+ _delegate.reduce(builder, expressions, reduced);
+ }
+ }
+
+ private static class LiteralRexBuilder extends RexBuilder {
+ LiteralRexBuilder() {
+ this(new TypeFactory());
+ }
+
+ LiteralRexBuilder(RelDataTypeFactory typeFactory) {
+ super(typeFactory);
+ }
+
+ RexLiteral stringLiteral(String value, SqlTypeName sourceType) {
+ // Preserve VARCHAR as the source SQL type; the public literal helper
normally lowers it to CHAR or a CAST.
+ return stringLiteral(value, getTypeFactory().createSqlType(sourceType,
Math.max(1, value.length())));
+ }
+
+ RexLiteral stringLiteral(String value, RelDataType type) {
+ return makeLiteral(new NlsString(value, null, null), type,
SqlTypeName.CHAR);
+ }
+ }
+}
diff --git
a/pinot-query-planner/src/test/java/org/apache/pinot/query/ConstantCastPlanningTest.java
b/pinot-query-planner/src/test/java/org/apache/pinot/query/ConstantCastPlanningTest.java
new file mode 100644
index 00000000000..bc87ed4292b
--- /dev/null
+++
b/pinot-query-planner/src/test/java/org/apache/pinot/query/ConstantCastPlanningTest.java
@@ -0,0 +1,186 @@
+/**
+ * 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.query;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.runtime.Hook;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.util.SqlBasicVisitor;
+import org.apache.pinot.sql.parsers.CalciteSqlParser;
+import org.apache.pinot.sql.parsers.SqlNodeAndOptions;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+
+
+/// Verifies that SQL-to-rel conversion reduces constant string casts without
changing the resulting plan,
+/// and restores the planner's executor afterward. The expression-reduction
hook is thread-local and
+/// closed after each compilation.
+public class ConstantCastPlanningTest extends QueryEnvironmentTestBase {
+ @DataProvider
+ public Object[][] timestampQueries() {
+ return new Object[][]{
+ {false, "account_1", false},
+ {false, "account_2", false},
+ {true, "account_1", false},
+ {true, "account_2", false},
+ {false, "account_1", true},
+ {false, "account_2", true},
+ {true, "account_1", true},
+ {true, "account_2", true}
+ };
+ }
+
+ @Test(dataProvider = "timestampQueries")
+ // The hook resource installs and removes the callback; it is not otherwise
referenced.
+ @SuppressWarnings("try")
+ public void testConstantTimestampCastPlanning(boolean window, String
accountId, boolean epochString) {
+ String lowerBound = epochString ? "1704067200000" : "2024-01-01 00:00:00";
+ String upperBound = epochString ? "1706745600000" : "2024-02-01 00:00:00";
+ String castQuery = query(window, accountId, "CAST('" + lowerBound + "' AS
TIMESTAMP)",
+ "CAST('" + upperBound + "' AS TIMESTAMP)");
+ String literalQuery = query(window, accountId, "TIMESTAMP '2024-01-01
00:00:00'",
+ "TIMESTAMP '2024-02-01 00:00:00'");
+ SqlNodeAndOptions parsedQuery =
CalciteSqlParser.compileToSqlNodeAndOptions(castQuery);
+ AtomicInteger parsedCasts = new AtomicInteger();
+ parsedQuery.getSqlNode().accept(new SqlBasicVisitor<Void>() {
+ @Override
+ public Void visit(SqlCall call) {
+ if (call.getKind() == SqlKind.CAST) {
+ parsedCasts.incrementAndGet();
+ }
+ return super.visit(call);
+ }
+ });
+ assertEquals(parsedCasts.get(), 2, "The query must reach compilation with
both string-to-timestamp casts");
+
+ AtomicInteger reductions = new AtomicInteger();
+ try (QueryEnvironment.CompiledQuery literalPlan =
_queryEnvironment.compile(literalQuery);
+ Hook.Closeable ignored = Hook.EXPRESSION_REDUCER.addThread(value -> {
+ reductions.incrementAndGet();
+ });
+ QueryEnvironment.CompiledQuery castPlan =
_queryEnvironment.compile(castQuery, parsedQuery)) {
+ assertEquals(RelOptUtil.toString(castPlan.getRelNode()),
RelOptUtil.toString(literalPlan.getRelNode()));
+ assertEquals(castPlan.getRelRoot().validatedRowType,
literalPlan.getRelRoot().validatedRowType);
+ assertEquals(castPlan.getTableNames(), literalPlan.getTableNames());
+ assertNull(castPlan.getPlannerContext().getRelOptPlanner().getExecutor(),
+ "SQL-to-rel conversion must restore the planner's executor before
optimization");
+ assertEquals(reductions.get(), 0, "Timestamp casts must use cached
templates instead of the fallback reducer");
+ }
+ }
+
+ @DataProvider
+ public Object[][] bigintQueries() {
+ List<Object[]> cases = new ArrayList<>();
+ for (String value : List.of("0", "123", "9007199254740993",
"9223372036854775807", "-9223372036854775808")) {
+ for (boolean explicitCast : List.of(false, true)) {
+ for (boolean window : List.of(false, true)) {
+ cases.add(new Object[]{value, explicitCast, window});
+ }
+ }
+ }
+ return cases.toArray(new Object[0][]);
+ }
+
+ @Test(dataProvider = "bigintQueries")
+ @SuppressWarnings("try") // The resource scopes the thread-local
fallback-reducer hook.
+ public void testConstantBigintCastPlanning(String value, boolean
explicitCast, boolean window) {
+ String quoted = explicitCast ? "CAST('" + value + "' AS BIGINT)" : "'" +
value + "'";
+ String numeric = "CAST(" + value + " AS BIGINT)";
+ AtomicInteger reductions = new AtomicInteger();
+ try (QueryEnvironment.CompiledQuery literalPlan =
_queryEnvironment.compile(bigintQuery(numeric, window));
+ Hook.Closeable ignored = Hook.EXPRESSION_REDUCER.addThread(v -> {
+ reductions.incrementAndGet();
+ });
+ QueryEnvironment.CompiledQuery castPlan =
_queryEnvironment.compile(bigintQuery(quoted, window))) {
+ assertEquals(RelOptUtil.toString(castPlan.getRelNode()),
RelOptUtil.toString(literalPlan.getRelNode()));
+ assertEquals(castPlan.getRelRoot().validatedRowType,
literalPlan.getRelRoot().validatedRowType);
+ assertEquals(castPlan.getTableNames(), literalPlan.getTableNames());
+
assertNull(castPlan.getPlannerContext().getRelOptPlanner().getExecutor());
+ assertEquals(reductions.get(), 0, "BIGINT casts must use cached
templates instead of the fallback reducer");
+ }
+ }
+
+ @DataProvider
+ public Object[][] scalarCastQueries() {
+ return new Object[][]{
+ {"INTEGER", "2147483647", "CAST(2147483647 AS INTEGER)",
SqlTypeName.INTEGER},
+ {"SMALLINT", "32767", "CAST(32767 AS SMALLINT)", SqlTypeName.SMALLINT},
+ {"TINYINT", "127", "CAST(127 AS TINYINT)", SqlTypeName.TINYINT},
+ {"REAL", "1.25", "CAST(1.25 AS REAL)", SqlTypeName.REAL},
+ {"FLOAT", "1.25", "CAST(1.25 AS FLOAT)", SqlTypeName.FLOAT},
+ {"DOUBLE", "1.25", "CAST(1.25 AS DOUBLE)", SqlTypeName.DOUBLE},
+ {"DECIMAL(6, 2)", "1234.50", "CAST(1234.50 AS DECIMAL(6, 2))",
SqlTypeName.DECIMAL},
+ {"DECIMAL(20, 4)", "9007199254740993.1250",
"CAST(9007199254740993.1250 AS DECIMAL(20, 4))",
+ SqlTypeName.DECIMAL},
+ {"BOOLEAN", "true", "TRUE", SqlTypeName.BOOLEAN},
+ {"BOOLEAN", "false", "FALSE", SqlTypeName.BOOLEAN},
+ {"DATE", "2024-02-29", "DATE '2024-02-29'", SqlTypeName.DATE},
+ {"TIME", "12:34:56", "TIME '12:34:56'", SqlTypeName.TIME},
+ {"CHAR(3)", "abcdef", "'abc'", SqlTypeName.CHAR}
+ };
+ }
+
+ @Test(dataProvider = "scalarCastQueries")
+ public void testConstantScalarCastPlanning(String targetType, String value,
String literal,
+ SqlTypeName expectedType) {
+ String castQuery = "SELECT CAST('" + value + "' AS " + targetType + ") AS
cast_value, col3 FROM a WHERE col3 > 0";
+ String literalQuery = "SELECT " + literal + " AS cast_value, col3 FROM a
WHERE col3 > 0";
+ try (QueryEnvironment.CompiledQuery literalPlan =
_queryEnvironment.compile(literalQuery);
+ QueryEnvironment.CompiledQuery castPlan =
_queryEnvironment.compile(castQuery)) {
+ assertEquals(RelOptUtil.toString(castPlan.getRelNode()),
RelOptUtil.toString(literalPlan.getRelNode()));
+ assertEquals(castPlan.getRelRoot().validatedRowType,
literalPlan.getRelRoot().validatedRowType);
+
assertEquals(castPlan.getRelRoot().validatedRowType.getFieldList().get(0).getType().getSqlTypeName(),
+ expectedType);
+ assertEquals(castPlan.getTableNames(), literalPlan.getTableNames());
+
assertNull(castPlan.getPlannerContext().getRelOptPlanner().getExecutor());
+ // Exercise conversion to dispatchable stages as well as SQL-to-rel
conversion and optimization.
+ assertNotNull(literalPlan.planQuery(0).getQueryPlan());
+ assertNotNull(castPlan.planQuery(0).getQueryPlan());
+ }
+ }
+
+ private static String bigintQuery(String value, boolean window) {
+ String filter = " FROM a WHERE col7 = " + value;
+ if (window) {
+ return "SELECT col7, col3 FROM (SELECT col7, col3, ROW_NUMBER() OVER "
+ + "(PARTITION BY col7 ORDER BY col3 DESC) AS row_num" + filter + ")
WHERE row_num = 1";
+ }
+ return "SELECT col7, col3" + filter;
+ }
+
+ private static String query(boolean window, String accountId, String
lowerBound, String upperBound) {
+ String filter = " FROM a WHERE col1 = '" + accountId + "' AND ts_timestamp
>= " + lowerBound
+ + " AND ts_timestamp < " + upperBound;
+ if (window) {
+ return "SELECT ts_timestamp, col3 FROM (SELECT ts_timestamp, col3,
ROW_NUMBER() OVER "
+ + "(PARTITION BY col1, ts_timestamp ORDER BY col3 DESC, col7 DESC)
AS row_num" + filter
+ + ") WHERE row_num = 1 ORDER BY ts_timestamp LIMIT 1000";
+ }
+ return "SELECT ts_timestamp, col3" + filter + " ORDER BY ts_timestamp
LIMIT 1000";
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
