From d3e3be38f4c946b5b109a0665be2cab680cddecc Mon Sep 17 00:00:00 2001 From: Gian Merlino Date: Thu, 15 Oct 2026 07:14:55 -0700 Subject: [PATCH 1/6] feat: sqlTimeCondition context parameter. This patch adds the context parameter "sqlTimeCondition", which adds a time filter to all tables. The filter is ANDed with any other filter that the user may have provided. The goal of this feature is to enable UIs and APIs to accept user-written SQL along with having a separate control for time. This patch also fixes two pre-existing bugs that were discovered while writing tests for this feature: 1) In some cases, WHERE on time + ORDER BY non-time over a join could lose the time filter. This is fixed by passing scan.getIntervals() to WindowOperatorQuery in DruidQuery#toScanAndSortQuery. testLeftJoinWithTimeFilterAndNonTimeOrderBy is a regression test. 2) The SQL planner would produce "intervals" rather than a __time filter for certain time filters over a join of a left-hand subquery. This would cause execution to fail, because execution assumes that "intervals" are only used for regular tables. This is fixed by adjusting ExecutionVertexExplorer to avoid traversing through noncollapsible subqueries, making such a query no longer viewed by the SQL planner as table-based. The regression test is in testJoinWithSubqueryOnLeftAndOuterTimeFilter. --- docs/querying/sql-query-context.md | 1 + .../druid/msq/logical/stages/ReadStage.java | 8 +- .../CalciteTimeConditionQueryMSQTest.java | 81 +++ ...pledDartCalciteTimeConditionQueryTest.java | 46 ++ .../druid/query/planning/ExecutionVertex.java | 8 +- .../query/planning/ExecutionVertexTest.java | 28 + .../calcite/planner/CatalogDdlHandler.java | 9 + .../sql/calcite/planner/IngestHandler.java | 10 + .../sql/calcite/planner/PlannerContext.java | 73 +++ .../planner/querygen/DruidQueryGenerator.java | 7 +- .../sql/calcite/rel/DruidJoinQueryRel.java | 13 +- .../druid/sql/calcite/rel/DruidQuery.java | 55 +- .../sql/calcite/run/NativeQueryMaker.java | 24 +- .../sql/calcite/CalciteCatalogDdlTest.java | 35 ++ .../sql/calcite/CalciteInsertDmlTest.java | 31 + .../sql/calcite/CalciteJoinQueryTest.java | 129 ++++ .../sql/calcite/CalciteReplaceDmlTest.java | 15 + .../CalciteTimeConditionQueryTest.java | 569 ++++++++++++++++++ ...PlanningCalciteTimeConditionQueryTest.java | 41 ++ .../calcite/planner/PlannerContextTest.java | 137 +++++ .../druid/sql/calcite/rel/DruidQueryTest.java | 151 +++++ ...oinWithSubqueryOnLeftAndOuterTimeFilter.iq | 103 ++++ ...LeftJoinWithTimeFilterAndNonTimeOrderBy.iq | 150 +++++ website/.spelling | 1 + 24 files changed, 1697 insertions(+), 28 deletions(-) create mode 100644 multi-stage-query/src/test/java/org/apache/druid/msq/test/CalciteTimeConditionQueryMSQTest.java create mode 100644 multi-stage-query/src/test/java/org/apache/druid/msq/test/DecoupledDartCalciteTimeConditionQueryTest.java create mode 100644 sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java create mode 100644 sql/src/test/java/org/apache/druid/sql/calcite/DecoupledPlanningCalciteTimeConditionQueryTest.java create mode 100644 sql/src/test/java/org/apache/druid/sql/calcite/planner/PlannerContextTest.java create mode 100644 sql/src/test/quidem/org.apache.druid.sql.calcite.DecoupledPlanningCalciteJoinQueryTest/testJoinWithSubqueryOnLeftAndOuterTimeFilter.iq create mode 100644 sql/src/test/quidem/org.apache.druid.sql.calcite.DecoupledPlanningCalciteJoinQueryTest/testLeftJoinWithTimeFilterAndNonTimeOrderBy.iq diff --git a/docs/querying/sql-query-context.md b/docs/querying/sql-query-context.md index 88cafe1ced71..037ea40c2b5c 100644 --- a/docs/querying/sql-query-context.md +++ b/docs/querying/sql-query-context.md @@ -51,6 +51,7 @@ The table below lists the query context parameters you can use with Druid SQL. |`sqlReverseLookup`|Whether to consider the [reverse-lookup rewrite](lookups.md#reverse-lookup) of the `LOOKUP` function during SQL planning.

Druid reverses calls to `LOOKUP` only when the number of matching keys is lower than both `inSubQueryThreshold` and `sqlReverseLookupThreshold`.|`true`| |`sqlReverseLookupThreshold`|Maximum size of `IN` filter to create when applying a [reverse-lookup rewrite](lookups.md#reverse-lookup). If a `LOOKUP` call matches more keys than the specified threshold, it remains unchanged.

If `inSubQueryThreshold` is lower than `sqlReverseLookupThreshold`, Druid uses `inSubQueryThreshold` threshold instead.|10000| |`sqlPullUpLookup`|Whether to consider the [pull-up rewrite](lookups.md#pull-up) of the `LOOKUP` function during SQL planning.|`true`| +|`sqlTimeCondition`|Time condition to apply to all tables that the query reads. Provide either a comma-separated string of ISO8601 intervals, such as `"2000-01-01/2000-02-01,2001-01-01/2001-02-01"`, or an array of ISO8601 interval strings. An empty array matches nothing. An empty string is an error.

Druid ANDs this condition with any other condition you provide on the `__time` column. Intervals without an explicit offset are in UTC; `sqlTimeZone` doesn't affect them.

You can't use this parameter with `INSERT`, `REPLACE`, or [catalog DDL](../development/extensions-core/catalog.md#sql-ddl) statements.|none| |`enableJoinLeftTableScanDirect`|This parameter applies to queries with joins. By default, when the left child is a simple scan with a filter, Druid runs the scan as a query, then joins it with the right child on the Broker. Setting this parameter to `true` overrides that behavior and pushes the join to the data servers instead. Even if a query doesn't explicitly include a join, this parameter may still apply since the SQL planner can translate the query into a join internally.|`false`| |`maxNumericInFilters`|Max limit for the amount of numeric values that Druid can compare for a string type dimension when the entire SQL WHERE clause of a query translates only to an [OR](../querying/filters.md#or) of [bound filter](../querying/filters.md#bound-filter). By default, Druid doesn't restrict the amount of numeric bound filters on string columns, although this situation may block other queries from running. Set this parameter to a smaller value to prevent Druid from running queries that have prohibitively long segment processing times. The optimal limit requires some trial and error. We recommend starting with 100. Users who submit a query that exceeds the limit of `maxNumericInFilters` should rewrite their queries to use strings in the `WHERE` clause instead of numbers. For example, `WHERE someString IN (‘123’, ‘456’)`. This value can't exceed the set system configuration `druid.sql.planner.maxNumericInFilters`. If `druid.sql.planner.maxNumericInFilters` isn't set explicitly, Druid ignores this value.|`-1`| |`inFunctionThreshold`| At or beyond this threshold number of values, Druid converts SQL `IN` to [`SCALAR_IN_ARRAY`](sql-functions.md#scalar_in_array). A threshold of 0 forces this conversion in all cases. A threshold of `Integer.MAX_VALUE` disables this conversion. The converted function is eligible for fewer planning-time optimizations, which speeds up planning, but may prevent certain planning-time optimizations.| `100`| diff --git a/multi-stage-query/src/main/java/org/apache/druid/msq/logical/stages/ReadStage.java b/multi-stage-query/src/main/java/org/apache/druid/msq/logical/stages/ReadStage.java index a4ece30a0cd6..7584040415d9 100644 --- a/multi-stage-query/src/main/java/org/apache/druid/msq/logical/stages/ReadStage.java +++ b/multi-stage-query/src/main/java/org/apache/druid/msq/logical/stages/ReadStage.java @@ -20,7 +20,6 @@ package org.apache.druid.msq.logical.stages; import org.apache.druid.error.DruidException; -import org.apache.druid.java.util.common.Intervals; import org.apache.druid.msq.exec.StageProcessor; import org.apache.druid.msq.input.InputSpec; import org.apache.druid.msq.input.inline.InlineInputSpec; @@ -131,17 +130,18 @@ public static Optional buildReadStage(DruidNodeStack stack) private static Optional translateViaSourceDesc(PlannerContext plannerContext, SourceDescProducer node) { SourceDesc sd = node.getSourceDesc(plannerContext, Collections.emptyList()); - InputSpec inputSpec = translateDataSource(sd.dataSource); + InputSpec inputSpec = translateDataSource(plannerContext, sd.dataSource); ReadStage stage = new ReadStage(sd.rowSignature, LogicalInputSpec.of(inputSpec, sd.rowSignature)); return Optional.of(stage); } @Nonnull - private static InputSpec translateDataSource(DataSource dataSource) + private static InputSpec translateDataSource(PlannerContext plannerContext, DataSource dataSource) { if (dataSource instanceof TableDataSource) { TableDataSource ids = (TableDataSource) dataSource; - TableInputSpec inputSpec = new TableInputSpec(ids.getName(), Intervals.ONLY_ETERNITY, null); + // TableInputSpec treats null intervals as eternity. + TableInputSpec inputSpec = new TableInputSpec(ids.getName(), plannerContext.getTimeCondition(), null); return inputSpec; } if (dataSource instanceof InlineDataSource) { diff --git a/multi-stage-query/src/test/java/org/apache/druid/msq/test/CalciteTimeConditionQueryMSQTest.java b/multi-stage-query/src/test/java/org/apache/druid/msq/test/CalciteTimeConditionQueryMSQTest.java new file mode 100644 index 000000000000..2d6b0d52d9d0 --- /dev/null +++ b/multi-stage-query/src/test/java/org/apache/druid/msq/test/CalciteTimeConditionQueryMSQTest.java @@ -0,0 +1,81 @@ +/* + * 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.druid.msq.test; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import org.apache.druid.msq.sql.MSQTaskSqlEngine; +import org.apache.druid.query.JoinAlgorithm; +import org.apache.druid.sql.calcite.CalciteTimeConditionQueryTest; +import org.apache.druid.sql.calcite.QueryTestBuilder; +import org.apache.druid.sql.calcite.SqlTestFrameworkConfig; +import org.apache.druid.sql.calcite.planner.PlannerContext; +import org.junit.jupiter.api.Test; + +import java.util.Map; + +/** + * Runs {@link CalciteTimeConditionQueryTest} with the MSQ engine. + */ +@SqlTestFrameworkConfig.ComponentSupplier(StandardMSQComponentSupplier.class) +public class CalciteTimeConditionQueryMSQTest extends CalciteTimeConditionQueryTest +{ + @Override + protected QueryTestBuilder testBuilder() + { + return new QueryTestBuilder(new CalciteTestConfig(true)) + .addCustomRunner( + new ExtractResultsFactory( + () -> (MSQTestOverlordServiceClient) ((MSQTaskSqlEngine) queryFramework().engine()).overlordClient() + ) + ) + .skipVectorize(true) + .verifyNativeQueries(new VerifyMSQSupportedNativeQueriesPredicate()); + } + + @Test + public void testSortMergeJoin() + { + final Map context = ImmutableMap.builder() + .putAll(makeContext("2000/2001")) + .put( + PlannerContext.CTX_SQL_JOIN_ALGORITHM, + JoinAlgorithm.SORT_MERGE.toString() + ) + .build(); + + // Both sides of a sort-merge join are subqueries, so the time condition applies within each subquery. + testBuilder() + .sql("SELECT f1.dim1, f2.dim2 FROM foo f1 INNER JOIN foo f2 ON f1.dim1 = f2.dim1") + .queryContext(context) + .expectedResults( + new UnorderedResultsVerifier( + ImmutableList.of( + new Object[]{"", "a"}, + new Object[]{"10.1", null}, + new Object[]{"2", ""} + ), + ResultMatchMode.EQUALS, + null + ) + ) + .run(); + } +} diff --git a/multi-stage-query/src/test/java/org/apache/druid/msq/test/DecoupledDartCalciteTimeConditionQueryTest.java b/multi-stage-query/src/test/java/org/apache/druid/msq/test/DecoupledDartCalciteTimeConditionQueryTest.java new file mode 100644 index 000000000000..77a7e879e88d --- /dev/null +++ b/multi-stage-query/src/test/java/org/apache/druid/msq/test/DecoupledDartCalciteTimeConditionQueryTest.java @@ -0,0 +1,46 @@ +/* + * 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.druid.msq.test; + +import org.apache.druid.sql.calcite.CalciteTimeConditionQueryTest; +import org.apache.druid.sql.calcite.NotYetSupported; +import org.apache.druid.sql.calcite.NotYetSupported.NotYetSupportedProcessor; +import org.apache.druid.sql.calcite.QueryTestBuilder; +import org.apache.druid.sql.calcite.SqlTestFrameworkConfig; +import org.junit.jupiter.api.extension.RegisterExtension; + +/** + * Runs {@link CalciteTimeConditionQueryTest} with Dart and the decoupled planner. + */ +@SqlTestFrameworkConfig.ComponentSupplier(DartComponentSupplier.class) +public class DecoupledDartCalciteTimeConditionQueryTest extends CalciteTimeConditionQueryTest +{ + @RegisterExtension + NotYetSupportedProcessor notYetSupportedProcessor = new NotYetSupportedProcessor(NotYetSupported.Scope.DECOUPLED_DART); + + @RegisterExtension + DecoupledDartExtension decoupledExtension = new DecoupledDartExtension(this); + + @Override + protected QueryTestBuilder testBuilder() + { + return decoupledExtension.testBuilder(); + } +} diff --git a/processing/src/main/java/org/apache/druid/query/planning/ExecutionVertex.java b/processing/src/main/java/org/apache/druid/query/planning/ExecutionVertex.java index c80b79d4d268..2a1c18de76a7 100644 --- a/processing/src/main/java/org/apache/druid/query/planning/ExecutionVertex.java +++ b/processing/src/main/java/org/apache/druid/query/planning/ExecutionVertex.java @@ -347,10 +347,10 @@ protected boolean mayTraverseDataSource(EVNode node) return false; } if (node.dataSource instanceof QueryDataSource) { - EVNode parentNode = parents.get(parents.size() - 2); - if (parentNode.isQuery()) { - return parentNode.getQuery().mayCollapseQueryDataSource(); - } + // Subqueries are part of this vertex only if the query directly above them can collapse them. Subqueries + // under other datasources, such as the left-hand side of a join, are executed separately. + final EVNode parentNode = parents.get(parents.size() - 2); + return parentNode.isQuery() && parentNode.getQuery().mayCollapseQueryDataSource(); } if (node.dataSource instanceof UnionDataSource) { return false; diff --git a/processing/src/test/java/org/apache/druid/query/planning/ExecutionVertexTest.java b/processing/src/test/java/org/apache/druid/query/planning/ExecutionVertexTest.java index 8be659301806..73a973110f20 100644 --- a/processing/src/test/java/org/apache/druid/query/planning/ExecutionVertexTest.java +++ b/processing/src/test/java/org/apache/druid/query/planning/ExecutionVertexTest.java @@ -761,6 +761,34 @@ public void testJoinLookupToTable() Assertions.assertTrue(ev.isSegmentMapFunctionExpensive()); } + @Test + public void testJoinSubqueryToLookup() + { + final QueryDataSource queryDataSource = new QueryDataSource( + GroupByQuery.builder() + .setDataSource(TABLE_FOO) + .setInterval(Intervals.ONLY_ETERNITY) + .setGranularity(Granularities.ALL) + .build() + ); + final JoinDataSource joinDataSource = join( + queryDataSource, + LOOKUP_LOOKYLOO, + "1.", + JoinType.INNER + ); + + final ExecutionVertex ev = ExecutionVertex.of(makeScanQuery(joinDataSource)); + + // The subquery is executed separately, so it is the base, and the effective intervals are the outer query's. + Assertions.assertFalse(ev.isProcessable()); + Assertions.assertFalse(ev.isTableBased()); + Assertions.assertEquals(queryDataSource, ev.getBaseDataSource()); + Assertions.assertThrows(DruidException.class, () -> ev.getBaseTableDataSource()); + Assertions.assertEquals(new MultipleIntervalSegmentSpec(MILLENIUM_INTERVALS), ev.getEffectiveQuerySegmentSpec()); + Assertions.assertTrue(ev.isSegmentMapFunctionExpensive()); + } + @Test public void testHashCodeIsNotSupported() { diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/CatalogDdlHandler.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/CatalogDdlHandler.java index a59aaa730d0b..2fe37ff38da8 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/CatalogDdlHandler.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/CatalogDdlHandler.java @@ -129,6 +129,15 @@ public void validate() if (!handlerContext.plannerContext().getParameters().isEmpty()) { throw InvalidSqlInput.exception("Dynamic parameters are not supported for [%s]", operationName()); } + // Fail if CTX_SQL_TIME_CONDITION is specified, since projection bodies are planned with the statement's context, + // so the condition would be stored in the catalog as part of their filters. + if (handlerContext.queryContextMap().get(PlannerContext.CTX_SQL_TIME_CONDITION) != null) { + throw InvalidSqlInput.exception( + "Context parameter [%s] cannot be provided on operator [%s]", + PlannerContext.CTX_SQL_TIME_CONDITION, + operationName() + ); + } tableId = TableId.datasource(resolveTableName()); final Resource resource = new Resource(tableId.name(), ResourceType.DATASOURCE); resourceActions = ImmutableSet.of( diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/IngestHandler.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/IngestHandler.java index 1098e1adfb7c..16c59a33df5c 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/IngestHandler.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/IngestHandler.java @@ -157,6 +157,16 @@ public void validate() operationName() ); } + + // Fail the query if CTX_SQL_TIME_CONDITION is specified, since it limits the rows read without limiting the time + // chunks that are overwritten, so REPLACE would delete data outside the condition. + if (handlerContext.queryContextMap().get(PlannerContext.CTX_SQL_TIME_CONDITION) != null) { + throw InvalidSqlInput.exception( + "Context parameter [%s] cannot be provided on operator [%s]", + PlannerContext.CTX_SQL_TIME_CONDITION, + operationName() + ); + } DruidSqlIngest ingestNode = ingestNode(); DruidSqlIngest validatedNode = (DruidSqlIngest) validate(ingestNode); validatedQueryNode = validatedNode.getSource(); diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/PlannerContext.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/PlannerContext.java index 923bc1d198a2..fb5b07415f9b 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/PlannerContext.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/PlannerContext.java @@ -31,9 +31,13 @@ import org.apache.calcite.schema.Table; import org.apache.calcite.sql.SqlNode; import org.apache.calcite.util.CancelFlag; +import org.apache.druid.error.DruidException; +import org.apache.druid.error.InvalidInput; import org.apache.druid.error.InvalidSqlInput; import org.apache.druid.java.util.common.DateTimes; import org.apache.druid.java.util.common.ISE; +import org.apache.druid.java.util.common.Intervals; +import org.apache.druid.java.util.common.JodaUtils; import org.apache.druid.java.util.common.Numbers; import org.apache.druid.java.util.common.StringUtils; import org.apache.druid.java.util.common.granularity.Granularities; @@ -72,6 +76,7 @@ import org.joda.time.Interval; import javax.annotation.Nullable; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.HashSet; @@ -138,6 +143,11 @@ public class PlannerContext public static final String CTX_SQL_USE_GRANULARITY = "sqlUseGranularity"; public static final boolean DEFAULT_SQL_USE_GRANULARITY = true; + /** + * Context key for {@link PlannerContext#getTimeCondition()}. + */ + public static final String CTX_SQL_TIME_CONDITION = "sqlTimeCondition"; + // DataContext keys public static final String DATA_CTX_AUTHENTICATION_RESULT = "authenticationResult"; @@ -179,6 +189,8 @@ public class PlannerContext private boolean pullUpLookup; private boolean reverseLookup; private boolean useGranularity; + @Nullable + private List timeCondition; private DateTime localNow; // bindings for dynamic parameters to bind during planning @@ -485,6 +497,16 @@ public boolean isUseGranularity() return useGranularity; } + /** + * Returns intervals that should be applied (via AND) to table reads. The returned list is sorted and + * non-overlapping. + */ + @Nullable + public List getTimeCondition() + { + return timeCondition; + } + public List getParameters() { return parameters; @@ -800,6 +822,8 @@ private void initializeContextFieldsAndPlannerConfig() useGranularity = DEFAULT_SQL_USE_GRANULARITY; } + timeCondition = parseTimeCondition(queryContext.get(CTX_SQL_TIME_CONDITION)); + sqlQueryId = (String) this.queryContext.get(QueryContexts.CTX_SQL_QUERY_ID); // special handling for DruidViewMacro, normal client will allocate sqlid in SqlLifecyle if (Strings.isNullOrEmpty(sqlQueryId)) { @@ -814,6 +838,44 @@ private void initializeContextFieldsAndPlannerConfig() } } + /** + * Parses the value of {@link #CTX_SQL_TIME_CONDITION}, which may be a comma-separated string of ISO8601 intervals + * or an array of ISO8601 interval strings. Returns null if the value is null. Throws an error if the string is + * empty or if there are empty items after splitting on commas. The returned intervals are condensed and sorted. + */ + @Nullable + static List parseTimeCondition(@Nullable final Object value) + { + if (value == null) { + return null; + } + + final List intervalStrings; + if (value instanceof String) { + intervalStrings = Arrays.asList(((String) value).split(",", -1)); + } else if (value instanceof List) { + intervalStrings = (List) value; + } else { + throw invalidTimeConditionException(value, null); + } + + final List intervals = new ArrayList<>(); + for (final Object intervalString : intervalStrings) { + if (!(intervalString instanceof String)) { + throw invalidTimeConditionException(value, null); + } + + try { + intervals.add(Intervals.of(((String) intervalString).trim())); + } + catch (DruidException e) { + throw invalidTimeConditionException(value, e); + } + } + + return JodaUtils.condenseIntervals(intervals); + } + private static JoinAlgorithm getJoinAlgorithmFromContextValue(final Object object) { final String s = QueryContexts.getAsString( @@ -833,4 +895,15 @@ private static JoinAlgorithm getJoinAlgorithmFromContextValue(final Object objec ); } } + + private static DruidException invalidTimeConditionException(final Object value, @Nullable final Throwable cause) + { + return InvalidInput.exception( + cause, + "Query context parameter [%s] must be a comma-separated string of ISO8601 intervals, or an array of " + + "ISO8601 interval strings, but got [%s]", + CTX_SQL_TIME_CONDITION, + value + ); + } } diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/querygen/DruidQueryGenerator.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/querygen/DruidQueryGenerator.java index 0ed1c7949888..98058256db3d 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/querygen/DruidQueryGenerator.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/querygen/DruidQueryGenerator.java @@ -37,6 +37,7 @@ import org.apache.druid.sql.calcite.planner.PlannerContext; import org.apache.druid.sql.calcite.planner.querygen.DruidQueryGenerator.PDQVertexFactory.PDQVertex; import org.apache.druid.sql.calcite.planner.querygen.SourceDescProducer.SourceDesc; +import org.apache.druid.sql.calcite.rel.DruidJoinQueryRel; import org.apache.druid.sql.calcite.rel.DruidQuery; import org.apache.druid.sql.calcite.rel.DruidRels; import org.apache.druid.sql.calcite.rel.PartialDruidQuery; @@ -243,9 +244,9 @@ static VertexTweaks analyze(DruidNodeStack stack) return new VertexTweaks(JoinPosition.analyze(stack)); } - boolean forceSubQuery(SourceDesc sourceDesc) + boolean forceSubQuery(PlannerContext plannerContext, SourceDesc sourceDesc) { - if (sourceDesc.dataSource.isGlobal()) { + if (DruidJoinQueryRel.canUseDirectlyOnRight(plannerContext, sourceDesc.dataSource)) { return false; } return joinType == JoinPosition.RIGHT; @@ -513,7 +514,7 @@ public SourceDesc unwrapUnionArm() @Override public boolean canUnwrapSourceDesc() { - if (tweaks.forceSubQuery(getSource())) { + if (tweaks.forceSubQuery(plannerContext, getSource())) { return false; } if (partialDruidQuery.stage() == Stage.SCAN) { diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidJoinQueryRel.java b/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidJoinQueryRel.java index 553919fceb27..81723e032108 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidJoinQueryRel.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidJoinQueryRel.java @@ -430,7 +430,18 @@ public static boolean computeRightRequiresSubquery(final PlannerContext plannerC // ideally this would involve JoinableFactory.isDirectlyJoinable to check that the global datasources // are in fact possibly joinable, but for now isGlobal is coupled to joinability return !(DruidRels.isScanOrMapping(right, false) - && DruidRels.druidTableIfLeafRel(right).filter(table -> table.getDataSource().isGlobal()).isPresent()); + && DruidRels.druidTableIfLeafRel(right) + .filter(table -> canUseDirectlyOnRight(plannerContext, table.getDataSource())) + .isPresent()); + } + + /** + * Whether a datasource can be used directly as the right-hand side of a join, rather than through a subquery. + */ + public static boolean canUseDirectlyOnRight(final PlannerContext plannerContext, final DataSource dataSource) + { + return dataSource.isGlobal() + && (plannerContext.getTimeCondition() == null || dataSource.getTableNames().isEmpty()); } public static Set findExistingJoinPrefixes(DataSource... dataSources) diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidQuery.java b/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidQuery.java index aa2f9a253fb5..bdc7c3455aa4 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidQuery.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidQuery.java @@ -817,11 +817,16 @@ VirtualColumns computeVirtualColumns(final boolean includeDimensions) * Returns a pair of DataSource and Filtration object created on the query filter. In case the data source is * a join datasource, the datasource may be altered and left filter of join datasource may * be rid of time filters. + * + * @param timeCondition intervals to AND with the filtration intervals if the datasource is table-based, from + * {@link PlannerContext#getTimeCondition()}. Must be sorted and non-overlapping. Null means + * no additional condition. */ @VisibleForTesting static Pair getFiltration( DataSource dataSource, DimFilter filter, + @Nullable List timeCondition, VirtualColumnRegistry virtualColumnRegistry, JoinableFactoryWrapper joinableFactoryWrapper ) @@ -834,7 +839,7 @@ static Pair getFiltration( // Time filters within the datasource narrow the query intervals too, which lets the broker prune segments. final Filtration queryFiltration = - Filtration.create(filter, getBaseIntervals(dataSource, rowSignature)).optimize(rowSignature); + Filtration.create(filter, getBaseIntervals(dataSource, timeCondition, rowSignature)).optimize(rowSignature); if (dataSource instanceof JoinDataSource join && join.getLeftFilter() != null) { // Strip the time filters from the left filter, since the query intervals now carry them. @@ -855,34 +860,42 @@ static Pair getFiltration( } /** - * Returns the intervals that time filters within the given datasource restrict its {@code __time} to, or eternity - * if there are none. Filters along a single path are intersected, and the arms of a union are combined. Relies on - * every layer passing {@code __time} through unchanged, as the datasources we plan do. + * Returns the intervals that time filters within the given datasource, and the time condition on each table it + * reads, restrict its {@code __time} to, or eternity if there are none. Filters along a single path are intersected, + * and the arms of a union are combined. Relies on every layer passing {@code __time} through unchanged, as the + * datasources we plan do. + * + * @param timeCondition intervals from {@link PlannerContext#getTimeCondition()}, or null for no condition */ - private static List getBaseIntervals(final DataSource dataSource, final RowSignature rowSignature) + private static List getBaseIntervals( + final DataSource dataSource, + @Nullable final List timeCondition, + final RowSignature rowSignature + ) { if (dataSource instanceof FilteredDataSource filtered) { // Filtration intersects the filter's time bounds with the intervals it is created with. - return Filtration.create(filtered.getFilter(), getBaseIntervals(filtered.getBase(), rowSignature)) + return Filtration.create(filtered.getFilter(), getBaseIntervals(filtered.getBase(), timeCondition, rowSignature)) .optimize(rowSignature) .getIntervals(); } else if (dataSource instanceof JoinDataSource join) { // Right-hand columns are prefixed, so __time is the left-hand side's. - return Filtration.create(join.getLeftFilter(), getBaseIntervals(join.getLeft(), rowSignature)) + return Filtration.create(join.getLeftFilter(), getBaseIntervals(join.getLeft(), timeCondition, rowSignature)) .optimize(rowSignature) .getIntervals(); } else if (dataSource instanceof UnionDataSource) { final List intervals = new ArrayList<>(); for (final DataSource arm : dataSource.getChildren()) { - intervals.addAll(getBaseIntervals(arm, rowSignature)); + intervals.addAll(getBaseIntervals(arm, timeCondition, rowSignature)); } return JodaUtils.condenseIntervals(intervals); } else if (dataSource instanceof UnnestDataSource unnest) { - return getBaseIntervals(unnest.getBase(), rowSignature); + return getBaseIntervals(unnest.getBase(), timeCondition, rowSignature); } else if (dataSource instanceof ProjectDataSource project) { - return getBaseIntervals(project.getBase(), rowSignature); + return getBaseIntervals(project.getBase(), timeCondition, rowSignature); } else { - return Intervals.ONLY_ETERNITY; + // A table, since the caller only uses this for datasources where canUseIntervalFiltering is true. + return timeCondition != null ? timeCondition : Intervals.ONLY_ETERNITY; } } @@ -1078,6 +1091,7 @@ private TimeBoundaryQuery toTimeBoundaryQuery() final Pair dataSourceFiltrationPair = getFiltration( dataSource, filter, + plannerContext.getTimeCondition(), virtualColumnRegistry, plannerContext.getJoinableFactoryWrapper() ); @@ -1202,6 +1216,7 @@ private TimeseriesQuery toTimeseriesQuery() final Pair dataSourceFiltrationPair = getFiltration( dataSource, filter, + plannerContext.getTimeCondition(), virtualColumnRegistry, plannerContext.getJoinableFactoryWrapper() ); @@ -1308,6 +1323,7 @@ private TopNQuery toTopNQuery() final Pair dataSourceFiltrationPair = getFiltration( dataSource, filter, + plannerContext.getTimeCondition(), virtualColumnRegistry, plannerContext.getJoinableFactoryWrapper() ); @@ -1355,6 +1371,7 @@ private GroupByQuery toGroupByQuery() final Pair dataSourceFiltrationPair = getFiltration( dataSource, filter, + plannerContext.getTimeCondition(), virtualColumnRegistry, plannerContext.getJoinableFactoryWrapper() ); @@ -1514,9 +1531,18 @@ private WindowOperatorQuery toWindowQuery() boolean pushLeafOperator = plannerContext.queryContext() .getBoolean(PlannerContext.CTX_ENABLE_RAC_TRANSFER_OVER_WIRE, false) && !plannerContext.featureAvailable(EngineFeature.WINDOW_LEAF_OPERATOR); - return new WindowOperatorQuery( + + final Pair dataSourceFiltrationPair = getFiltration( dataSource, - new LegacySegmentSpec(Intervals.ETERNITY), + filter, + plannerContext.getTimeCondition(), + virtualColumnRegistry, + plannerContext.getJoinableFactoryWrapper() + ); + + return new WindowOperatorQuery( + dataSourceFiltrationPair.lhs, + new LegacySegmentSpec(dataSourceFiltrationPair.rhs.getIntervals()), plannerContext.queryContextMap(), windowing.getSignature(), operators, @@ -1580,7 +1606,7 @@ private WindowOperatorQuery toScanAndSortQuery() return new WindowOperatorQuery( newDataSource, - new LegacySegmentSpec(Intervals.ETERNITY), + new LegacySegmentSpec(scan.getIntervals()), plannerContext.queryContextMap(), signature, operators, @@ -1634,6 +1660,7 @@ private ScanQuery toScanQuery(final boolean considerSorting) final Pair dataSourceFiltrationPair = getFiltration( dataSource, filter, + plannerContext.getTimeCondition(), virtualColumnRegistry, plannerContext.getJoinableFactoryWrapper() ); diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/run/NativeQueryMaker.java b/sql/src/main/java/org/apache/druid/sql/calcite/run/NativeQueryMaker.java index f0fb59254a01..5e7be676f3e4 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/run/NativeQueryMaker.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/run/NativeQueryMaker.java @@ -32,8 +32,10 @@ import org.apache.druid.java.util.common.UOE; import org.apache.druid.java.util.common.guava.Sequence; import org.apache.druid.java.util.common.guava.Sequences; +import org.apache.druid.query.BaseQuery; import org.apache.druid.query.InlineDataSource; import org.apache.druid.query.Query; +import org.apache.druid.query.QueryDataSource; import org.apache.druid.query.QueryToolChest; import org.apache.druid.query.filter.BoundDimFilter; import org.apache.druid.query.filter.DimFilter; @@ -52,6 +54,7 @@ import org.apache.druid.sql.calcite.rel.CannotBuildQueryException; import org.apache.druid.sql.calcite.rel.DruidQuery; import org.apache.druid.sql.hook.DruidHook; +import org.joda.time.Interval; import java.util.ArrayList; import java.util.List; import java.util.Map.Entry; @@ -85,8 +88,7 @@ public QueryResponse runQuery(final DruidQuery druidQuery) if (plannerContext.getPlannerConfig().isRequireTimeCondition() && !(druidQuery.getDataSource() instanceof InlineDataSource)) { - ExecutionVertex ev = ExecutionVertex.of(query); - if (Intervals.ONLY_ETERNITY.equals(ev.getEffectiveQuerySegmentSpec().getIntervals())) { + if (Intervals.ONLY_ETERNITY.equals(getBaseIntervals(query))) { throw new CannotBuildQueryException( "requireTimeCondition is enabled, all queries must include a filter condition on the __time column" ); @@ -266,4 +268,22 @@ private static List mapColumnList(final List in, final List getBaseIntervals(final Query query) + { + final ExecutionVertex ev = ExecutionVertex.of(query); + if (ev.getBaseDataSource() instanceof QueryDataSource) { + final Query subQuery = ((QueryDataSource) ev.getBaseDataSource()).getQuery(); + + // ExecutionVertex can only analyze BaseQuery. + if (subQuery instanceof BaseQuery) { + return getBaseIntervals(subQuery); + } + } + return ev.getEffectiveQuerySegmentSpec().getIntervals(); + } } diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteCatalogDdlTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteCatalogDdlTest.java index d4c926bb9435..10cfd0f4c447 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteCatalogDdlTest.java +++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteCatalogDdlTest.java @@ -532,6 +532,41 @@ public void testProjectionBodyContextCannotOverrideDeterministicOverrides() thro ); } + @Test + public void testCreateTableWithSqlTimeCondition() + { + final DruidException e = assertThrows( + DruidException.class, + () -> execute( + "SET sqlTimeCondition = '2020-01-01/2020-02-01';\n" + + "CREATE TABLE tbl (__time TIMESTAMP, page VARCHAR, cnt BIGINT," + + " PROJECTION p AS (SELECT page, SUM(cnt) AS total GROUP BY page))" + ) + ); + assertEquals( + "Context parameter [sqlTimeCondition] cannot be provided on operator [CREATE TABLE]", + e.getMessage() + ); + assertTrue(WRITER.calls.isEmpty()); + } + + @Test + public void testAlterTableWithSqlTimeCondition() + { + final DruidException e = assertThrows( + DruidException.class, + () -> execute( + "SET sqlTimeCondition = '2020-01-01/2020-02-01';\n" + + "ALTER TABLE tbl SET PROPERTIES (description = 'hi')" + ) + ); + assertEquals( + "Context parameter [sqlTimeCondition] cannot be provided on operator [ALTER TABLE SET PROPERTIES]", + e.getMessage() + ); + assertTrue(WRITER.calls.isEmpty()); + } + /** * A projection defined with TIME_FLOOR must carry a granularity the segment layer can recover, which is how the * projection gets matched to time-grouped queries. diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteInsertDmlTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteInsertDmlTest.java index 47a31ba080c3..e3587b619e96 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteInsertDmlTest.java +++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteInsertDmlTest.java @@ -61,6 +61,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.stream.Collectors; @@ -1657,6 +1658,36 @@ public void testInsertWithSqlOuterLimit() .verify(); } + @Test + public void testInsertWithSqlTimeCondition() + { + HashMap context = new HashMap<>(DEFAULT_CONTEXT); + context.put(PlannerContext.CTX_SQL_TIME_CONDITION, "2000/2001"); + + testIngestionQuery() + .context(context) + .sql("INSERT INTO dst SELECT * FROM foo PARTITIONED BY ALL TIME") + .expectValidationError( + invalidSqlIs("Context parameter [sqlTimeCondition] cannot be provided on operator [INSERT]") + ) + .verify(); + } + + @Test + public void testInsertWithEmptySqlTimeCondition() + { + HashMap context = new HashMap<>(DEFAULT_CONTEXT); + context.put(PlannerContext.CTX_SQL_TIME_CONDITION, List.of()); + + testIngestionQuery() + .context(context) + .sql("INSERT INTO dst SELECT * FROM foo PARTITIONED BY ALL TIME") + .expectValidationError( + invalidSqlIs("Context parameter [sqlTimeCondition] cannot be provided on operator [INSERT]") + ) + .verify(); + } + @Test public void testErrorWithUnableToConstructColumnSignatureWithExtern() { diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteJoinQueryTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteJoinQueryTest.java index 66ba92b17ecf..df9375e298b1 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteJoinQueryTest.java +++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteJoinQueryTest.java @@ -41,6 +41,7 @@ import org.apache.druid.query.JoinAlgorithm; import org.apache.druid.query.JoinDataSource; import org.apache.druid.query.LookupDataSource; +import org.apache.druid.query.OperatorFactoryBuilders; import org.apache.druid.query.Order; import org.apache.druid.query.Query; import org.apache.druid.query.QueryContext; @@ -71,8 +72,11 @@ import org.apache.druid.query.groupby.orderby.NoopLimitSpec; import org.apache.druid.query.groupby.orderby.OrderByColumnSpec; import org.apache.druid.query.groupby.orderby.OrderByColumnSpec.Direction; +import org.apache.druid.query.operator.ColumnWithDirection; +import org.apache.druid.query.operator.WindowOperatorQuery; import org.apache.druid.query.ordering.StringComparators; import org.apache.druid.query.scan.ScanQuery; +import org.apache.druid.query.spec.LegacySegmentSpec; import org.apache.druid.query.timeboundary.TimeBoundaryQuery; import org.apache.druid.query.topn.DimensionTopNMetricSpec; import org.apache.druid.query.topn.InvertedTopNMetricSpec; @@ -95,6 +99,7 @@ import org.apache.druid.sql.calcite.run.EngineFeature; import org.apache.druid.sql.calcite.util.CalciteTests; import org.joda.time.DateTimeZone; +import org.joda.time.Interval; import org.joda.time.Period; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Disabled; @@ -6584,4 +6589,128 @@ public void testLeftJoinsOnTwoWithTables() ) ); } + + @DecoupledTestConfig(quidemReason = QuidemTestCaseReason.JOIN_FILTER_LOCATIONS) + @Test + public void testLeftJoinWithTimeFilterAndNonTimeOrderBy() + { + // Cannot vectorize direct queries on lookup tables. + cannotVectorize(); + + final Interval interval = Intervals.utc(DateTimes.of("2001-01-01").getMillis(), JodaUtils.MAX_INSTANT); + + // The filter on __time becomes the intervals of the scan. + testQuery( + "SELECT f1.dim1, f2.v\n" + + "FROM foo f1 LEFT JOIN (SELECT k, v FROM lookup.lookyloo GROUP BY k, v) f2 ON f1.dim2 = f2.k\n" + + "WHERE f1.__time >= TIMESTAMP '2001-01-01 00:00:00'\n" + + "ORDER BY f2.v, f1.dim1", + ImmutableList.of( + new WindowOperatorQuery( + new QueryDataSource( + newScanQueryBuilder() + .dataSource( + join( + new TableDataSource(CalciteTests.DATASOURCE1), + new QueryDataSource( + GroupByQuery.builder() + .setDataSource(new LookupDataSource("lookyloo")) + .setInterval(querySegmentSpec(Filtration.eternity())) + .setGranularity(Granularities.ALL) + .setDimensions( + new DefaultDimensionSpec("k", "d0"), + new DefaultDimensionSpec("v", "d1") + ) + .setContext(QUERY_CONTEXT_DEFAULT) + .build() + ), + "j0.", + equalsCondition(makeColumnExpression("dim2"), makeColumnExpression("j0.d0")), + JoinType.LEFT + ) + ) + .intervals(querySegmentSpec(interval)) + .columns("dim1", "j0.d1") + .columnTypes(ColumnType.STRING, ColumnType.STRING) + .context(QUERY_CONTEXT_DEFAULT) + .build() + ), + new LegacySegmentSpec(interval), + QUERY_CONTEXT_DEFAULT, + RowSignature.builder().add("dim1", ColumnType.STRING).add("j0.d1", ColumnType.STRING).build(), + ImmutableList.of( + OperatorFactoryBuilders.naiveSortOperator( + ColumnWithDirection.ascending("j0.d1"), + ColumnWithDirection.ascending("dim1") + ), + OperatorFactoryBuilders.scanOperatorFactoryBuilder().setProjectedColumns("dim1", "j0.d1").build() + ), + null + ) + ), + ImmutableList.of( + new Object[]{"abc", null}, + new Object[]{"1", "xa"}, + new Object[]{"def", "xabc"} + ) + ); + } + + @DecoupledTestConfig(quidemReason = QuidemTestCaseReason.JOIN_FILTER_LOCATIONS) + @Test + public void testJoinWithSubqueryOnLeftAndOuterTimeFilter() + { + // The filter on __time applies to the subquery's output, so it stays a filter rather than becoming intervals. + testBuilder() + .sql( + "SELECT base.dim2, lookyloo.v\n" + + "FROM (SELECT __time, dim2 FROM foo ORDER BY __time LIMIT 10) base\n" + + "LEFT JOIN lookup.lookyloo ON base.dim2 = lookyloo.k\n" + + "WHERE base.__time >= TIMESTAMP '2000-01-03 00:00:00'" + ) + .expectedQueries( + ImmutableList.of( + newScanQueryBuilder() + .dataSource( + join( + new QueryDataSource( + newScanQueryBuilder() + .dataSource(CalciteTests.DATASOURCE1) + .intervals(querySegmentSpec(Filtration.eternity())) + .columns("__time", "dim2") + .columnTypes(ColumnType.LONG, ColumnType.STRING) + .order(Order.ASCENDING) + .limit(10) + .context(QUERY_CONTEXT_DEFAULT) + .build() + ), + new LookupDataSource("lookyloo"), + "j0.", + equalsCondition(makeColumnExpression("dim2"), makeColumnExpression("j0.k")), + JoinType.LEFT + ) + ) + .intervals(querySegmentSpec(Filtration.eternity())) + .filters(range("__time", ColumnType.LONG, timestamp("2000-01-03"), null, false, false)) + .columns("dim2", "j0.v") + .columnTypes(ColumnType.STRING, ColumnType.STRING) + .context(QUERY_CONTEXT_DEFAULT) + .build() + ) + ) + // Unordered, since sort-merge joins return rows in join key order. + .expectedResults( + new UnorderedResultsVerifier( + ImmutableList.of( + new Object[]{"", null}, + new Object[]{"a", "xa"}, + new Object[]{"abc", "xabc"}, + new Object[]{null, null} + ), + ResultMatchMode.EQUALS, + null + ) + ) + .run(); + } } diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteReplaceDmlTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteReplaceDmlTest.java index 5e71878a4269..7648f339ce3b 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteReplaceDmlTest.java +++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteReplaceDmlTest.java @@ -1112,4 +1112,19 @@ public void testReplaceWithSqlOuterLimit() )) .verify(); } + + @Test + public void testReplaceWithSqlTimeCondition() + { + HashMap context = new HashMap<>(DEFAULT_CONTEXT); + context.put(PlannerContext.CTX_SQL_TIME_CONDITION, "2000/2001"); + + testIngestionQuery() + .context(context) + .sql("REPLACE INTO dst OVERWRITE ALL SELECT * FROM foo PARTITIONED BY ALL TIME") + .expectValidationError(DruidExceptionMatcher.invalidInput().expectMessageIs( + "Context parameter [sqlTimeCondition] cannot be provided on operator [REPLACE]" + )) + .verify(); + } } diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java new file mode 100644 index 000000000000..9646b95cb331 --- /dev/null +++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java @@ -0,0 +1,569 @@ +/* + * 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.druid.sql.calcite; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import org.apache.druid.error.DruidException; +import org.apache.druid.error.DruidExceptionMatcher; +import org.apache.druid.java.util.common.Intervals; +import org.apache.druid.java.util.common.granularity.Granularities; +import org.apache.druid.query.Druids; +import org.apache.druid.query.GlobalTableDataSource; +import org.apache.druid.query.LookupDataSource; +import org.apache.druid.query.QueryContexts; +import org.apache.druid.query.QueryDataSource; +import org.apache.druid.query.TableDataSource; +import org.apache.druid.query.UnionDataSource; +import org.apache.druid.query.UnnestDataSource; +import org.apache.druid.query.aggregation.CountAggregatorFactory; +import org.apache.druid.query.aggregation.LongSumAggregatorFactory; +import org.apache.druid.query.dimension.DefaultDimensionSpec; +import org.apache.druid.query.groupby.GroupByQuery; +import org.apache.druid.query.timeboundary.TimeBoundaryQuery; +import org.apache.druid.segment.column.ColumnType; +import org.apache.druid.segment.join.JoinType; +import org.apache.druid.sql.calcite.NotYetSupported.Modes; +import org.apache.druid.sql.calcite.filtration.Filtration; +import org.apache.druid.sql.calcite.planner.PlannerContext; +import org.apache.druid.sql.calcite.util.CalciteTests; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Map; + +/** + * Tests for the {@link PlannerContext#CTX_SQL_TIME_CONDITION} context parameter. + */ +public class CalciteTimeConditionQueryTest extends BaseCalciteQueryTest +{ + // __time for foo is [2000-01-01, 2000-01-02, 2000-01-03, 2001-01-01, 2001-01-02, 2001-01-03] + + @Test + public void testScan() + { + final Map context = makeContext("2000-01-02/2000-01-03, 2001-01-01/2001-01-02"); + + testBuilder() + .sql("SELECT __time, dim1 FROM foo") + .queryContext(context) + .expectedQueries( + ImmutableList.of( + newScanQueryBuilder() + .dataSource(CalciteTests.DATASOURCE1) + .intervals( + querySegmentSpec( + Intervals.of("2000-01-02/2000-01-03"), + Intervals.of("2001-01-01/2001-01-02") + ) + ) + .columns("__time", "dim1") + .columnTypes(ColumnType.LONG, ColumnType.STRING) + .context(context) + .build() + ) + ) + // Unordered, since MSQ does not guarantee the order of scan results across segments. + .expectedResults( + new UnorderedResultsVerifier( + ImmutableList.of( + new Object[]{timestamp("2000-01-02"), "10.1"}, + new Object[]{timestamp("2001-01-01"), "1"} + ), + ResultMatchMode.EQUALS, + null + ) + ) + .run(); + } + + @Test + public void testScanWithEmptyIntersection() + { + final Map context = makeContext("2000/2001"); + + testQuery( + "SELECT dim1 FROM foo WHERE __time >= TIMESTAMP '2001-01-01 00:00:00'", + context, + ImmutableList.of( + newScanQueryBuilder() + .dataSource(CalciteTests.DATASOURCE1) + .intervals(querySegmentSpec()) + .columns("dim1") + .columnTypes(ColumnType.STRING) + .context(context) + .build() + ), + ImmutableList.of() + ); + } + + @Test + public void testTimeseriesWithTimeFilter() + { + final Map context = makeContext("2000/2001"); + + testQuery( + "SELECT COUNT(*) FROM foo WHERE __time >= TIMESTAMP '2000-01-02 00:00:00'", + context, + ImmutableList.of( + Druids.newTimeseriesQueryBuilder() + .dataSource(CalciteTests.DATASOURCE1) + .intervals(querySegmentSpec(Intervals.of("2000-01-02/2001-01-01"))) + .granularity(Granularities.ALL) + .aggregators(aggregators(new CountAggregatorFactory("a0"))) + .context(context) + .build() + ), + ImmutableList.of( + new Object[]{2L} + ) + ); + } + + @Test + public void testGroupByWithArray() + { + final Map context = makeContext(List.of("2001-01-01/2001-01-03", "2000-01-01/2000-01-02")); + + testQuery( + "SELECT dim2, COUNT(*) FROM foo GROUP BY dim2", + context, + ImmutableList.of( + GroupByQuery.builder() + .setDataSource(CalciteTests.DATASOURCE1) + .setInterval( + querySegmentSpec( + Intervals.of("2000-01-01/2000-01-02"), + Intervals.of("2001-01-01/2001-01-03") + ) + ) + .setGranularity(Granularities.ALL) + .setDimensions(dimensions(new DefaultDimensionSpec("dim2", "d0"))) + .setAggregatorSpecs(aggregators(new CountAggregatorFactory("a0"))) + .setContext(context) + .build() + ), + ImmutableList.of( + new Object[]{"a", 2L}, + new Object[]{"abc", 1L} + ) + ); + } + + @Test + public void testTimeBoundary() + { + final Map context = ImmutableMap.builder() + .putAll(makeContext("2000/2001")) + .put(QueryContexts.TIME_BOUNDARY_PLANNING_KEY, true) + .build(); + + testQuery( + "SELECT MAX(__time) FROM foo", + context, + ImmutableList.of( + Druids.newTimeBoundaryQueryBuilder() + .dataSource(CalciteTests.DATASOURCE1) + .intervals(querySegmentSpec(Intervals.of("2000/2001"))) + .bound(TimeBoundaryQuery.MAX_TIME) + .context( + ImmutableMap.builder() + .putAll(context) + .put(TimeBoundaryQuery.MAX_TIME_ARRAY_OUTPUT_NAME, "a0") + .build() + ) + .build() + ), + ImmutableList.of( + new Object[]{timestamp("2000-01-03")} + ) + ); + } + + @NotYetSupported(Modes.DD_UNION) + @Test + public void testUnionDataSource() + { + final Map context = makeContext("2000-01-01/2000-01-03"); + + testQuery( + "SELECT dim1, COUNT(*) FROM (SELECT * FROM foo UNION ALL SELECT * FROM foo) GROUP BY dim1", + context, + ImmutableList.of( + GroupByQuery.builder() + .setDataSource( + new UnionDataSource( + ImmutableList.of( + new TableDataSource(CalciteTests.DATASOURCE1), + new TableDataSource(CalciteTests.DATASOURCE1) + ) + ) + ) + .setInterval(querySegmentSpec(Intervals.of("2000-01-01/2000-01-03"))) + .setGranularity(Granularities.ALL) + .setDimensions(dimensions(new DefaultDimensionSpec("dim1", "d0"))) + .setAggregatorSpecs(aggregators(new CountAggregatorFactory("a0"))) + .setContext(context) + .build() + ), + ImmutableList.of( + new Object[]{"", 2L}, + new Object[]{"10.1", 2L} + ) + ); + } + + @Test + public void testUnnest() + { + final Map context = makeContext("2000-01-01/2000-01-03"); + + testQuery( + "SELECT d3 FROM foo, UNNEST(MV_TO_ARRAY(dim3)) AS unnested (d3)", + context, + ImmutableList.of( + newScanQueryBuilder() + .dataSource( + UnnestDataSource.create( + new TableDataSource(CalciteTests.DATASOURCE1), + nestedExpressionVirtualColumn("j0.unnest", "\"dim3\"", ColumnType.STRING), + null + ) + ) + .intervals(querySegmentSpec(Intervals.of("2000-01-01/2000-01-03"))) + .columns("j0.unnest") + .columnTypes(ColumnType.STRING) + .context(context) + .build() + ), + ImmutableList.of( + new Object[]{"a"}, + new Object[]{"b"}, + new Object[]{"b"}, + new Object[]{"c"} + ) + ); + } + + @Test + public void testJoinWithLookup() + { + final Map context = makeContext("2001/2002"); + + testQuery( + "SELECT foo.dim1, lookyloo.v FROM foo INNER JOIN lookup.lookyloo ON foo.dim2 = lookyloo.k", + context, + ImmutableList.of( + newScanQueryBuilder() + .dataSource( + join( + new TableDataSource(CalciteTests.DATASOURCE1), + new LookupDataSource("lookyloo"), + "j0.", + equalsCondition(makeColumnExpression("dim2"), makeColumnExpression("j0.k")), + JoinType.INNER + ) + ) + .intervals(querySegmentSpec(Intervals.of("2001/2002"))) + .columns("dim1", "j0.v") + .columnTypes(ColumnType.STRING, ColumnType.STRING) + .context(context) + .build() + ), + ImmutableList.of( + new Object[]{"1", "xa"}, + new Object[]{"def", "xabc"} + ) + ); + } + + @Test + public void testJoinWithTableSubquery() + { + final Map context = makeContext("2000/2001"); + + testQuery( + "SELECT f1.dim1, f2.dim2 FROM foo f1 INNER JOIN foo f2 ON f1.dim1 = f2.dim1", + context, + ImmutableList.of( + newScanQueryBuilder() + .dataSource( + join( + new TableDataSource(CalciteTests.DATASOURCE1), + new QueryDataSource( + newScanQueryBuilder() + .dataSource(CalciteTests.DATASOURCE1) + .intervals(querySegmentSpec(Intervals.of("2000/2001"))) + .columns("dim1", "dim2") + .columnTypes(ColumnType.STRING, ColumnType.STRING) + .context(context) + .build() + ), + "j0.", + equalsCondition(makeColumnExpression("dim1"), makeColumnExpression("j0.dim1")), + JoinType.INNER + ) + ) + .intervals(querySegmentSpec(Intervals.of("2000/2001"))) + .columns("dim1", "j0.dim2") + .columnTypes(ColumnType.STRING, ColumnType.STRING) + .context(context) + .build() + ), + ImmutableList.of( + new Object[]{"", "a"}, + new Object[]{"10.1", null}, + new Object[]{"2", ""} + ) + ); + } + + @Test + public void testJoinWithBroadcastTable() + { + // Cannot vectorize JOIN operator. + cannotVectorize(); + + final Map context = makeContext("2000-01-01/2001-01-02"); + + // The broadcast table on the right-hand side is read through a subquery, so the time condition applies to it. + testQuery( + "SELECT broadcast.dim4, COUNT(*)\n" + + "FROM numfoo INNER JOIN broadcast ON numfoo.dim4 = broadcast.dim4\n" + + "GROUP BY 1", + context, + ImmutableList.of( + GroupByQuery.builder() + .setDataSource( + join( + new TableDataSource(CalciteTests.DATASOURCE3), + new QueryDataSource( + newScanQueryBuilder() + .dataSource(new GlobalTableDataSource(CalciteTests.BROADCAST_DATASOURCE)) + .intervals(querySegmentSpec(Intervals.of("2000-01-01/2001-01-02"))) + .columns("dim4") + .columnTypes(ColumnType.STRING) + .context(context) + .build() + ), + "j0.", + equalsCondition(makeColumnExpression("dim4"), makeColumnExpression("j0.dim4")), + JoinType.INNER + ) + ) + .setInterval(querySegmentSpec(Intervals.of("2000-01-01/2001-01-02"))) + .setGranularity(Granularities.ALL) + .setDimensions(dimensions(new DefaultDimensionSpec("j0.dim4", "d0"))) + .setAggregatorSpecs(aggregators(new CountAggregatorFactory("a0"))) + .setContext(context) + .build() + ), + ImmutableList.of( + new Object[]{"a", 9L}, + new Object[]{"b", 1L} + ) + ); + } + + @Test + public void testJoinWithSubqueryOnLeft() + { + final Map context = makeContext("2000-01-01/2000-01-02,2000-01-03/2001"); + + // Time condition applies to the subquery on the left-hand side, but not the outer query, which reads the join. + testQuery( + "SELECT base.dim2, lookyloo.v, base.cnt\n" + + "FROM (SELECT dim2, COUNT(*) AS cnt FROM foo GROUP BY dim2) base\n" + + "LEFT JOIN lookup.lookyloo ON base.dim2 = lookyloo.k", + context, + ImmutableList.of( + newScanQueryBuilder() + .dataSource( + join( + new QueryDataSource( + GroupByQuery.builder() + .setDataSource(CalciteTests.DATASOURCE1) + .setInterval( + querySegmentSpec( + Intervals.of("2000-01-01/2000-01-02"), + Intervals.of("2000-01-03/2001") + ) + ) + .setGranularity(Granularities.ALL) + .setDimensions(dimensions(new DefaultDimensionSpec("dim2", "d0"))) + .setAggregatorSpecs(aggregators(new CountAggregatorFactory("a0"))) + .setContext(context) + .build() + ), + new LookupDataSource("lookyloo"), + "j0.", + equalsCondition(makeColumnExpression("d0"), makeColumnExpression("j0.k")), + JoinType.LEFT + ) + ) + .intervals(querySegmentSpec(Filtration.eternity())) + .columns("d0", "j0.v", "a0") + .columnTypes(ColumnType.STRING, ColumnType.STRING, ColumnType.LONG) + .context(context) + .build() + ), + ImmutableList.of( + new Object[]{"", null, 1L}, + new Object[]{"a", "xa", 1L} + ) + ); + } + + @Test + public void testJoinWithNonTimeOrderBy() + { + // Cannot vectorize direct queries on lookup tables. + cannotVectorize(); + + final Map context = makeContext("2001/2002"); + + testBuilder() + .sql( + "SELECT f1.dim1, f2.v\n" + + "FROM foo f1 JOIN (SELECT k, v FROM lookup.lookyloo GROUP BY k, v) f2 ON f1.dim2 = f2.k\n" + + "ORDER BY f1.dim1" + ) + .queryContext(context) + .expectedResults( + ImmutableList.of( + new Object[]{"1", "xa"}, + new Object[]{"def", "xabc"} + ) + ) + .run(); + } + + @NotYetSupported(Modes.DD_WINDOW) + @Test + public void testWindowOverJoin() + { + // Cannot vectorize direct queries on lookup tables. + cannotVectorize(); + + final Map context = makeContext("2000-01-01/2000-01-02"); + + testBuilder() + .sql( + "SELECT f1.__time, f1.dim2, COUNT(*) OVER () AS c\n" + + "FROM foo f1 JOIN (SELECT k FROM lookup.lookyloo GROUP BY k) f2 ON f1.dim2 = f2.k" + ) + .queryContext(context) + .expectedResults( + ImmutableList.of( + new Object[]{timestamp("2000-01-01"), "a", 1L} + ) + ) + .run(); + } + + @Test + public void testSubquery() + { + final Map context = makeContext("2000/2001"); + + // Time condition applies to the inner query, which is on a table, but not the outer query. + testQuery( + "SELECT SUM(cnt), COUNT(*) FROM (SELECT dim2, SUM(cnt) AS cnt FROM foo GROUP BY dim2)", + context, + ImmutableList.of( + GroupByQuery.builder() + .setDataSource( + new QueryDataSource( + GroupByQuery.builder() + .setDataSource(CalciteTests.DATASOURCE1) + .setInterval(querySegmentSpec(Intervals.of("2000/2001"))) + .setGranularity(Granularities.ALL) + .setDimensions(dimensions(new DefaultDimensionSpec("dim2", "d0"))) + .setAggregatorSpecs(aggregators(new LongSumAggregatorFactory("a0", "cnt"))) + .setContext(context) + .build() + ) + ) + .setInterval(querySegmentSpec(Filtration.eternity())) + .setGranularity(Granularities.ALL) + .setAggregatorSpecs( + aggregators( + new LongSumAggregatorFactory("_a0", "a0"), + new CountAggregatorFactory("_a1") + ) + ) + .setContext(context) + .build() + ), + ImmutableList.of( + new Object[]{3L, 3L} + ) + ); + } + + @Test + public void testLookupIsUnaffected() + { + // Cannot vectorize direct queries on lookup tables. + cannotVectorize(); + + final Map context = makeContext("2000/2001"); + + testQuery( + "SELECT COUNT(*) FROM lookup.lookyloo", + context, + ImmutableList.of( + Druids.newTimeseriesQueryBuilder() + .dataSource(new LookupDataSource("lookyloo")) + .intervals(querySegmentSpec(Filtration.eternity())) + .granularity(Granularities.ALL) + .aggregators(aggregators(new CountAggregatorFactory("a0"))) + .context(context) + .build() + ), + ImmutableList.of( + new Object[]{4L} + ) + ); + } + + @Test + public void testInvalidTimeCondition() + { + testQueryThrows( + "SELECT COUNT(*) FROM foo", + makeContext("2000/2001,foo"), + DruidException.class, + DruidExceptionMatcher.invalidInput().expectMessageIs( + "Query context parameter [sqlTimeCondition] must be a comma-separated string of ISO8601 intervals, " + + "or an array of ISO8601 interval strings, but got [2000/2001,foo]" + ) + ); + } + + protected static Map makeContext(final Object timeCondition) + { + return ImmutableMap.builder() + .putAll(QUERY_CONTEXT_DEFAULT) + .put(PlannerContext.CTX_SQL_TIME_CONDITION, timeCondition) + .build(); + } +} diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/DecoupledPlanningCalciteTimeConditionQueryTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/DecoupledPlanningCalciteTimeConditionQueryTest.java new file mode 100644 index 000000000000..ca2242b2d1fe --- /dev/null +++ b/sql/src/test/java/org/apache/druid/sql/calcite/DecoupledPlanningCalciteTimeConditionQueryTest.java @@ -0,0 +1,41 @@ +/* + * 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.druid.sql.calcite; + +import org.apache.druid.sql.calcite.NotYetSupported.NotYetSupportedProcessor; +import org.junit.jupiter.api.extension.RegisterExtension; + +/** + * Runs {@link CalciteTimeConditionQueryTest} with the decoupled planner. + */ +public class DecoupledPlanningCalciteTimeConditionQueryTest extends CalciteTimeConditionQueryTest +{ + @RegisterExtension + NotYetSupportedProcessor notYetSupportedProcessor = new NotYetSupportedProcessor(NotYetSupported.Scope.DECOUPLED); + + @RegisterExtension + DecoupledExtension decoupledExtension = new DecoupledExtension(this); + + @Override + protected QueryTestBuilder testBuilder() + { + return decoupledExtension.testBuilder(); + } +} diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/planner/PlannerContextTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/planner/PlannerContextTest.java new file mode 100644 index 000000000000..a98fc108e2c6 --- /dev/null +++ b/sql/src/test/java/org/apache/druid/sql/calcite/planner/PlannerContextTest.java @@ -0,0 +1,137 @@ +/* + * 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.druid.sql.calcite.planner; + +import org.apache.druid.error.DruidException; +import org.apache.druid.error.DruidExceptionMatcher; +import org.apache.druid.java.util.common.Intervals; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +/** + * Tests for {@link PlannerContext}. + */ +public class PlannerContextTest +{ + @Test + public void test_parseTimeCondition_null() + { + Assertions.assertNull(PlannerContext.parseTimeCondition(null)); + } + + @Test + public void test_parseTimeCondition_singleInterval() + { + Assertions.assertEquals( + List.of(Intervals.of("2000-01-01/2000-01-02")), + PlannerContext.parseTimeCondition("2000-01-01/2000-01-02") + ); + } + + @Test + public void test_parseTimeCondition_commaSeparated() + { + // Out of order, with whitespace, and with overlapping and abutting intervals. + Assertions.assertEquals( + List.of(Intervals.of("2000-01-01/2000-01-04"), Intervals.of("2001-01-01/2001-01-02")), + PlannerContext.parseTimeCondition( + "2001-01-01/2001-01-02, 2000-01-02/2000-01-03 ,2000-01-01/2000-01-02,2000-01-02T12/2000-01-04" + ) + ); + } + + @Test + public void test_parseTimeCondition_array() + { + Assertions.assertEquals( + List.of(Intervals.of("2000-01-01/2000-01-02"), Intervals.of("2001-01-01/2001-01-02")), + PlannerContext.parseTimeCondition(List.of("2001-01-01/2001-01-02", "2000-01-01/2000-01-02")) + ); + } + + @Test + public void test_parseTimeCondition_timeZoneOffset() + { + Assertions.assertEquals( + List.of(Intervals.of("2000-01-01T08:00:00Z/2000-01-02T08:00:00Z")), + PlannerContext.parseTimeCondition("2000-01-01T00:00:00-08:00/2000-01-02T00:00:00-08:00") + ); + } + + @Test + public void test_parseTimeCondition_emptyArray() + { + Assertions.assertEquals( + Collections.emptyList(), + PlannerContext.parseTimeCondition(Collections.emptyList()) + ); + } + + @Test + public void test_parseTimeCondition_emptyString() + { + assertInvalidTimeCondition("", ""); + } + + @Test + public void test_parseTimeCondition_onlyComma() + { + assertInvalidTimeCondition(",", ","); + } + + @Test + public void test_parseTimeCondition_trailingComma() + { + assertInvalidTimeCondition("2000-01-01/2000-01-02,", "2000-01-01/2000-01-02,"); + } + + @Test + public void test_parseTimeCondition_invalidInterval() + { + assertInvalidTimeCondition("2000-01-01/2000-01-02,foo", "2000-01-01/2000-01-02,foo"); + } + + @Test + public void test_parseTimeCondition_nonStringArrayElement() + { + assertInvalidTimeCondition(Arrays.asList("2000-01-01/2000-01-02", 3), "[2000-01-01/2000-01-02, 3]"); + } + + @Test + public void test_parseTimeCondition_wrongType() + { + assertInvalidTimeCondition(3, "3"); + } + + private static void assertInvalidTimeCondition(final Object value, final String expectedValueString) + { + DruidExceptionMatcher.assertThat( + Assertions.assertThrows(DruidException.class, () -> PlannerContext.parseTimeCondition(value)), + DruidExceptionMatcher.invalidInput().expectMessageIs( + "Query context parameter [sqlTimeCondition] must be a comma-separated string of ISO8601 intervals, " + + "or an array of ISO8601 interval strings, but got [" + expectedValueString + "]" + ) + ); + } +} diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/rel/DruidQueryTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/rel/DruidQueryTest.java index 10c98ea9d312..2ed50f561238 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/rel/DruidQueryTest.java +++ b/sql/src/test/java/org/apache/druid/sql/calcite/rel/DruidQueryTest.java @@ -23,9 +23,11 @@ import org.apache.druid.java.util.common.Pair; import org.apache.druid.math.expr.ExprMacroTable; import org.apache.druid.query.DataSource; +import org.apache.druid.query.Druids; import org.apache.druid.query.FilteredDataSource; import org.apache.druid.query.JoinAlgorithm; import org.apache.druid.query.JoinDataSource; +import org.apache.druid.query.QueryDataSource; import org.apache.druid.query.TableDataSource; import org.apache.druid.query.UnionDataSource; import org.apache.druid.query.expression.TestExprMacroTable; @@ -34,6 +36,7 @@ import org.apache.druid.query.filter.DimFilter; import org.apache.druid.query.filter.SelectorDimFilter; import org.apache.druid.query.ordering.StringComparators; +import org.apache.druid.query.spec.MultipleIntervalSegmentSpec; import org.apache.druid.segment.column.RowSignature; import org.apache.druid.segment.join.JoinType; import org.apache.druid.sql.calcite.filtration.Filtration; @@ -64,6 +67,7 @@ public void test_filtration_noJoinAndInterval() Pair pair = DruidQuery.getFiltration( dataSource, selectorFilter, + null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -82,6 +86,7 @@ public void test_filtration_intervalInQueryFilter() Pair pair = DruidQuery.getFiltration( dataSource, filterWithInterval, + null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -100,6 +105,7 @@ public void test_filtration_joinDataSource_intervalInQueryFilter() Pair pair = DruidQuery.getFiltration( dataSource, filterWithInterval, + null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -119,6 +125,7 @@ public void test_filtration_joinDataSource_intervalInBaseTableFilter_inner() Pair pair = DruidQuery.getFiltration( dataSource, otherFilter, + null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -138,6 +145,7 @@ public void test_filtration_joinDataSource_intervalInBaseTableFilter_left() Pair pair = DruidQuery.getFiltration( dataSource, otherFilter, + null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -157,6 +165,7 @@ public void test_filtration_joinDataSource_intervalInBaseTableFilter_right() Pair pair = DruidQuery.getFiltration( dataSource, otherFilter, + null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -176,6 +185,7 @@ public void test_filtration_joinDataSource_intervalInBaseTableFilter_full() Pair pair = DruidQuery.getFiltration( dataSource, otherFilter, + null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -200,6 +210,7 @@ public void test_filtration_intervalsInBothFilters() Pair pair = DruidQuery.getFiltration( dataSource, queryFilter, + null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -229,6 +240,7 @@ public void test_filtration_filteredUnionDataSource_intervalsInArmFilters() final Pair pair = DruidQuery.getFiltration( dataSource, otherFilter, + null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -242,6 +254,145 @@ public void test_filtration_filteredUnionDataSource_intervalsInArmFilters() Assertions.assertEquals(List.of(Intervals.utc(100, 200), Intervals.utc(300, 400)), pair.rhs.getIntervals()); } + @Test + public void test_filtration_timeCondition_noIntervalInQueryFilter() + { + DataSource dataSource = new TableDataSource("test"); + Pair pair = DruidQuery.getFiltration( + dataSource, + selectorFilter, + Collections.singletonList(Intervals.utc(50, 150)), + VirtualColumnRegistry.create( + RowSignature.empty(), + new ExpressionParserImpl(TestExprMacroTable.INSTANCE), + false, + Set.of() + ), + CalciteTests.createJoinableFactoryWrapper() + ); + verify(pair, dataSource, selectorFilter, Intervals.utc(50, 150)); + } + + @Test + public void test_filtration_timeCondition_noQueryFilter() + { + DataSource dataSource = new TableDataSource("test"); + Pair pair = DruidQuery.getFiltration( + dataSource, + null, + Collections.singletonList(Intervals.utc(50, 150)), + VirtualColumnRegistry.create( + RowSignature.empty(), + new ExpressionParserImpl(TestExprMacroTable.INSTANCE), + false, + Set.of() + ), + CalciteTests.createJoinableFactoryWrapper() + ); + verify(pair, dataSource, null, Intervals.utc(50, 150)); + } + + @Test + public void test_filtration_timeCondition_intervalInQueryFilter() + { + DataSource dataSource = new TableDataSource("test"); + Pair pair = DruidQuery.getFiltration( + dataSource, + filterWithInterval, + Collections.singletonList(Intervals.utc(150, 300)), + VirtualColumnRegistry.create( + RowSignature.empty(), + new ExpressionParserImpl(TestExprMacroTable.INSTANCE), + false, + Set.of() + ), + CalciteTests.createJoinableFactoryWrapper() + ); + verify(pair, dataSource, selectorFilter, Intervals.utc(150, 200)); + } + + @Test + public void test_filtration_timeCondition_joinDataSource_intervalInBaseTableFilter() + { + DataSource dataSource = join(JoinType.INNER, filterWithInterval); + DataSource expectedDataSource = join(JoinType.INNER, selectorFilter); + Pair pair = DruidQuery.getFiltration( + dataSource, + otherFilter, + Collections.singletonList(Intervals.utc(50, 150)), + VirtualColumnRegistry.create( + RowSignature.empty(), + new ExpressionParserImpl(TestExprMacroTable.INSTANCE), + false, + Set.of() + ), + CalciteTests.createJoinableFactoryWrapper() + ); + verify(pair, expectedDataSource, otherFilter, Intervals.utc(100, 150)); + } + + @Test + public void test_filtration_timeCondition_queryDataSource() + { + DataSource dataSource = new QueryDataSource( + Druids.newScanQueryBuilder() + .dataSource("test") + .intervals(new MultipleIntervalSegmentSpec(Intervals.ONLY_ETERNITY)) + .build() + ); + Pair pair = DruidQuery.getFiltration( + dataSource, + selectorFilter, + Collections.singletonList(Intervals.utc(150, 300)), + VirtualColumnRegistry.create( + RowSignature.empty(), + new ExpressionParserImpl(TestExprMacroTable.INSTANCE), + false, + Set.of() + ), + CalciteTests.createJoinableFactoryWrapper() + ); + + // Time condition is not applied, since the datasource is not table-based. + verify(pair, dataSource, selectorFilter, Intervals.ETERNITY); + } + + @Test + public void test_filtration_timeCondition_joinDataSource_subqueryOnLeft() + { + DataSource dataSource = JoinDataSource.create( + new QueryDataSource( + Druids.newScanQueryBuilder() + .dataSource("left") + .intervals(new MultipleIntervalSegmentSpec(Intervals.ONLY_ETERNITY)) + .build() + ), + new TableDataSource("right"), + "r.", + "c == \"r.c\"", + JoinType.INNER, + null, + ExprMacroTable.nil(), + CalciteTests.createJoinableFactoryWrapper(), + JoinAlgorithm.BROADCAST + ); + Pair pair = DruidQuery.getFiltration( + dataSource, + selectorFilter, + Collections.singletonList(Intervals.utc(150, 300)), + VirtualColumnRegistry.create( + RowSignature.empty(), + new ExpressionParserImpl(TestExprMacroTable.INSTANCE), + false, + Set.of() + ), + CalciteTests.createJoinableFactoryWrapper() + ); + + // Time condition is not applied, since the base of the join is a subquery, which is not table-based. + verify(pair, dataSource, selectorFilter, Intervals.ETERNITY); + } + private JoinDataSource join(JoinType joinType, DimFilter filter) { return JoinDataSource.create( diff --git a/sql/src/test/quidem/org.apache.druid.sql.calcite.DecoupledPlanningCalciteJoinQueryTest/testJoinWithSubqueryOnLeftAndOuterTimeFilter.iq b/sql/src/test/quidem/org.apache.druid.sql.calcite.DecoupledPlanningCalciteJoinQueryTest/testJoinWithSubqueryOnLeftAndOuterTimeFilter.iq new file mode 100644 index 000000000000..45d0aa1e9a1a --- /dev/null +++ b/sql/src/test/quidem/org.apache.druid.sql.calcite.DecoupledPlanningCalciteJoinQueryTest/testJoinWithSubqueryOnLeftAndOuterTimeFilter.iq @@ -0,0 +1,103 @@ +# testJoinWithSubqueryOnLeftAndOuterTimeFilter case-crc:229dd4c0 +# quidem testcase reason: JOIN_FILTER_LOCATIONS +!set debug true +!set defaultTimeout 300000 +!set maxScatterGatherBytes 9223372036854775807 +!set plannerStrategy DECOUPLED +!set sqlCurrentTimestamp 2000-01-01T00:00:00Z +!set sqlQueryId dummy +!set outputformat mysql +!use druidtest:/// +SELECT base.dim2, lookyloo.v +FROM (SELECT __time, dim2 FROM foo ORDER BY __time LIMIT 10) base +LEFT JOIN lookup.lookyloo ON base.dim2 = lookyloo.k +WHERE base.__time >= TIMESTAMP '2000-01-03 00:00:00'; +LogicalProject(dim2=[$1], v=[$3]) + LogicalJoin(condition=[=($1, $2)], joinType=[left]) + LogicalFilter(condition=[>=($0, 2000-01-03 00:00:00)]) + LogicalSort(sort0=[$0], dir0=[ASC], fetch=[10]) + LogicalProject(__time=[$0], dim2=[$2]) + LogicalTableScan(table=[[druid, foo]]) + LogicalTableScan(table=[[lookup, lookyloo]]) + +!logicalPlan +DruidProject(dim2=[$1], v=[$3], druid=[logical]) + DruidJoin(condition=[=($1, $2)], joinType=[left]) + DruidFilter(condition=[>=($0, 2000-01-03 00:00:00)]) + DruidSort(sort0=[$0], dir0=[ASC], fetch=[10], druid=[logical]) + DruidProject(__time=[$0], dim2=[$2], druid=[logical]) + DruidTableScan(table=[[druid, foo]], druid=[logical]) + DruidTableScan(table=[[lookup, lookyloo]], druid=[logical]) + +!druidPlan +{ + "queryType" : "scan", + "dataSource" : { + "type" : "join", + "left" : { + "type" : "query", + "query" : { + "queryType" : "scan", + "dataSource" : { + "type" : "query", + "query" : { + "queryType" : "scan", + "dataSource" : { + "type" : "table", + "name" : "foo" + }, + "intervals" : { + "type" : "intervals", + "intervals" : [ "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z" ] + }, + "resultFormat" : "compactedList", + "limit" : 10, + "order" : "ascending", + "columns" : [ "__time", "dim2" ], + "columnTypes" : [ "LONG", "STRING" ], + "granularity" : { + "type" : "all" + }, + "legacy" : false + } + }, + "intervals" : { + "type" : "intervals", + "intervals" : [ "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z" ] + }, + "resultFormat" : "compactedList", + "filter" : { + "type" : "range", + "column" : "__time", + "matchValueType" : "LONG", + "lower" : 946857600000 + }, + "columns" : [ "__time", "dim2" ], + "columnTypes" : [ "LONG", "STRING" ], + "granularity" : { + "type" : "all" + }, + "legacy" : false + } + }, + "right" : { + "type" : "lookup", + "lookup" : "lookyloo" + }, + "rightPrefix" : "j0.", + "condition" : "(\"dim2\" == \"j0.k\")", + "joinType" : "LEFT" + }, + "intervals" : { + "type" : "intervals", + "intervals" : [ "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z" ] + }, + "resultFormat" : "compactedList", + "columns" : [ "dim2", "j0.v" ], + "columnTypes" : [ "STRING", "STRING" ], + "granularity" : { + "type" : "all" + }, + "legacy" : false +} +!nativePlan diff --git a/sql/src/test/quidem/org.apache.druid.sql.calcite.DecoupledPlanningCalciteJoinQueryTest/testLeftJoinWithTimeFilterAndNonTimeOrderBy.iq b/sql/src/test/quidem/org.apache.druid.sql.calcite.DecoupledPlanningCalciteJoinQueryTest/testLeftJoinWithTimeFilterAndNonTimeOrderBy.iq new file mode 100644 index 000000000000..afda25015f77 --- /dev/null +++ b/sql/src/test/quidem/org.apache.druid.sql.calcite.DecoupledPlanningCalciteJoinQueryTest/testLeftJoinWithTimeFilterAndNonTimeOrderBy.iq @@ -0,0 +1,150 @@ +# testLeftJoinWithTimeFilterAndNonTimeOrderBy case-crc:b7c5b25b +# quidem testcase reason: JOIN_FILTER_LOCATIONS +!set debug true +!set defaultTimeout 300000 +!set maxScatterGatherBytes 9223372036854775807 +!set plannerStrategy DECOUPLED +!set sqlCurrentTimestamp 2000-01-01T00:00:00Z +!set sqlQueryId dummy +!set outputformat mysql +!use druidtest:/// +SELECT f1.dim1, f2.v +FROM foo f1 LEFT JOIN (SELECT k, v FROM lookup.lookyloo GROUP BY k, v) f2 ON f1.dim2 = f2.k +WHERE f1.__time >= TIMESTAMP '2001-01-01 00:00:00' +ORDER BY f2.v, f1.dim1; ++------+------+ +| dim1 | v | ++------+------+ +| abc | | +| 1 | xa | +| def | xabc | ++------+------+ +(3 rows) + +!ok +LogicalProject(dim1=[$0], v=[$3]) + LogicalSort(sort0=[$3], sort1=[$0], dir0=[ASC], dir1=[ASC]) + LogicalJoin(condition=[=($1, $2)], joinType=[left]) + LogicalProject(dim1=[$1], dim2=[$2]) + LogicalFilter(condition=[>=($0, 2001-01-01 00:00:00)]) + LogicalTableScan(table=[[druid, foo]]) + LogicalAggregate(group=[{0, 1}]) + LogicalTableScan(table=[[lookup, lookyloo]]) + +!logicalPlan +DruidProject(dim1=[$0], v=[$3], druid=[logical]) + DruidSort(sort0=[$3], sort1=[$0], dir0=[ASC], dir1=[ASC], druid=[logical]) + DruidJoin(condition=[=($1, $2)], joinType=[left]) + DruidProject(dim1=[$1], dim2=[$2], druid=[logical]) + DruidFilter(condition=[>=($0, 2001-01-01 00:00:00)]) + DruidTableScan(table=[[druid, foo]], druid=[logical]) + DruidAggregate(group=[{0, 1}], druid=[logical]) + DruidTableScan(table=[[lookup, lookyloo]], druid=[logical]) + +!druidPlan +{ + "queryType" : "windowOperator", + "dataSource" : { + "type" : "join", + "left" : { + "type" : "query", + "query" : { + "queryType" : "scan", + "dataSource" : { + "type" : "table", + "name" : "foo" + }, + "intervals" : { + "type" : "intervals", + "intervals" : [ "2001-01-01T00:00:00.000Z/146140482-04-24T15:36:27.903Z" ] + }, + "resultFormat" : "compactedList", + "columns" : [ "dim1", "dim2" ], + "columnTypes" : [ "STRING", "STRING" ], + "granularity" : { + "type" : "all" + }, + "legacy" : false + } + }, + "right" : { + "type" : "query", + "query" : { + "queryType" : "groupBy", + "dataSource" : { + "type" : "lookup", + "lookup" : "lookyloo" + }, + "intervals" : { + "type" : "intervals", + "intervals" : [ "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z" ] + }, + "granularity" : { + "type" : "all" + }, + "dimensions" : [ { + "type" : "default", + "dimension" : "k", + "outputName" : "d0", + "outputType" : "STRING" + }, { + "type" : "default", + "dimension" : "v", + "outputName" : "d1", + "outputType" : "STRING" + } ], + "limitSpec" : { + "type" : "NoopLimitSpec" + } + } + }, + "rightPrefix" : "j0.", + "condition" : "(\"dim2\" == \"j0.d0\")", + "joinType" : "LEFT" + }, + "intervals" : { + "type" : "LegacySegmentSpec", + "intervals" : [ "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z" ] + }, + "outputSignature" : [ { + "name" : "dim1", + "type" : "STRING" + }, { + "name" : "j0.d1", + "type" : "STRING" + } ], + "operatorDefinition" : [ { + "type" : "naiveSort", + "columns" : [ { + "column" : "j0.d1", + "direction" : "ASC" + }, { + "column" : "dim1", + "direction" : "ASC" + } ] + }, { + "type" : "scan", + "timeRange" : null, + "filter" : null, + "offsetLimit" : null, + "projectedColumns" : [ "dim1", "j0.d1" ], + "virtualColumns" : null, + "ordering" : null + } ], + "leafOperators" : [ { + "type" : "scan", + "timeRange" : null, + "filter" : null, + "offsetLimit" : { + "offset" : 0, + "limit" : 9223372036854775807 + }, + "projectedColumns" : [ "dim1", "j0.d1" ], + "virtualColumns" : null, + "ordering" : null + } ], + "granularity" : { + "type" : "all" + } +} +!nativePlan diff --git a/website/.spelling b/website/.spelling index 15124c69ebac..beb29a5181ce 100644 --- a/website/.spelling +++ b/website/.spelling @@ -2111,6 +2111,7 @@ addr affinityConfig allowAll ANDed +ANDs array_mod autoscale autoscalers From 4e7d285da3f53f06846fbde131e1a2949f49c334 Mon Sep 17 00:00:00 2001 From: Gian Merlino Date: Fri, 23 Oct 2026 23:29:09 -0700 Subject: [PATCH 2/6] Changes: - sqlTimeCondition must be a string. - add test for restricted datasources - simplify getFiltration javadoc --- docs/querying/sql-query-context.md | 2 +- .../query/context/QueryContextParameters.java | 10 +- .../sql/calcite/planner/PlannerContext.java | 45 +++------ .../druid/sql/calcite/rel/DruidQuery.java | 7 +- .../sql/calcite/CalciteInsertDmlTest.java | 16 --- .../CalciteTimeConditionQueryTest.java | 98 ++++++++++++++----- .../calcite/planner/PlannerContextTest.java | 36 +------ 7 files changed, 95 insertions(+), 119 deletions(-) diff --git a/docs/querying/sql-query-context.md b/docs/querying/sql-query-context.md index 68955780c948..44660b2e5e9d 100644 --- a/docs/querying/sql-query-context.md +++ b/docs/querying/sql-query-context.md @@ -52,7 +52,7 @@ The table below lists the query context parameters you can use with Druid SQL. |`sqlReverseLookup`|Whether to consider the [reverse-lookup rewrite](lookups.md#reverse-lookup) of the `LOOKUP` function during SQL planning.

Druid reverses calls to `LOOKUP` only when the number of matching keys is lower than both `inSubQueryThreshold` and `sqlReverseLookupThreshold`.|`true`| |`sqlReverseLookupThreshold`|Maximum size of `IN` filter to create when applying a [reverse-lookup rewrite](lookups.md#reverse-lookup). If a `LOOKUP` call matches more keys than the specified threshold, it remains unchanged.

If `inSubQueryThreshold` is lower than `sqlReverseLookupThreshold`, Druid uses `inSubQueryThreshold` threshold instead.|`10000`| |`sqlPullUpLookup`|Whether to consider the [pull-up rewrite](lookups.md#pull-up) of the `LOOKUP` function during SQL planning.|`true`| -|`sqlTimeCondition`|Time condition to apply to all tables that the query reads. Provide either a comma-separated string of ISO8601 intervals, such as `"2000-01-01/2000-02-01,2001-01-01/2001-02-01"`, or an array of ISO8601 interval strings. An empty array matches nothing. An empty string is an error.

Druid ANDs this condition with any other condition you provide on the `__time` column. Intervals without an explicit offset are in UTC; `sqlTimeZone` doesn't affect them.

You can't use this parameter with `INSERT`, `REPLACE`, or [catalog DDL](../development/extensions-core/catalog.md#sql-ddl) statements.|none| +|`sqlTimeCondition`|Time condition to apply to all tables that the query reads, as a comma-separated string of ISO8601 intervals, such as `"2000-01-01/2000-02-01,2001-01-01/2001-02-01"`. An empty string is an error.

Druid ANDs this condition with any other condition you provide on the `__time` column. Intervals without an explicit offset are in UTC; `sqlTimeZone` doesn't affect them.

You can't use this parameter with `INSERT`, `REPLACE`, or [catalog DDL](../development/extensions-core/catalog.md#sql-ddl) statements.|none| |`enableJoinLeftTableScanDirect`|This parameter applies to queries with joins. By default, when the left child is a simple scan with a filter, Druid runs the scan as a query, then joins it with the right child on the Broker. Setting this parameter to `true` overrides that behavior and pushes the join to the data servers instead. Even if a query doesn't explicitly include a join, this parameter may still apply since the SQL planner can translate the query into a join internally.|`false`| |`maxNumericInFilters`|Max limit for the amount of numeric values that Druid can compare for a string type dimension when the entire SQL WHERE clause of a query translates only to an [OR](../querying/filters.md#or) of [bound filter](../querying/filters.md#bound-filter). By default, Druid doesn't restrict the amount of numeric bound filters on string columns, although this situation may block other queries from running. Set this parameter to a smaller value to prevent Druid from running queries that have prohibitively long segment processing times. The optimal limit requires some trial and error. We recommend starting with 100. Users who submit a query that exceeds the limit of `maxNumericInFilters` should rewrite their queries to use strings in the `WHERE` clause instead of numbers. For example, `WHERE someString IN (‘123’, ‘456’)`. This value can't exceed the set system configuration `druid.sql.planner.maxNumericInFilters`. If `druid.sql.planner.maxNumericInFilters` isn't set explicitly, Druid ignores this value.|`druid.sql.planner.maxNumericInFilters`| |`inFunctionThreshold`|At or beyond this threshold number of values, Druid converts SQL `IN` to [`SCALAR_IN_ARRAY`](sql-functions.md#scalar_in_array). A threshold of 0 forces this conversion in all cases. A threshold of `Integer.MAX_VALUE` disables this conversion. The converted function is eligible for fewer planning-time optimizations, which speeds up planning, but may prevent certain planning-time optimizations.|`100`| diff --git a/processing/src/main/java/org/apache/druid/query/context/QueryContextParameters.java b/processing/src/main/java/org/apache/druid/query/context/QueryContextParameters.java index 288f4112306f..38e740660831 100644 --- a/processing/src/main/java/org/apache/druid/query/context/QueryContextParameters.java +++ b/processing/src/main/java/org/apache/druid/query/context/QueryContextParameters.java @@ -871,15 +871,13 @@ When set to false, no additional logs will be produced (logs produced will be en .description("Whether to consider the [pull-up rewrite](lookups.md#pull-up) of the `LOOKUP` function during SQL planning.") .build(); - public static final QueryContextParameter SQL_TIME_CONDITION = - // An object parameter, since the value may be a string or an array. The SQL planner parses it. - objectParameter("sqlTimeCondition") + public static final QueryContextParameter SQL_TIME_CONDITION = + // The SQL planner parses the intervals. + stringParameter("sqlTimeCondition") .since("40.0.0").query(Query.SQL).engine(Engine.NATIVE, Engine.MSQ, Engine.DART) .description( """ - Time condition to apply to all tables that the query reads. - Provide either a comma-separated string of ISO8601 intervals, such as `"2000-01-01/2000-02-01,2001-01-01/2001-02-01"`, or an array of ISO8601 interval strings. - An empty array matches nothing. + Time condition to apply to all tables that the query reads, as a comma-separated string of ISO8601 intervals, such as `"2000-01-01/2000-02-01,2001-01-01/2001-02-01"`. An empty string is an error.\

Druid ANDs this condition with any other condition you provide on the `__time` column. Intervals without an explicit offset are in UTC; `sqlTimeZone` doesn't affect them.\ diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/PlannerContext.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/PlannerContext.java index 1e62b465b830..95095f422e8f 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/PlannerContext.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/PlannerContext.java @@ -77,7 +77,6 @@ import javax.annotation.Nullable; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; import java.util.HashSet; import java.util.LinkedHashMap; @@ -847,52 +846,32 @@ private void initializeContextFieldsAndPlannerConfig() } /** - * Parses the value of {@link QueryContextParameters#SQL_TIME_CONDITION}, which may be a comma-separated string of - * ISO8601 intervals or an array of ISO8601 interval strings. Returns null if the value is null. Throws an error if - * the string is empty or if there are empty items after splitting on commas. The returned intervals are condensed - * and sorted. + * Parses the value of {@link QueryContextParameters#SQL_TIME_CONDITION}, a comma-separated string of ISO8601 + * intervals. Returns null if the value is null. Throws an error if the string is empty or if there are empty items + * after splitting on commas. The returned intervals are condensed and sorted. */ @Nullable - static List parseTimeCondition(@Nullable final Object value) + static List parseTimeCondition(@Nullable final String value) { if (value == null) { return null; } - final List intervalStrings; - if (value instanceof String) { - intervalStrings = Arrays.asList(((String) value).split(",", -1)); - } else if (value instanceof List) { - intervalStrings = (List) value; - } else { - throw invalidTimeConditionException(value, null); - } - final List intervals = new ArrayList<>(); - for (final Object intervalString : intervalStrings) { - if (!(intervalString instanceof String)) { - throw invalidTimeConditionException(value, null); - } - + for (final String intervalString : value.split(",", -1)) { try { - intervals.add(Intervals.of(((String) intervalString).trim())); + intervals.add(Intervals.of(intervalString.trim())); } catch (DruidException e) { - throw invalidTimeConditionException(value, e); + throw InvalidInput.exception( + e, + "Query context parameter [%s] must be a comma-separated string of ISO8601 intervals, but got [%s]", + QueryContextParameters.SQL_TIME_CONDITION, + value + ); } } return JodaUtils.condenseIntervals(intervals); } - - private static DruidException invalidTimeConditionException(final Object value, @Nullable final Throwable cause) - { - return InvalidInput.exception( - cause, - "Query context parameter [%s] must be a comma-separated string of ISO8601 intervals, or an array of " - + "ISO8601 interval strings, but got [%s]", - QueryContextParameters.SQL_TIME_CONDITION.getName(), - value - ); - } } diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidQuery.java b/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidQuery.java index 24fb40ec9689..c0732c41d83d 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidQuery.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidQuery.java @@ -818,9 +818,8 @@ VirtualColumns computeVirtualColumns(final boolean includeDimensions) * a join datasource, the datasource may be altered and left filter of join datasource may * be rid of time filters. * - * @param timeCondition intervals to AND with the filtration intervals if the datasource is table-based, from - * {@link PlannerContext#getTimeCondition()}. Must be sorted and non-overlapping. Null means - * no additional condition. + * @param timeCondition intervals to AND with the filtration intervals if the datasource is table-based. Must be + * sorted and non-overlapping. Null means no additional condition. */ @VisibleForTesting static Pair getFiltration( @@ -865,7 +864,7 @@ static Pair getFiltration( * and the arms of a union are combined. Relies on every layer passing {@code __time} through unchanged, as the * datasources we plan do. * - * @param timeCondition intervals from {@link PlannerContext#getTimeCondition()}, or null for no condition + * @param timeCondition intervals to apply to each table, or null for no condition */ private static List getBaseIntervals( final DataSource dataSource, diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteInsertDmlTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteInsertDmlTest.java index 1e9fed70cba1..2cfdecc1da8b 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteInsertDmlTest.java +++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteInsertDmlTest.java @@ -61,7 +61,6 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; -import java.util.List; import java.util.Map; import java.util.stream.Collectors; @@ -1675,21 +1674,6 @@ public void testInsertWithSqlTimeCondition() .verify(); } - @Test - public void testInsertWithEmptySqlTimeCondition() - { - HashMap context = new HashMap<>(DEFAULT_CONTEXT); - context.put(QueryContextParameters.SQL_TIME_CONDITION.getName(), List.of()); - - testIngestionQuery() - .context(context) - .sql("INSERT INTO dst SELECT * FROM foo PARTITIONED BY ALL TIME") - .expectValidationError( - invalidSqlIs("Context parameter [sqlTimeCondition] cannot be provided on operator [INSERT]") - ) - .verify(); - } - @Test public void testErrorWithUnableToConstructColumnSignatureWithExtern() { diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java index eeb0a5378b4f..a7ebe08604a1 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java +++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java @@ -25,11 +25,13 @@ import org.apache.druid.error.DruidExceptionMatcher; import org.apache.druid.java.util.common.Intervals; import org.apache.druid.java.util.common.granularity.Granularities; +import org.apache.druid.query.BadQueryContextException; import org.apache.druid.query.Druids; import org.apache.druid.query.GlobalTableDataSource; import org.apache.druid.query.LookupDataSource; import org.apache.druid.query.QueryContexts; import org.apache.druid.query.QueryDataSource; +import org.apache.druid.query.RestrictedDataSource; import org.apache.druid.query.TableDataSource; import org.apache.druid.query.UnionDataSource; import org.apache.druid.query.UnnestDataSource; @@ -44,6 +46,7 @@ import org.apache.druid.sql.calcite.NotYetSupported.Modes; import org.apache.druid.sql.calcite.filtration.Filtration; import org.apache.druid.sql.calcite.util.CalciteTests; +import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import java.util.List; @@ -139,31 +142,15 @@ public void testTimeseriesWithTimeFilter() } @Test - public void testGroupByWithArray() + public void testArrayIsInvalid() { - final Map context = makeContext(List.of("2001-01-01/2001-01-03", "2000-01-01/2000-01-02")); - - testQuery( - "SELECT dim2, COUNT(*) FROM foo GROUP BY dim2", - context, - ImmutableList.of( - GroupByQuery.builder() - .setDataSource(CalciteTests.DATASOURCE1) - .setInterval( - querySegmentSpec( - Intervals.of("2000-01-01/2000-01-02"), - Intervals.of("2001-01-01/2001-01-03") - ) - ) - .setGranularity(Granularities.ALL) - .setDimensions(dimensions(new DefaultDimensionSpec("dim2", "d0"))) - .setAggregatorSpecs(aggregators(new CountAggregatorFactory("a0"))) - .setContext(context) - .build() - ), - ImmutableList.of( - new Object[]{"a", 2L}, - new Object[]{"abc", 1L} + testQueryThrows( + "SELECT COUNT(*) FROM foo", + makeContext(List.of("2000/2001")), + BadQueryContextException.class, + e -> Assertions.assertEquals( + "Query context parameter [sqlTimeCondition] should be a string, but got [[2000/2001]]", + e.getMessage() ) ); } @@ -545,6 +532,67 @@ public void testLookupIsUnaffected() ); } + @NotYetSupported(Modes.DD_RESTRICTED_DATASOURCE_SUPPORT) + @Test + public void testRestrictedTable() + { + final Map context = makeContext("2001-01-01/2001-01-03"); + + testBuilder() + .sql("SELECT COUNT(*) FROM druid.restrictedDatasource_m1_is_6") + .queryContext(context) + .authResult(CalciteTests.SUPER_USER_AUTH_RESULT) + .expectedQueries( + ImmutableList.of( + Druids.newTimeseriesQueryBuilder() + .dataSource( + RestrictedDataSource.create( + TableDataSource.create(CalciteTests.RESTRICTED_DATASOURCE), + CalciteTests.POLICY_NO_RESTRICTION_SUPERUSER + ) + ) + .intervals(querySegmentSpec(Intervals.of("2001-01-01/2001-01-03"))) + .granularity(Granularities.ALL) + .aggregators(aggregators(new CountAggregatorFactory("a0"))) + .context(context) + .build() + ) + ) + .expectedResults(ImmutableList.of(new Object[]{2L})) + .run(); + } + + @NotYetSupported(Modes.DD_RESTRICTED_DATASOURCE_SUPPORT) + @Test + public void testRestrictedTable2() + { + final Map context = makeContext("2001-01-01/2001-01-03"); + + // The policy only admits the row with m1 = 6, at 2001-01-03, which is outside the time condition. + testBuilder() + .sql("SELECT COUNT(*) FROM druid.restrictedDatasource_m1_is_6") + .queryContext(context) + .authResult(CalciteTests.REGULAR_USER_AUTH_RESULT) + .expectedQueries( + ImmutableList.of( + Druids.newTimeseriesQueryBuilder() + .dataSource( + RestrictedDataSource.create( + TableDataSource.create(CalciteTests.RESTRICTED_DATASOURCE), + CalciteTests.POLICY_RESTRICTION + ) + ) + .intervals(querySegmentSpec(Intervals.of("2001-01-01/2001-01-03"))) + .granularity(Granularities.ALL) + .aggregators(aggregators(new CountAggregatorFactory("a0"))) + .context(context) + .build() + ) + ) + .expectedResults(ImmutableList.of(new Object[]{0L})) + .run(); + } + @Test public void testInvalidTimeCondition() { @@ -554,7 +602,7 @@ public void testInvalidTimeCondition() DruidException.class, DruidExceptionMatcher.invalidInput().expectMessageIs( "Query context parameter [sqlTimeCondition] must be a comma-separated string of ISO8601 intervals, " - + "or an array of ISO8601 interval strings, but got [2000/2001,foo]" + + "but got [2000/2001,foo]" ) ); } diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/planner/PlannerContextTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/planner/PlannerContextTest.java index a98fc108e2c6..a9bf8f40dcb1 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/planner/PlannerContextTest.java +++ b/sql/src/test/java/org/apache/druid/sql/calcite/planner/PlannerContextTest.java @@ -25,8 +25,6 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; -import java.util.Arrays; -import java.util.Collections; import java.util.List; /** @@ -61,15 +59,6 @@ public void test_parseTimeCondition_commaSeparated() ); } - @Test - public void test_parseTimeCondition_array() - { - Assertions.assertEquals( - List.of(Intervals.of("2000-01-01/2000-01-02"), Intervals.of("2001-01-01/2001-01-02")), - PlannerContext.parseTimeCondition(List.of("2001-01-01/2001-01-02", "2000-01-01/2000-01-02")) - ); - } - @Test public void test_parseTimeCondition_timeZoneOffset() { @@ -79,15 +68,6 @@ public void test_parseTimeCondition_timeZoneOffset() ); } - @Test - public void test_parseTimeCondition_emptyArray() - { - Assertions.assertEquals( - Collections.emptyList(), - PlannerContext.parseTimeCondition(Collections.emptyList()) - ); - } - @Test public void test_parseTimeCondition_emptyString() { @@ -112,25 +92,13 @@ public void test_parseTimeCondition_invalidInterval() assertInvalidTimeCondition("2000-01-01/2000-01-02,foo", "2000-01-01/2000-01-02,foo"); } - @Test - public void test_parseTimeCondition_nonStringArrayElement() - { - assertInvalidTimeCondition(Arrays.asList("2000-01-01/2000-01-02", 3), "[2000-01-01/2000-01-02, 3]"); - } - - @Test - public void test_parseTimeCondition_wrongType() - { - assertInvalidTimeCondition(3, "3"); - } - - private static void assertInvalidTimeCondition(final Object value, final String expectedValueString) + private static void assertInvalidTimeCondition(final String value, final String expectedValueString) { DruidExceptionMatcher.assertThat( Assertions.assertThrows(DruidException.class, () -> PlannerContext.parseTimeCondition(value)), DruidExceptionMatcher.invalidInput().expectMessageIs( "Query context parameter [sqlTimeCondition] must be a comma-separated string of ISO8601 intervals, " - + "or an array of ISO8601 interval strings, but got [" + expectedValueString + "]" + + "but got [" + expectedValueString + "]" ) ); } From 2d59a548ddcf8ff0d5ecbb52ccc07fb8685c7bf4 Mon Sep 17 00:00:00 2001 From: Gian Merlino Date: Fri, 23 Oct 2026 23:32:49 -0700 Subject: [PATCH 3/6] Improve test names. --- .../druid/sql/calcite/CalciteTimeConditionQueryTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java index a7ebe08604a1..6dd416c21343 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java +++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java @@ -534,7 +534,7 @@ public void testLookupIsUnaffected() @NotYetSupported(Modes.DD_RESTRICTED_DATASOURCE_SUPPORT) @Test - public void testRestrictedTable() + public void testRestrictedTable_superUser() { final Map context = makeContext("2001-01-01/2001-01-03"); @@ -564,7 +564,7 @@ public void testRestrictedTable() @NotYetSupported(Modes.DD_RESTRICTED_DATASOURCE_SUPPORT) @Test - public void testRestrictedTable2() + public void testRestrictedTable_regularUser() { final Map context = makeContext("2001-01-01/2001-01-03"); From 646c6ff7a5b7666c7beb0fae96355d2effc78854 Mon Sep 17 00:00:00 2001 From: Gian Merlino Date: Sat, 24 Oct 2026 00:16:20 -0700 Subject: [PATCH 4/6] Exempt global tables on the right side of a JOIN. --- docs/querying/sql-query-context.md | 2 +- .../druid/msq/logical/stages/ReadStage.java | 47 ++++++--- .../query/context/QueryContextParameters.java | 3 +- .../planner/querygen/DruidQueryGenerator.java | 12 ++- .../sql/calcite/rel/DruidJoinQueryRel.java | 13 +-- .../sql/calcite/CalciteJoinQueryTest.java | 45 +++++++++ .../CalciteTimeConditionQueryTest.java | 15 +-- ...TimeConditionBroadcastRightSidePositive.iq | 98 +++++++++++++++++++ 8 files changed, 192 insertions(+), 43 deletions(-) create mode 100644 sql/src/test/quidem/org.apache.druid.sql.calcite.DecoupledPlanningCalciteJoinQueryTest/testRequireTimeConditionBroadcastRightSidePositive.iq diff --git a/docs/querying/sql-query-context.md b/docs/querying/sql-query-context.md index 44660b2e5e9d..82d65886af2b 100644 --- a/docs/querying/sql-query-context.md +++ b/docs/querying/sql-query-context.md @@ -52,7 +52,7 @@ The table below lists the query context parameters you can use with Druid SQL. |`sqlReverseLookup`|Whether to consider the [reverse-lookup rewrite](lookups.md#reverse-lookup) of the `LOOKUP` function during SQL planning.

Druid reverses calls to `LOOKUP` only when the number of matching keys is lower than both `inSubQueryThreshold` and `sqlReverseLookupThreshold`.|`true`| |`sqlReverseLookupThreshold`|Maximum size of `IN` filter to create when applying a [reverse-lookup rewrite](lookups.md#reverse-lookup). If a `LOOKUP` call matches more keys than the specified threshold, it remains unchanged.

If `inSubQueryThreshold` is lower than `sqlReverseLookupThreshold`, Druid uses `inSubQueryThreshold` threshold instead.|`10000`| |`sqlPullUpLookup`|Whether to consider the [pull-up rewrite](lookups.md#pull-up) of the `LOOKUP` function during SQL planning.|`true`| -|`sqlTimeCondition`|Time condition to apply to all tables that the query reads, as a comma-separated string of ISO8601 intervals, such as `"2000-01-01/2000-02-01,2001-01-01/2001-02-01"`. An empty string is an error.

Druid ANDs this condition with any other condition you provide on the `__time` column. Intervals without an explicit offset are in UTC; `sqlTimeZone` doesn't affect them.

You can't use this parameter with `INSERT`, `REPLACE`, or [catalog DDL](../development/extensions-core/catalog.md#sql-ddl) statements.|none| +|`sqlTimeCondition`|Time condition to apply to all tables that the query reads, except global tables on the right side of a JOIN. Comma-separated string of ISO8601 intervals, such as `"2000-01-01/2000-02-01,2001-01-01/2001-02-01"`. An empty string is an error.

Druid ANDs this condition with any other condition you provide on the `__time` column. Intervals without an explicit offset are in UTC; `sqlTimeZone` doesn't affect them.

You can't use this parameter with `INSERT`, `REPLACE`, or [catalog DDL](../development/extensions-core/catalog.md#sql-ddl) statements.|none| |`enableJoinLeftTableScanDirect`|This parameter applies to queries with joins. By default, when the left child is a simple scan with a filter, Druid runs the scan as a query, then joins it with the right child on the Broker. Setting this parameter to `true` overrides that behavior and pushes the join to the data servers instead. Even if a query doesn't explicitly include a join, this parameter may still apply since the SQL planner can translate the query into a join internally.|`false`| |`maxNumericInFilters`|Max limit for the amount of numeric values that Druid can compare for a string type dimension when the entire SQL WHERE clause of a query translates only to an [OR](../querying/filters.md#or) of [bound filter](../querying/filters.md#bound-filter). By default, Druid doesn't restrict the amount of numeric bound filters on string columns, although this situation may block other queries from running. Set this parameter to a smaller value to prevent Druid from running queries that have prohibitively long segment processing times. The optimal limit requires some trial and error. We recommend starting with 100. Users who submit a query that exceeds the limit of `maxNumericInFilters` should rewrite their queries to use strings in the `WHERE` clause instead of numbers. For example, `WHERE someString IN (‘123’, ‘456’)`. This value can't exceed the set system configuration `druid.sql.planner.maxNumericInFilters`. If `druid.sql.planner.maxNumericInFilters` isn't set explicitly, Druid ignores this value.|`druid.sql.planner.maxNumericInFilters`| |`inFunctionThreshold`|At or beyond this threshold number of values, Druid converts SQL `IN` to [`SCALAR_IN_ARRAY`](sql-functions.md#scalar_in_array). A threshold of 0 forces this conversion in all cases. A threshold of `Integer.MAX_VALUE` disables this conversion. The converted function is eligible for fewer planning-time optimizations, which speeds up planning, but may prevent certain planning-time optimizations.|`100`| diff --git a/multi-stage-query/src/main/java/org/apache/druid/msq/logical/stages/ReadStage.java b/multi-stage-query/src/main/java/org/apache/druid/msq/logical/stages/ReadStage.java index 7584040415d9..0dd1428678a5 100644 --- a/multi-stage-query/src/main/java/org/apache/druid/msq/logical/stages/ReadStage.java +++ b/multi-stage-query/src/main/java/org/apache/druid/msq/logical/stages/ReadStage.java @@ -42,6 +42,7 @@ import org.apache.druid.sql.calcite.rel.VirtualColumnRegistry; import org.apache.druid.sql.calcite.rel.logical.DruidAggregate; import org.apache.druid.sql.calcite.rel.logical.DruidFilter; +import org.apache.druid.sql.calcite.rel.logical.DruidJoin; import org.apache.druid.sql.calcite.rel.logical.DruidLogicalNode; import org.apache.druid.sql.calcite.rel.logical.DruidProject; import org.apache.druid.sql.calcite.rel.logical.DruidTableScan; @@ -122,37 +123,59 @@ public static Optional buildReadStage(DruidNodeStack stack) { DruidLogicalNode node = stack.getNode(); if (node instanceof DruidValues || node instanceof DruidTableScan) { - return translateViaSourceDesc(stack.getPlannerContext(), (SourceDescProducer) node); + return translateViaSourceDesc(stack, (SourceDescProducer) node); } return Optional.empty(); } - private static Optional translateViaSourceDesc(PlannerContext plannerContext, SourceDescProducer node) + private static Optional translateViaSourceDesc(DruidNodeStack stack, SourceDescProducer node) { + PlannerContext plannerContext = stack.getPlannerContext(); SourceDesc sd = node.getSourceDesc(plannerContext, Collections.emptyList()); - InputSpec inputSpec = translateDataSource(plannerContext, sd.dataSource); + InputSpec inputSpec = translateDataSource(plannerContext, sd.dataSource, isDirectRightOfBroadcastJoin(stack)); ReadStage stage = new ReadStage(sd.rowSignature, LogicalInputSpec.of(inputSpec, sd.rowSignature)); return Optional.of(stage); } @Nonnull - private static InputSpec translateDataSource(PlannerContext plannerContext, DataSource dataSource) + private static InputSpec translateDataSource( + PlannerContext plannerContext, + DataSource dataSource, + boolean directRightOfBroadcastJoin + ) { - if (dataSource instanceof TableDataSource) { - TableDataSource ids = (TableDataSource) dataSource; + if (dataSource instanceof TableDataSource ids) { + // Don't apply the time condition to global tables that are joined directly on the right-hand side. + final boolean applyTimeCondition = !(ids.isGlobal() && directRightOfBroadcastJoin); + // TableInputSpec treats null intervals as eternity. - TableInputSpec inputSpec = new TableInputSpec(ids.getName(), plannerContext.getTimeCondition(), null); - return inputSpec; + return new TableInputSpec(ids.getName(), applyTimeCondition ? plannerContext.getTimeCondition() : null, null); } - if (dataSource instanceof InlineDataSource) { - InlineDataSource ids = (InlineDataSource) dataSource; + if (dataSource instanceof InlineDataSource ids) { InlineInputSpec inputSpec = new InlineInputSpec(ids); return inputSpec; } - if (dataSource instanceof LookupDataSource) { - LookupDataSource lookupDataSource = (LookupDataSource) dataSource; + if (dataSource instanceof LookupDataSource lookupDataSource) { return new LookupInputSpec(lookupDataSource.getLookupName()); } throw DruidException.defensive("This type of data source [%s] is not currently supported.", dataSource.getClass()); } + + /** + * Whether the node at the top of the stack is a right-hand input of a broadcast join, possibly through projections + * that only remap columns. + */ + private static boolean isDirectRightOfBroadcastJoin(DruidNodeStack stack) + { + for (int i = 0; i < stack.size() - 1; i++) { + final DruidLogicalNode parent = stack.getNode(i + 1).node; + if (parent instanceof DruidJoin join) { + return stack.getNode(i).operandIndex > 0 + && !join.getJoinAlgorithm(stack.getPlannerContext()).requiresSubquery(); + } else if (!(parent instanceof DruidProject project) || !project.isMapping()) { + return false; + } + } + return false; + } } diff --git a/processing/src/main/java/org/apache/druid/query/context/QueryContextParameters.java b/processing/src/main/java/org/apache/druid/query/context/QueryContextParameters.java index 38e740660831..4913f765e73c 100644 --- a/processing/src/main/java/org/apache/druid/query/context/QueryContextParameters.java +++ b/processing/src/main/java/org/apache/druid/query/context/QueryContextParameters.java @@ -877,7 +877,8 @@ When set to false, no additional logs will be produced (logs produced will be en .since("40.0.0").query(Query.SQL).engine(Engine.NATIVE, Engine.MSQ, Engine.DART) .description( """ - Time condition to apply to all tables that the query reads, as a comma-separated string of ISO8601 intervals, such as `"2000-01-01/2000-02-01,2001-01-01/2001-02-01"`. + Time condition to apply to all tables that the query reads, except global tables on the right side of a JOIN. + Comma-separated string of ISO8601 intervals, such as `"2000-01-01/2000-02-01,2001-01-01/2001-02-01"`. An empty string is an error.\

Druid ANDs this condition with any other condition you provide on the `__time` column. Intervals without an explicit offset are in UTC; `sqlTimeZone` doesn't affect them.\ diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/querygen/DruidQueryGenerator.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/querygen/DruidQueryGenerator.java index 98058256db3d..f114f48e7aa1 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/querygen/DruidQueryGenerator.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/querygen/DruidQueryGenerator.java @@ -37,7 +37,6 @@ import org.apache.druid.sql.calcite.planner.PlannerContext; import org.apache.druid.sql.calcite.planner.querygen.DruidQueryGenerator.PDQVertexFactory.PDQVertex; import org.apache.druid.sql.calcite.planner.querygen.SourceDescProducer.SourceDesc; -import org.apache.druid.sql.calcite.rel.DruidJoinQueryRel; import org.apache.druid.sql.calcite.rel.DruidQuery; import org.apache.druid.sql.calcite.rel.DruidRels; import org.apache.druid.sql.calcite.rel.PartialDruidQuery; @@ -76,7 +75,10 @@ public DruidQueryGenerator(PlannerContext plannerContext, DruidLogicalNode relRo */ public static class DruidNodeStack { - static class Entry + /** + * Node on the stack, along with its operand index within its parent. + */ + public static class Entry { public final DruidLogicalNode node; public final int operandIndex; @@ -244,9 +246,9 @@ static VertexTweaks analyze(DruidNodeStack stack) return new VertexTweaks(JoinPosition.analyze(stack)); } - boolean forceSubQuery(PlannerContext plannerContext, SourceDesc sourceDesc) + boolean forceSubQuery(SourceDesc sourceDesc) { - if (DruidJoinQueryRel.canUseDirectlyOnRight(plannerContext, sourceDesc.dataSource)) { + if (sourceDesc.dataSource.isGlobal()) { return false; } return joinType == JoinPosition.RIGHT; @@ -514,7 +516,7 @@ public SourceDesc unwrapUnionArm() @Override public boolean canUnwrapSourceDesc() { - if (tweaks.forceSubQuery(plannerContext, getSource())) { + if (tweaks.forceSubQuery(getSource())) { return false; } if (partialDruidQuery.stage() == Stage.SCAN) { diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidJoinQueryRel.java b/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidJoinQueryRel.java index 81723e032108..553919fceb27 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidJoinQueryRel.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidJoinQueryRel.java @@ -430,18 +430,7 @@ public static boolean computeRightRequiresSubquery(final PlannerContext plannerC // ideally this would involve JoinableFactory.isDirectlyJoinable to check that the global datasources // are in fact possibly joinable, but for now isGlobal is coupled to joinability return !(DruidRels.isScanOrMapping(right, false) - && DruidRels.druidTableIfLeafRel(right) - .filter(table -> canUseDirectlyOnRight(plannerContext, table.getDataSource())) - .isPresent()); - } - - /** - * Whether a datasource can be used directly as the right-hand side of a join, rather than through a subquery. - */ - public static boolean canUseDirectlyOnRight(final PlannerContext plannerContext, final DataSource dataSource) - { - return dataSource.isGlobal() - && (plannerContext.getTimeCondition() == null || dataSource.getTableNames().isEmpty()); + && DruidRels.druidTableIfLeafRel(right).filter(table -> table.getDataSource().isGlobal()).isPresent()); } public static Set findExistingJoinPrefixes(DataSource... dataSources) diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteJoinQueryTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteJoinQueryTest.java index 6fdacb92d2e4..b878b396b006 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteJoinQueryTest.java +++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteJoinQueryTest.java @@ -5409,6 +5409,51 @@ public void testTopNOnStringWithNonSortedOrUniqueDictionaryOrderByDim(Map= TIMESTAMP '2000-01-02 00:00:00' AND numfoo.__time < TIMESTAMP '2001-01-02 00:00:00'\n" + + "GROUP BY 1", + CalciteTests.REGULAR_USER_AUTH_RESULT, + ImmutableList.of( + GroupByQuery.builder() + .setDataSource( + join( + new TableDataSource(CalciteTests.DATASOURCE3), + new GlobalTableDataSource(CalciteTests.BROADCAST_DATASOURCE), + "j0.", + equalsCondition( + makeColumnExpression("dim4"), + makeColumnExpression("j0.dim4") + ), + JoinType.INNER + ) + ) + .setInterval(querySegmentSpec(Intervals.of("2000-01-02/2001-01-02"))) + .setGranularity(Granularities.ALL) + .setDimensions(dimensions(new DefaultDimensionSpec("j0.dim4", "d0"))) + .setAggregatorSpecs(aggregators(new CountAggregatorFactory("a0"))) + .setContext(QUERY_CONTEXT_DEFAULT) + .build() + ), + ImmutableList.of( + new Object[]{"a", 6L}, + new Object[]{"b", 3L} + ) + ); + } + @DecoupledTestConfig(quidemReason = QuidemTestCaseReason.SLIGHTLY_WORSE_FILTER_PUSHED_TO_JOIN_OPERAND) @MethodSource("provideQueryContexts") @ParameterizedTest(name = "{0}") diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java index 6dd416c21343..5ae7a3e5b7fe 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java +++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java @@ -331,7 +331,7 @@ public void testJoinWithBroadcastTable() final Map context = makeContext("2000-01-01/2001-01-02"); - // The broadcast table on the right-hand side is read through a subquery, so the time condition applies to it. + // The broadcast table on the right-hand side is joined directly, so the time condition does not apply to it. testQuery( "SELECT broadcast.dim4, COUNT(*)\n" + "FROM numfoo INNER JOIN broadcast ON numfoo.dim4 = broadcast.dim4\n" @@ -342,15 +342,7 @@ public void testJoinWithBroadcastTable() .setDataSource( join( new TableDataSource(CalciteTests.DATASOURCE3), - new QueryDataSource( - newScanQueryBuilder() - .dataSource(new GlobalTableDataSource(CalciteTests.BROADCAST_DATASOURCE)) - .intervals(querySegmentSpec(Intervals.of("2000-01-01/2001-01-02"))) - .columns("dim4") - .columnTypes(ColumnType.STRING) - .context(context) - .build() - ), + new GlobalTableDataSource(CalciteTests.BROADCAST_DATASOURCE), "j0.", equalsCondition(makeColumnExpression("dim4"), makeColumnExpression("j0.dim4")), JoinType.INNER @@ -365,7 +357,7 @@ public void testJoinWithBroadcastTable() ), ImmutableList.of( new Object[]{"a", 9L}, - new Object[]{"b", 1L} + new Object[]{"b", 3L} ) ); } @@ -532,7 +524,6 @@ public void testLookupIsUnaffected() ); } - @NotYetSupported(Modes.DD_RESTRICTED_DATASOURCE_SUPPORT) @Test public void testRestrictedTable_superUser() { diff --git a/sql/src/test/quidem/org.apache.druid.sql.calcite.DecoupledPlanningCalciteJoinQueryTest/testRequireTimeConditionBroadcastRightSidePositive.iq b/sql/src/test/quidem/org.apache.druid.sql.calcite.DecoupledPlanningCalciteJoinQueryTest/testRequireTimeConditionBroadcastRightSidePositive.iq new file mode 100644 index 000000000000..890825b51a25 --- /dev/null +++ b/sql/src/test/quidem/org.apache.druid.sql.calcite.DecoupledPlanningCalciteJoinQueryTest/testRequireTimeConditionBroadcastRightSidePositive.iq @@ -0,0 +1,98 @@ +# testRequireTimeConditionBroadcastRightSidePositive case-crc:d3617fb4 +# quidem testcase reason: JOIN_FILTER_LOCATIONS +!set debug true +!set defaultTimeout 300000 +!set maxScatterGatherBytes 9223372036854775807 +!set plannerStrategy DECOUPLED +!set requireTimeCondition true +!set sqlCurrentTimestamp 2000-01-01T00:00:00Z +!set sqlQueryId dummy +!set outputformat mysql +!use druidtest:/// +SELECT druid.broadcast.dim4, COUNT(*) +FROM druid.numfoo +INNER JOIN druid.broadcast ON numfoo.dim4 = broadcast.dim4 +WHERE numfoo.__time >= TIMESTAMP '2000-01-02 00:00:00' AND numfoo.__time < TIMESTAMP '2001-01-02 00:00:00' +GROUP BY 1; ++------+--------+ +| dim4 | EXPR$1 | ++------+--------+ +| a | 6 | +| b | 3 | ++------+--------+ +(2 rows) + +!ok +LogicalAggregate(group=[{1}], EXPR$1=[COUNT()]) + LogicalJoin(condition=[=($0, $1)], joinType=[inner]) + LogicalProject(dim4=[$4]) + LogicalFilter(condition=[SEARCH($0, Sarg[[2000-01-02 00:00:00:TIMESTAMP(3)..2001-01-02 00:00:00:TIMESTAMP(3))]:TIMESTAMP(3))]) + LogicalTableScan(table=[[druid, numfoo]]) + LogicalProject(dim4=[$4]) + LogicalTableScan(table=[[druid, broadcast]]) + +!logicalPlan +DruidAggregate(group=[{1}], EXPR$1=[COUNT()], druid=[logical]) + DruidJoin(condition=[=($0, $1)], joinType=[inner]) + DruidProject(dim4=[$4], druid=[logical]) + DruidFilter(condition=[SEARCH($0, Sarg[[2000-01-02 00:00:00:TIMESTAMP(3)..2001-01-02 00:00:00:TIMESTAMP(3))]:TIMESTAMP(3))]) + DruidTableScan(table=[[druid, numfoo]], druid=[logical]) + DruidProject(dim4=[$4], druid=[logical]) + DruidTableScan(table=[[druid, broadcast]], druid=[logical]) + +!druidPlan +{ + "queryType" : "groupBy", + "dataSource" : { + "type" : "join", + "left" : { + "type" : "query", + "query" : { + "queryType" : "scan", + "dataSource" : { + "type" : "table", + "name" : "numfoo" + }, + "intervals" : { + "type" : "intervals", + "intervals" : [ "2000-01-02T00:00:00.000Z/2001-01-02T00:00:00.000Z" ] + }, + "resultFormat" : "compactedList", + "columns" : [ "dim4" ], + "columnTypes" : [ "STRING" ], + "granularity" : { + "type" : "all" + }, + "legacy" : false + } + }, + "right" : { + "type" : "globalTable", + "name" : "broadcast" + }, + "rightPrefix" : "j0.", + "condition" : "(\"dim4\" == \"j0.dim4\")", + "joinType" : "INNER" + }, + "intervals" : { + "type" : "intervals", + "intervals" : [ "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z" ] + }, + "granularity" : { + "type" : "all" + }, + "dimensions" : [ { + "type" : "default", + "dimension" : "j0.dim4", + "outputName" : "d0", + "outputType" : "STRING" + } ], + "aggregations" : [ { + "type" : "count", + "name" : "a0" + } ], + "limitSpec" : { + "type" : "NoopLimitSpec" + } +} +!nativePlan From 94f451e7afbaeff23c5cd3ac5fd95179fe5857e1 Mon Sep 17 00:00:00 2001 From: Gian Merlino Date: Sat, 24 Oct 2026 00:32:04 -0700 Subject: [PATCH 5/6] Fold time condition into filter. --- .../druid/sql/calcite/rel/DruidQuery.java | 82 +++++++++++-------- .../druid/sql/calcite/rel/DruidQueryTest.java | 27 ++---- 2 files changed, 53 insertions(+), 56 deletions(-) diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidQuery.java b/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidQuery.java index c0732c41d83d..ee14fa09a01c 100644 --- a/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidQuery.java +++ b/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidQuery.java @@ -68,6 +68,7 @@ import org.apache.druid.query.context.QueryContextParameters; import org.apache.druid.query.dimension.DimensionSpec; import org.apache.druid.query.filter.DimFilter; +import org.apache.druid.query.filter.DimFilters; import org.apache.druid.query.groupby.GroupByQuery; import org.apache.druid.query.groupby.having.DimFilterHavingSpec; import org.apache.druid.query.groupby.orderby.DefaultLimitSpec; @@ -100,6 +101,8 @@ import org.apache.druid.sql.calcite.expression.DruidExpression; import org.apache.druid.sql.calcite.expression.Expressions; import org.apache.druid.sql.calcite.filtration.Filtration; +import org.apache.druid.sql.calcite.filtration.RangeRefKey; +import org.apache.druid.sql.calcite.filtration.Ranges; import org.apache.druid.sql.calcite.planner.Calcites; import org.apache.druid.sql.calcite.planner.OffsetLimit; import org.apache.druid.sql.calcite.planner.PlannerContext; @@ -817,15 +820,11 @@ VirtualColumns computeVirtualColumns(final boolean includeDimensions) * Returns a pair of DataSource and Filtration object created on the query filter. In case the data source is * a join datasource, the datasource may be altered and left filter of join datasource may * be rid of time filters. - * - * @param timeCondition intervals to AND with the filtration intervals if the datasource is table-based. Must be - * sorted and non-overlapping. Null means no additional condition. */ @VisibleForTesting static Pair getFiltration( DataSource dataSource, DimFilter filter, - @Nullable List timeCondition, VirtualColumnRegistry virtualColumnRegistry, JoinableFactoryWrapper joinableFactoryWrapper ) @@ -838,7 +837,7 @@ static Pair getFiltration( // Time filters within the datasource narrow the query intervals too, which lets the broker prune segments. final Filtration queryFiltration = - Filtration.create(filter, getBaseIntervals(dataSource, timeCondition, rowSignature)).optimize(rowSignature); + Filtration.create(filter, getBaseIntervals(dataSource, rowSignature)).optimize(rowSignature); if (dataSource instanceof JoinDataSource join && join.getLeftFilter() != null) { // Strip the time filters from the left filter, since the query intervals now carry them. @@ -859,45 +858,64 @@ static Pair getFiltration( } /** - * Returns the intervals that time filters within the given datasource, and the time condition on each table it - * reads, restrict its {@code __time} to, or eternity if there are none. Filters along a single path are intersected, - * and the arms of a union are combined. Relies on every layer passing {@code __time} through unchanged, as the - * datasources we plan do. - * - * @param timeCondition intervals to apply to each table, or null for no condition + * Returns the intervals that time filters within the given datasource restrict its {@code __time} to, or eternity + * if there are none. Filters along a single path are intersected, and the arms of a union are combined. Relies on + * every layer passing {@code __time} through unchanged, as the datasources we plan do. */ - private static List getBaseIntervals( - final DataSource dataSource, - @Nullable final List timeCondition, - final RowSignature rowSignature - ) + private static List getBaseIntervals(final DataSource dataSource, final RowSignature rowSignature) { if (dataSource instanceof FilteredDataSource filtered) { // Filtration intersects the filter's time bounds with the intervals it is created with. - return Filtration.create(filtered.getFilter(), getBaseIntervals(filtered.getBase(), timeCondition, rowSignature)) + return Filtration.create(filtered.getFilter(), getBaseIntervals(filtered.getBase(), rowSignature)) .optimize(rowSignature) .getIntervals(); } else if (dataSource instanceof JoinDataSource join) { // Right-hand columns are prefixed, so __time is the left-hand side's. - return Filtration.create(join.getLeftFilter(), getBaseIntervals(join.getLeft(), timeCondition, rowSignature)) + return Filtration.create(join.getLeftFilter(), getBaseIntervals(join.getLeft(), rowSignature)) .optimize(rowSignature) .getIntervals(); } else if (dataSource instanceof UnionDataSource) { final List intervals = new ArrayList<>(); for (final DataSource arm : dataSource.getChildren()) { - intervals.addAll(getBaseIntervals(arm, timeCondition, rowSignature)); + intervals.addAll(getBaseIntervals(arm, rowSignature)); } return JodaUtils.condenseIntervals(intervals); } else if (dataSource instanceof UnnestDataSource unnest) { - return getBaseIntervals(unnest.getBase(), timeCondition, rowSignature); + return getBaseIntervals(unnest.getBase(), rowSignature); } else if (dataSource instanceof ProjectDataSource project) { - return getBaseIntervals(project.getBase(), timeCondition, rowSignature); + return getBaseIntervals(project.getBase(), rowSignature); } else { - // A table, since the caller only uses this for datasources where canUseIntervalFiltering is true. - return timeCondition != null ? timeCondition : Intervals.ONLY_ETERNITY; + return Intervals.ONLY_ETERNITY; } } + /** + * Returns the filter ANDed with a filter on {@code __time} matching the time condition, if the datasource is + * table-based. + */ + @Nullable + @VisibleForTesting + static DimFilter withTimeCondition( + final DataSource dataSource, + @Nullable final DimFilter filter, + @Nullable final List timeCondition + ) + { + if (timeCondition == null || !canUseIntervalFiltering(dataSource)) { + return filter; + } + + final RangeRefKey timeKey = new RangeRefKey(ColumnHolder.TIME_COLUMN_NAME, ColumnType.LONG); + final List timeFilters = new ArrayList<>(); + for (final Interval interval : timeCondition) { + timeFilters.add(Ranges.interval(timeKey, interval)); + } + return DimFilters.conjunction( + filter, + timeFilters.size() == 1 ? timeFilters.getFirst() : DimFilters.or(timeFilters) + ); + } + /** * Whether the given datasource can make use of "intervals" based filtering. The is true for anything based on * regular tables ({@link TableDataSource}). @@ -1089,8 +1107,7 @@ private TimeBoundaryQuery toTimeBoundaryQuery() } final Pair dataSourceFiltrationPair = getFiltration( dataSource, - filter, - plannerContext.getTimeCondition(), + withTimeCondition(dataSource, filter, plannerContext.getTimeCondition()), virtualColumnRegistry, plannerContext.getJoinableFactoryWrapper() ); @@ -1214,8 +1231,7 @@ private TimeseriesQuery toTimeseriesQuery() final VirtualColumns virtualColumns = computeVirtualColumns(false); final Pair dataSourceFiltrationPair = getFiltration( dataSource, - filter, - plannerContext.getTimeCondition(), + withTimeCondition(dataSource, filter, plannerContext.getTimeCondition()), virtualColumnRegistry, plannerContext.getJoinableFactoryWrapper() ); @@ -1321,8 +1337,7 @@ private TopNQuery toTopNQuery() final VirtualColumns virtualColumns = computeVirtualColumns(true); final Pair dataSourceFiltrationPair = getFiltration( dataSource, - filter, - plannerContext.getTimeCondition(), + withTimeCondition(dataSource, filter, plannerContext.getTimeCondition()), virtualColumnRegistry, plannerContext.getJoinableFactoryWrapper() ); @@ -1369,8 +1384,7 @@ private GroupByQuery toGroupByQuery() final VirtualColumns virtualColumns = computeVirtualColumns(true); final Pair dataSourceFiltrationPair = getFiltration( dataSource, - filter, - plannerContext.getTimeCondition(), + withTimeCondition(dataSource, filter, plannerContext.getTimeCondition()), virtualColumnRegistry, plannerContext.getJoinableFactoryWrapper() ); @@ -1536,8 +1550,7 @@ private WindowOperatorQuery toWindowQuery() final Pair dataSourceFiltrationPair = getFiltration( dataSource, - filter, - plannerContext.getTimeCondition(), + withTimeCondition(dataSource, filter, plannerContext.getTimeCondition()), virtualColumnRegistry, plannerContext.getJoinableFactoryWrapper() ); @@ -1661,8 +1674,7 @@ private ScanQuery toScanQuery(final boolean considerSorting) final VirtualColumns virtualColumns = computeVirtualColumns(true); final Pair dataSourceFiltrationPair = getFiltration( dataSource, - filter, - plannerContext.getTimeCondition(), + withTimeCondition(dataSource, filter, plannerContext.getTimeCondition()), virtualColumnRegistry, plannerContext.getJoinableFactoryWrapper() ); diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/rel/DruidQueryTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/rel/DruidQueryTest.java index 2ed50f561238..b5be90043d98 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/rel/DruidQueryTest.java +++ b/sql/src/test/java/org/apache/druid/sql/calcite/rel/DruidQueryTest.java @@ -67,7 +67,6 @@ public void test_filtration_noJoinAndInterval() Pair pair = DruidQuery.getFiltration( dataSource, selectorFilter, - null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -86,7 +85,6 @@ public void test_filtration_intervalInQueryFilter() Pair pair = DruidQuery.getFiltration( dataSource, filterWithInterval, - null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -105,7 +103,6 @@ public void test_filtration_joinDataSource_intervalInQueryFilter() Pair pair = DruidQuery.getFiltration( dataSource, filterWithInterval, - null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -125,7 +122,6 @@ public void test_filtration_joinDataSource_intervalInBaseTableFilter_inner() Pair pair = DruidQuery.getFiltration( dataSource, otherFilter, - null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -145,7 +141,6 @@ public void test_filtration_joinDataSource_intervalInBaseTableFilter_left() Pair pair = DruidQuery.getFiltration( dataSource, otherFilter, - null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -165,7 +160,6 @@ public void test_filtration_joinDataSource_intervalInBaseTableFilter_right() Pair pair = DruidQuery.getFiltration( dataSource, otherFilter, - null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -185,7 +179,6 @@ public void test_filtration_joinDataSource_intervalInBaseTableFilter_full() Pair pair = DruidQuery.getFiltration( dataSource, otherFilter, - null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -210,7 +203,6 @@ public void test_filtration_intervalsInBothFilters() Pair pair = DruidQuery.getFiltration( dataSource, queryFilter, - null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -240,7 +232,6 @@ public void test_filtration_filteredUnionDataSource_intervalsInArmFilters() final Pair pair = DruidQuery.getFiltration( dataSource, otherFilter, - null, VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -260,8 +251,7 @@ public void test_filtration_timeCondition_noIntervalInQueryFilter() DataSource dataSource = new TableDataSource("test"); Pair pair = DruidQuery.getFiltration( dataSource, - selectorFilter, - Collections.singletonList(Intervals.utc(50, 150)), + DruidQuery.withTimeCondition(dataSource, selectorFilter, Collections.singletonList(Intervals.utc(50, 150))), VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -279,8 +269,7 @@ public void test_filtration_timeCondition_noQueryFilter() DataSource dataSource = new TableDataSource("test"); Pair pair = DruidQuery.getFiltration( dataSource, - null, - Collections.singletonList(Intervals.utc(50, 150)), + DruidQuery.withTimeCondition(dataSource, null, Collections.singletonList(Intervals.utc(50, 150))), VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -298,8 +287,7 @@ public void test_filtration_timeCondition_intervalInQueryFilter() DataSource dataSource = new TableDataSource("test"); Pair pair = DruidQuery.getFiltration( dataSource, - filterWithInterval, - Collections.singletonList(Intervals.utc(150, 300)), + DruidQuery.withTimeCondition(dataSource, filterWithInterval, Collections.singletonList(Intervals.utc(150, 300))), VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -318,8 +306,7 @@ public void test_filtration_timeCondition_joinDataSource_intervalInBaseTableFilt DataSource expectedDataSource = join(JoinType.INNER, selectorFilter); Pair pair = DruidQuery.getFiltration( dataSource, - otherFilter, - Collections.singletonList(Intervals.utc(50, 150)), + DruidQuery.withTimeCondition(dataSource, otherFilter, Collections.singletonList(Intervals.utc(50, 150))), VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -342,8 +329,7 @@ public void test_filtration_timeCondition_queryDataSource() ); Pair pair = DruidQuery.getFiltration( dataSource, - selectorFilter, - Collections.singletonList(Intervals.utc(150, 300)), + DruidQuery.withTimeCondition(dataSource, selectorFilter, Collections.singletonList(Intervals.utc(150, 300))), VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), @@ -378,8 +364,7 @@ public void test_filtration_timeCondition_joinDataSource_subqueryOnLeft() ); Pair pair = DruidQuery.getFiltration( dataSource, - selectorFilter, - Collections.singletonList(Intervals.utc(150, 300)), + DruidQuery.withTimeCondition(dataSource, selectorFilter, Collections.singletonList(Intervals.utc(150, 300))), VirtualColumnRegistry.create( RowSignature.empty(), new ExpressionParserImpl(TestExprMacroTable.INSTANCE), From 6d03a04a0c7f8aa062ee93a53de846fb85fab7b5 Mon Sep 17 00:00:00 2001 From: Gian Merlino Date: Sat, 24 Oct 2026 02:18:22 -0700 Subject: [PATCH 6/6] Fix test. --- .../java/org/apache/druid/sql/calcite/rel/DruidRelTest.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/rel/DruidRelTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/rel/DruidRelTest.java index 84eb767c881f..2386f341de10 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/rel/DruidRelTest.java +++ b/sql/src/test/java/org/apache/druid/sql/calcite/rel/DruidRelTest.java @@ -132,6 +132,8 @@ public void setup() throws Exception when(mockPlannerContext.getPlannerConfig()).thenReturn(PlannerConfig.builder().build()); when(mockPlannerContext.getJsonMapper()).thenReturn(JsonMapper.builder().build()); when(mockPlannerContext.getAuthorizationResult()).thenReturn(AUTHORIZATION_RESULT); + // Mockito would otherwise return an empty list, which means "match nothing" rather than "no time condition". + when(mockPlannerContext.getTimeCondition()).thenReturn(null); LogicalTableScan logicalTableScan = new LogicalTableScan( mockRelOptCluster,