diff --git a/docs/querying/sql-query-context.md b/docs/querying/sql-query-context.md index 26d65ca821e1..82d65886af2b 100644 --- a/docs/querying/sql-query-context.md +++ b/docs/querying/sql-query-context.md @@ -52,6 +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, 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 a4ece30a0cd6..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 @@ -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; @@ -43,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; @@ -123,36 +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(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(DataSource dataSource) + private static InputSpec translateDataSource( + PlannerContext plannerContext, + DataSource dataSource, + boolean directRightOfBroadcastJoin + ) { - if (dataSource instanceof TableDataSource) { - TableDataSource ids = (TableDataSource) dataSource; - TableInputSpec inputSpec = new TableInputSpec(ids.getName(), Intervals.ONLY_ETERNITY, null); - return inputSpec; + 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. + 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/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/context/QueryContextParameters.java b/processing/src/main/java/org/apache/druid/query/context/QueryContextParameters.java index e5461adb9782..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 @@ -871,6 +871,24 @@ 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 = + // 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, 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. + """ + ) + .defaultDescription("none") + .statement(StatementType.SELECT) + .build(); + public static final QueryContextParameter SQL_JOIN_ALGORITHM = enumParameter("sqlJoinAlgorithm", JoinAlgorithm.class, JoinAlgorithm::fromString) .since("26.0.0").query(Query.SQL).engine(Engine.MSQ, Engine.DART) 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..9eea7fb22444 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 @@ -47,6 +47,7 @@ import org.apache.druid.java.util.common.granularity.Granularity; import org.apache.druid.java.util.common.granularity.PeriodGranularity; import org.apache.druid.java.util.common.guava.Sequences; +import org.apache.druid.query.context.QueryContextParameters; import org.apache.druid.query.explain.ExplainAttributes; import org.apache.druid.segment.column.ColumnType; import org.apache.druid.segment.projections.Projections; @@ -129,6 +130,15 @@ public void validate() if (!handlerContext.plannerContext().getParameters().isEmpty()) { throw InvalidSqlInput.exception("Dynamic parameters are not supported for [%s]", operationName()); } + // Fail if sqlTimeCondition 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.queryContext().has(QueryContextParameters.SQL_TIME_CONDITION)) { + throw InvalidSqlInput.exception( + "Context parameter [%s] cannot be provided on operator [%s]", + QueryContextParameters.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 d7e5864e5dea..c9349e40e6c2 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 @@ -158,6 +158,16 @@ public void validate() operationName() ); } + + // Fail the query if sqlTimeCondition 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.queryContext().has(QueryContextParameters.SQL_TIME_CONDITION)) { + throw InvalidSqlInput.exception( + "Context parameter [%s] cannot be provided on operator [%s]", + QueryContextParameters.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 0e8c596befd4..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 @@ -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.StringUtils; import org.apache.druid.java.util.common.granularity.Granularities; import org.apache.druid.math.expr.Expr; @@ -72,6 +76,7 @@ import org.joda.time.Interval; import javax.annotation.Nullable; +import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; import java.util.LinkedHashMap; @@ -245,6 +250,8 @@ public static JoinAlgorithm getJoinAlgorithm(Map queryContext) private boolean pullUpLookup; private boolean reverseLookup; private boolean useGranularity; + @Nullable + private List timeCondition; private DateTime localNow; // bindings for dynamic parameters to bind during planning @@ -535,6 +542,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; @@ -812,6 +829,8 @@ private void initializeContextFieldsAndPlannerConfig() useGranularity = context.getOrDefault(QueryContextParameters.SQL_USE_GRANULARITY); + timeCondition = parseTimeCondition(context.get(QueryContextParameters.SQL_TIME_CONDITION)); + sqlQueryId = context.get(QueryContextParameters.SQL_QUERY_ID); // special handling for DruidViewMacro, normal client will allocate sqlid in SqlLifecyle if (Strings.isNullOrEmpty(sqlQueryId)) { @@ -825,4 +844,34 @@ private void initializeContextFieldsAndPlannerConfig() plannerConfig = getPlannerToolbox().plannerConfig.withOverrides(queryContext); } } + + /** + * 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 String value) + { + if (value == null) { + return null; + } + + final List intervals = new ArrayList<>(); + for (final String intervalString : value.split(",", -1)) { + try { + intervals.add(Intervals.of(intervalString.trim())); + } + catch (DruidException 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); + } } 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..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 @@ -75,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; 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 928dc7971278..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; @@ -886,6 +889,33 @@ private static List getBaseIntervals(final DataSource dataSource, fina } } + /** + * 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}). @@ -1077,7 +1107,7 @@ private TimeBoundaryQuery toTimeBoundaryQuery() } final Pair dataSourceFiltrationPair = getFiltration( dataSource, - filter, + withTimeCondition(dataSource, filter, plannerContext.getTimeCondition()), virtualColumnRegistry, plannerContext.getJoinableFactoryWrapper() ); @@ -1201,7 +1231,7 @@ private TimeseriesQuery toTimeseriesQuery() final VirtualColumns virtualColumns = computeVirtualColumns(false); final Pair dataSourceFiltrationPair = getFiltration( dataSource, - filter, + withTimeCondition(dataSource, filter, plannerContext.getTimeCondition()), virtualColumnRegistry, plannerContext.getJoinableFactoryWrapper() ); @@ -1307,7 +1337,7 @@ private TopNQuery toTopNQuery() final VirtualColumns virtualColumns = computeVirtualColumns(true); final Pair dataSourceFiltrationPair = getFiltration( dataSource, - filter, + withTimeCondition(dataSource, filter, plannerContext.getTimeCondition()), virtualColumnRegistry, plannerContext.getJoinableFactoryWrapper() ); @@ -1354,7 +1384,7 @@ private GroupByQuery toGroupByQuery() final VirtualColumns virtualColumns = computeVirtualColumns(true); final Pair dataSourceFiltrationPair = getFiltration( dataSource, - filter, + withTimeCondition(dataSource, filter, plannerContext.getTimeCondition()), virtualColumnRegistry, plannerContext.getJoinableFactoryWrapper() ); @@ -1517,9 +1547,17 @@ private WindowOperatorQuery toWindowQuery() boolean pushLeafOperator = plannerContext.queryContext().getOrDefault(QueryContextParameters.ENABLE_RAC_OVER_WIRE) && !plannerContext.featureAvailable(EngineFeature.WINDOW_LEAF_OPERATOR); - return new WindowOperatorQuery( + + final Pair dataSourceFiltrationPair = getFiltration( dataSource, - new LegacySegmentSpec(Intervals.ETERNITY), + withTimeCondition(dataSource, filter, plannerContext.getTimeCondition()), + virtualColumnRegistry, + plannerContext.getJoinableFactoryWrapper() + ); + + return new WindowOperatorQuery( + dataSourceFiltrationPair.lhs, + new LegacySegmentSpec(dataSourceFiltrationPair.rhs.getIntervals()), plannerContext.queryContextMap(), windowing.getSignature(), operators, @@ -1583,7 +1621,7 @@ private WindowOperatorQuery toScanAndSortQuery() return new WindowOperatorQuery( newDataSource, - new LegacySegmentSpec(Intervals.ETERNITY), + new LegacySegmentSpec(scan.getIntervals()), plannerContext.queryContextMap(), signature, operators, @@ -1636,7 +1674,7 @@ private ScanQuery toScanQuery(final boolean considerSorting) final VirtualColumns virtualColumns = computeVirtualColumns(true); final Pair dataSourceFiltrationPair = getFiltration( dataSource, - filter, + withTimeCondition(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 31b5aa183c81..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 @@ -1659,6 +1659,21 @@ public void testInsertWithSqlOuterLimit() .verify(); } + @Test + public void testInsertWithSqlTimeCondition() + { + HashMap context = new HashMap<>(DEFAULT_CONTEXT); + context.put(QueryContextParameters.SQL_TIME_CONDITION.getName(), "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 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 9ea3573ed7b4..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 @@ -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; @@ -94,6 +98,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; @@ -5404,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}") @@ -6583,4 +6633,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 2497bc42f3cc..998ef28900c1 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(QueryContextParameters.SQL_TIME_CONDITION.getName(), "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..5ae7a3e5b7fe --- /dev/null +++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeConditionQueryTest.java @@ -0,0 +1,608 @@ +/* + * 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.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; +import org.apache.druid.query.aggregation.CountAggregatorFactory; +import org.apache.druid.query.aggregation.LongSumAggregatorFactory; +import org.apache.druid.query.context.QueryContextParameters; +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.util.CalciteTests; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Map; + +/** + * Tests for the {@link QueryContextParameters#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 testArrayIsInvalid() + { + 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() + ) + ); + } + + @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 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" + + "GROUP BY 1", + context, + 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-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", 3L} + ) + ); + } + + @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 testRestrictedTable_superUser() + { + 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 testRestrictedTable_regularUser() + { + 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() + { + 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, " + + "but got [2000/2001,foo]" + ) + ); + } + + protected static Map makeContext(final Object timeCondition) + { + return ImmutableMap.builder() + .putAll(QUERY_CONTEXT_DEFAULT) + .put(QueryContextParameters.SQL_TIME_CONDITION.getName(), 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..a9bf8f40dcb1 --- /dev/null +++ b/sql/src/test/java/org/apache/druid/sql/calcite/planner/PlannerContextTest.java @@ -0,0 +1,105 @@ +/* + * 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.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_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_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"); + } + + 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, " + + "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..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 @@ -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; @@ -242,6 +245,139 @@ 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, + DruidQuery.withTimeCondition(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, + DruidQuery.withTimeCondition(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, + DruidQuery.withTimeCondition(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, + DruidQuery.withTimeCondition(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, + DruidQuery.withTimeCondition(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, + DruidQuery.withTimeCondition(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/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, 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/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 diff --git a/website/.spelling b/website/.spelling index 14b88a809d21..c60a7a3e1d44 100644 --- a/website/.spelling +++ b/website/.spelling @@ -2122,6 +2122,7 @@ addr affinityConfig allowAll ANDed +ANDs array_mod autoscale autoscalers