feat: sqlTimeCondition context parameter.

docs/querying/sql-query-context.md

@@ -51,6 +51,7 @@ The table below lists the query context parameters you can use with Druid SQL.
 |`sqlReverseLookup`|Whether to consider the [reverse-lookup rewrite](lookups.md#reverse-lookup) of the `LOOKUP` function during SQL planning.<br /><br />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.<br /><br />If `inSubQueryThreshold` is lower than `sqlReverseLookupThreshold`, Druid uses `inSubQueryThreshold` threshold instead.|10000|
 |`sqlPullUpLookup`|Whether to consider the [pull-up rewrite](lookups.md#pull-up) of the `LOOKUP` function during SQL planning.|`true`|
+|`sqlTimeCondition`|Time condition to apply to all tables that the query reads. Provide either a comma-separated string of ISO8601 intervals, such as `"2000-01-01/2000-02-01,2001-01-01/2001-02-01"`, or an array of ISO8601 interval strings. An empty array matches nothing. An empty string is an error.<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|
 |`enableJoinLeftTableScanDirect`|This parameter applies to queries with joins. By default, when the left child is a simple scan with a filter, Druid runs the scan as a query, then joins it with the right child on the Broker. Setting this parameter to `true` overrides that behavior and pushes the join to the data servers instead. Even if a query doesn't explicitly include a join, this parameter may still apply since the SQL planner can translate the query into a join internally.|`false`|
 |`maxNumericInFilters`|Max limit for the amount of numeric values that Druid can compare for a string type dimension when the entire SQL WHERE clause of a query translates only to an [OR](../querying/filters.md#or) of [bound filter](../querying/filters.md#bound-filter). By default, Druid doesn't restrict the amount of numeric bound filters on string columns, although this situation may block other queries from running. Set this parameter to a smaller value to prevent Druid from running queries that have prohibitively long segment processing times. The optimal limit requires some trial and error. We recommend starting with 100. Users who submit a query that exceeds the limit of `maxNumericInFilters` should rewrite their queries to use strings in the `WHERE` clause instead of numbers. For example, `WHERE someString IN (‘123’, ‘456’)`. This value can't exceed the set system configuration `druid.sql.planner.maxNumericInFilters`. If `druid.sql.planner.maxNumericInFilters` isn't set explicitly, Druid ignores this value.|`-1`|
 |`inFunctionThreshold`| At or beyond this threshold number of values, Druid converts SQL `IN` to [`SCALAR_IN_ARRAY`](sql-functions.md#scalar_in_array). A threshold of 0 forces this conversion in all cases. A threshold of `Integer.MAX_VALUE` disables this conversion. The converted function is eligible for fewer planning-time optimizations, which speeds up planning, but may prevent certain planning-time optimizations.| `100`|

multi-stage-query/src/main/java/org/apache/druid/msq/logical/stages/ReadStage.java

@@ -20,7 +20,6 @@
 package org.apache.druid.msq.logical.stages;
 
 import org.apache.druid.error.DruidException;
-import org.apache.druid.java.util.common.Intervals;
 import org.apache.druid.msq.exec.StageProcessor;
 import org.apache.druid.msq.input.InputSpec;
 import org.apache.druid.msq.input.inline.InlineInputSpec;
@@ -131,17 +130,18 @@ public static Optional<ReadStage> buildReadStage(DruidNodeStack stack)
   private static Optional<ReadStage> translateViaSourceDesc(PlannerContext plannerContext, SourceDescProducer node)
   {
     SourceDesc sd = node.getSourceDesc(plannerContext, Collections.emptyList());
-    InputSpec inputSpec = translateDataSource(sd.dataSource);
+    InputSpec inputSpec = translateDataSource(plannerContext, sd.dataSource);
     ReadStage stage = new ReadStage(sd.rowSignature, LogicalInputSpec.of(inputSpec, sd.rowSignature));
     return Optional.of(stage);
   }
 
   @Nonnull
-  private static InputSpec translateDataSource(DataSource dataSource)
+  private static InputSpec translateDataSource(PlannerContext plannerContext, DataSource dataSource)
   {
     if (dataSource instanceof TableDataSource) {
       TableDataSource ids = (TableDataSource) dataSource;
-      TableInputSpec inputSpec = new TableInputSpec(ids.getName(), Intervals.ONLY_ETERNITY, null);
+      // TableInputSpec treats null intervals as eternity.
+      TableInputSpec inputSpec = new TableInputSpec(ids.getName(), plannerContext.getTimeCondition(), null);
       return inputSpec;
     }
     if (dataSource instanceof InlineDataSource) {

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/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

@@ -129,6 +129,15 @@ public void validate()
     if (!handlerContext.plannerContext().getParameters().isEmpty()) {
       throw InvalidSqlInput.exception("Dynamic parameters are not supported for [%s]", operationName());
     }
+    // Fail if CTX_SQL_TIME_CONDITION is specified, since projection bodies are planned with the statement's context,
+    // so the condition would be stored in the catalog as part of their filters.
+    if (handlerContext.queryContextMap().get(PlannerContext.CTX_SQL_TIME_CONDITION) != null) {
+      throw InvalidSqlInput.exception(
+          "Context parameter [%s] cannot be provided on operator [%s]",
+          PlannerContext.CTX_SQL_TIME_CONDITION,
+          operationName()
+      );
+    }
     tableId = TableId.datasource(resolveTableName());
     final Resource resource = new Resource(tableId.name(), ResourceType.DATASOURCE);
     resourceActions = ImmutableSet.of(

sql/src/main/java/org/apache/druid/sql/calcite/planner/IngestHandler.java

@@ -157,6 +157,16 @@ public void validate()
           operationName()
       );
     }
+
+    // Fail the query if CTX_SQL_TIME_CONDITION is specified, since it limits the rows read without limiting the time
+    // chunks that are overwritten, so REPLACE would delete data outside the condition.
+    if (handlerContext.queryContextMap().get(PlannerContext.CTX_SQL_TIME_CONDITION) != null) {
+      throw InvalidSqlInput.exception(
+          "Context parameter [%s] cannot be provided on operator [%s]",
+          PlannerContext.CTX_SQL_TIME_CONDITION,
+          operationName()
+      );
+    }
     DruidSqlIngest ingestNode = ingestNode();
     DruidSqlIngest validatedNode = (DruidSqlIngest) validate(ingestNode);
     validatedQueryNode = validatedNode.getSource();

sql/src/main/java/org/apache/druid/sql/calcite/planner/PlannerContext.java

@@ -31,9 +31,13 @@
 import org.apache.calcite.schema.Table;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.util.CancelFlag;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.InvalidInput;
 import org.apache.druid.error.InvalidSqlInput;
 import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.JodaUtils;
 import org.apache.druid.java.util.common.Numbers;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.granularity.Granularities;
@@ -72,6 +76,7 @@
 import org.joda.time.Interval;
 
 import javax.annotation.Nullable;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
@@ -138,6 +143,11 @@ public class PlannerContext
   public static final String CTX_SQL_USE_GRANULARITY = "sqlUseGranularity";
   public static final boolean DEFAULT_SQL_USE_GRANULARITY = true;
 
+  /**
+   * Context key for {@link PlannerContext#getTimeCondition()}.
+   */
+  public static final String CTX_SQL_TIME_CONDITION = "sqlTimeCondition";
+
   // DataContext keys
   public static final String DATA_CTX_AUTHENTICATION_RESULT = "authenticationResult";
 
@@ -179,6 +189,8 @@ public class PlannerContext
   private boolean pullUpLookup;
   private boolean reverseLookup;
   private boolean useGranularity;
+  @Nullable
+  private List<Interval> timeCondition;
   private DateTime localNow;
 
   // bindings for dynamic parameters to bind during planning
@@ -485,6 +497,16 @@ public boolean isUseGranularity()
     return useGranularity;
   }
 
+  /**
+   * Returns intervals that should be applied (via AND) to table reads. The returned list is sorted and
+   * non-overlapping.
+   */
+  @Nullable
+  public List<Interval> getTimeCondition()
+  {
+    return timeCondition;
+  }
+
   public List<TypedValue> getParameters()
   {
     return parameters;
@@ -800,6 +822,8 @@ private void initializeContextFieldsAndPlannerConfig()
       useGranularity = DEFAULT_SQL_USE_GRANULARITY;
     }
 
+    timeCondition = parseTimeCondition(queryContext.get(CTX_SQL_TIME_CONDITION));
+
     sqlQueryId = (String) this.queryContext.get(QueryContexts.CTX_SQL_QUERY_ID);
     // special handling for DruidViewMacro, normal client will allocate sqlid in SqlLifecyle
     if (Strings.isNullOrEmpty(sqlQueryId)) {
@@ -814,6 +838,44 @@ private void initializeContextFieldsAndPlannerConfig()
     }
   }
 
+  /**
+   * Parses the value of {@link #CTX_SQL_TIME_CONDITION}, which may be a comma-separated string of ISO8601 intervals
+   * or an array of ISO8601 interval strings. Returns null if the value is null. Throws an error if the string is
+   * empty or if there are empty items after splitting on commas. The returned intervals are condensed and sorted.
+   */
+  @Nullable
+  static List<Interval> parseTimeCondition(@Nullable final Object value)
+  {
+    if (value == null) {
+      return null;
+    }
+
+    final List<?> intervalStrings;
+    if (value instanceof String) {
+      intervalStrings = Arrays.asList(((String) value).split(",", -1));
+    } else if (value instanceof List) {
+      intervalStrings = (List<?>) value;
+    } else {
+      throw invalidTimeConditionException(value, null);
+    }
+
+    final List<Interval> intervals = new ArrayList<>();
+    for (final Object intervalString : intervalStrings) {
+      if (!(intervalString instanceof String)) {
+        throw invalidTimeConditionException(value, null);
+      }
+
+      try {
+        intervals.add(Intervals.of(((String) intervalString).trim()));
+      }
+      catch (DruidException e) {
+        throw invalidTimeConditionException(value, e);
+      }
+    }
+
+    return JodaUtils.condenseIntervals(intervals);
+  }
+
   private static JoinAlgorithm getJoinAlgorithmFromContextValue(final Object object)
   {
     final String s = QueryContexts.getAsString(
@@ -833,4 +895,15 @@ private static JoinAlgorithm getJoinAlgorithmFromContextValue(final Object objec
       );
     }
   }
+
+  private static DruidException invalidTimeConditionException(final Object value, @Nullable final Throwable cause)
+  {
+    return InvalidInput.exception(
+        cause,
+        "Query context parameter [%s] must be a comma-separated string of ISO8601 intervals, or an array of "
+        + "ISO8601 interval strings, but got [%s]",
+        CTX_SQL_TIME_CONDITION,
+        value
+    );
+  }
 }