feat: sqlTimeCondition context parameter.

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.<br /><br />Druid reverses calls to `LOOKUP` only when the number of matching keys is lower than both `inSubQueryThreshold` and `sqlReverseLookupThreshold`.|`true`| <!-- GENERATED QUERY CONTEXT PARAMETER: sqlReverseLookup -->
 |`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.<br /><br />If `inSubQueryThreshold` is lower than `sqlReverseLookupThreshold`, Druid uses `inSubQueryThreshold` threshold instead.|`10000`| <!-- GENERATED QUERY CONTEXT PARAMETER: sqlReverseLookupThreshold -->
 |`sqlPullUpLookup`|Whether to consider the [pull-up rewrite](lookups.md#pull-up) of the `LOOKUP` function during SQL planning.|`true`| <!-- GENERATED QUERY CONTEXT PARAMETER: sqlPullUpLookup -->
+|`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.<br /><br />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.<br /><br />You can't use this parameter with `INSERT`, `REPLACE`, or [catalog DDL](../development/extensions-core/catalog.md#sql-ddl) statements.|none| <!-- GENERATED QUERY CONTEXT PARAMETER: sqlTimeCondition -->
 |`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`| <!-- GENERATED QUERY CONTEXT PARAMETER: enableJoinLeftTableScanDirect -->
 |`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`| <!-- GENERATED QUERY CONTEXT PARAMETER: 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`| <!-- GENERATED QUERY CONTEXT PARAMETER: inFunctionThreshold -->

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<ReadStage> 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<ReadStage> translateViaSourceDesc(PlannerContext plannerContext, SourceDescProducer node)
+  private static Optional<ReadStage> 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;
+  }
 }

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<String, Object> context = ImmutableMap.<String, Object>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();
+  }
+}

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();
+  }
+}

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<String> 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.\
+              <br /><br />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.\
+              <br /><br />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<JoinAlgorithm> SQL_JOIN_ALGORITHM =
       enumParameter("sqlJoinAlgorithm", JoinAlgorithm.class, JoinAlgorithm::fromString)
           .since("26.0.0").query(Query.SQL).engine(Engine.MSQ, Engine.DART)

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;

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()
   {

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(

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();