Strict window frame checks

processing/src/main/java/org/apache/druid/query/QueryContext.java

@@ -27,6 +27,7 @@
 import org.apache.druid.segment.QueryableIndexStorageAdapter;
 
 import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.util.Collections;
 import java.util.Map;
@@ -582,6 +583,15 @@ public boolean isTimeBoundaryPlanningEnabled()
     );
   }
 
+  public boolean isWindowingStrictValidation()
+  {
+    return getBoolean(
+        QueryContexts.WINDOWING_STRICT_VALIDATION,
+        QueryContexts.DEFAULT_WINDOWING_STRICT_VALIDATION
+    );
+  }
+
+
   public String getBrokerServiceName()
   {
     return getString(QueryContexts.BROKER_SERVICE_NAME);

processing/src/main/java/org/apache/druid/query/QueryContexts.java

@@ -85,6 +85,8 @@ public class QueryContexts
   public static final String SERIALIZE_DATE_TIME_AS_LONG_INNER_KEY = "serializeDateTimeAsLongInner";
   public static final String UNCOVERED_INTERVALS_LIMIT_KEY = "uncoveredIntervalsLimit";
   public static final String MIN_TOP_N_THRESHOLD = "minTopNThreshold";
+  public static final String WINDOWING_STRICT_VALIDATION = "windowingStrictValidation";
+
 
   // SQL query context keys
   public static final String CTX_SQL_QUERY_ID = BaseQuery.SQL_QUERY_ID;
@@ -117,6 +119,7 @@ public class QueryContexts
   public static final boolean DEFAULT_ENABLE_DEBUG = false;
   public static final int DEFAULT_IN_SUB_QUERY_THRESHOLD = Integer.MAX_VALUE;
   public static final boolean DEFAULT_ENABLE_TIME_BOUNDARY_PLANNING = false;
+  public static final boolean DEFAULT_WINDOWING_STRICT_VALIDATION = true;
 
   @SuppressWarnings("unused") // Used by Jackson serialization
   public enum Vectorize

processing/src/test/java/org/apache/druid/query/QueryContextsTest.java

@@ -151,6 +151,15 @@ public void testDefaultPlanTimeBoundarySql()
     );
   }
 
+  @Test
+  public void testDefaultWindowingStrictValidation()
+  {
+    Assert.assertEquals(
+        QueryContexts.DEFAULT_WINDOWING_STRICT_VALIDATION,
+        QueryContext.empty().isWindowingStrictValidation()
+    );
+  }
+
   @Test
   public void testGetEnableJoinLeftScanDirect()
   {

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

@@ -25,13 +25,19 @@
 import org.apache.calcite.runtime.CalciteContextException;
 import org.apache.calcite.runtime.CalciteException;
 import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlOperatorTable;
+import org.apache.calcite.sql.SqlUtil;
+import org.apache.calcite.sql.SqlWindow;
 import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.calcite.sql.validate.SqlValidatorScope;
+import org.apache.calcite.util.Util;
 import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.QueryContexts;
 import org.apache.druid.sql.calcite.run.EngineFeature;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
  * Druid extended SQL validator. (At present, it doesn't actually
@@ -53,6 +59,100 @@
     this.plannerContext = plannerContext;
   }
 
+  @Override
+  public void validateWindow(SqlNode windowOrId, SqlValidatorScope scope, @Nullable SqlCall call)
+  {
+    final SqlWindow targetWindow;
+    switch (windowOrId.getKind()) {
+      case IDENTIFIER:
+        targetWindow = getWindowByName((SqlIdentifier) windowOrId, scope);
+        break;
+      case WINDOW:
+        targetWindow = (SqlWindow) windowOrId;
+        break;
+      default:
+        throw Util.unexpected(windowOrId.getKind());
+    }
+
+
+    @Nullable
+    SqlNode lowerBound = targetWindow.getLowerBound();
+    @Nullable
+    SqlNode upperBound = targetWindow.getUpperBound();
+    if (!isValidEndpoint(lowerBound) || !isValidEndpoint(upperBound)) {
+      throw buildCalciteContextException(
+          "Window frames with expression based lower/upper bounds are not supported.",
+          windowOrId
+      );
+    }
+
+    if (isPrecedingOrFollowing(lowerBound) &&
+        isPrecedingOrFollowing(upperBound) &&
+        lowerBound.getKind() == upperBound.getKind()) {
+      // this limitation can be lifted when https://github.com/apache/druid/issues/15739 is addressed
+      throw buildCalciteContextException(
+          StringUtils.format(
+              "Query bounds with both lower and upper bounds as PRECEDING or FOLLOWING is not supported.",
+              QueryContexts.WINDOWING_STRICT_VALIDATION
+          ),
+          windowOrId
+      );
+    }
+
+    if (plannerContext.queryContext().isWindowingStrictValidation()) {
+      if (!targetWindow.isRows() &&
+          (!isValidRangeEndpoint(lowerBound) || !isValidRangeEndpoint(upperBound))) {
+        // this limitation can be lifted when https://github.com/apache/druid/issues/15767 is addressed
+        throw buildCalciteContextException(
+            StringUtils.format(
+                "The query contains a window frame which may return incorrect results. To disregard this warning, set [%s] to false in the query context.",
+                QueryContexts.WINDOWING_STRICT_VALIDATION
+            ),
+            windowOrId
+        );
+      }
+    }
+
+    super.validateWindow(windowOrId, scope, call);
+  }
+
+  private boolean isPrecedingOrFollowing(@Nullable SqlNode bound)
+  {
+    if (bound == null) {
+      return false;
+    }
+    SqlKind kind = bound.getKind();
+    return kind == SqlKind.PRECEDING || kind == SqlKind.FOLLOWING;
+  }
+
+  /**
+   * Checks if the given endpoint is acceptable.
+   */
+  private boolean isValidEndpoint(@Nullable SqlNode bound)
+  {
+    if (isValidRangeEndpoint(bound)) {
+      return true;
+    }
+    if (bound.getKind() == SqlKind.FOLLOWING || bound.getKind() == SqlKind.PRECEDING) {
+      final SqlNode boundVal = ((SqlCall) bound).operand(0);
+      if (SqlUtil.isLiteral(boundVal)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  /**
+   * Checks if the given endpoint is valid for a RANGE window frame.
+   */
+  private boolean isValidRangeEndpoint(@Nullable SqlNode bound)
+  {
+    return bound == null
+        || SqlWindow.isCurrentRow(bound)
+        || SqlWindow.isUnboundedFollowing(bound)
+        || SqlWindow.isUnboundedPreceding(bound);
+  }
+
   @Override
   public void validateCall(SqlCall call, SqlValidatorScope scope)
   {
@@ -80,7 +180,7 @@
     super.validateCall(call, scope);
   }
 
-  private CalciteContextException buildCalciteContextException(String message, SqlCall call)
+  private CalciteContextException buildCalciteContextException(String message, SqlNode call)
   {
     SqlParserPos pos = call.getParserPosition();
     return new CalciteContextException(message,

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

@@ -85,7 +85,7 @@ public class PlannerContext
   public static final String CTX_SQL_OUTER_LIMIT = "sqlOuterLimit";
 
   /**
-   * Undocumented context key, used to enable window functions.
+   * Key to enable window functions.
    */
   public static final String CTX_ENABLE_WINDOW_FNS = "enableWindowing";
 

sql/src/test/java/org/apache/druid/sql/calcite/BaseCalciteQueryTest.java

@@ -89,11 +89,13 @@
 import org.apache.druid.sql.SqlStatementFactory;
 import org.apache.druid.sql.calcite.QueryTestRunner.QueryResults;
 import org.apache.druid.sql.calcite.expression.DruidExpression;
+import org.apache.druid.sql.calcite.expression.ExpressionTestHelper;
 import org.apache.druid.sql.calcite.planner.Calcites;
 import org.apache.druid.sql.calcite.planner.PlannerConfig;
 import org.apache.druid.sql.calcite.planner.PlannerContext;
 import org.apache.druid.sql.calcite.planner.PlannerFactory;
 import org.apache.druid.sql.calcite.rule.ExtensionCalciteRuleProvider;
+import org.apache.druid.sql.calcite.run.EngineFeature;
 import org.apache.druid.sql.calcite.run.SqlEngine;
 import org.apache.druid.sql.calcite.schema.DruidSchemaManager;
 import org.apache.druid.sql.calcite.util.CalciteTestBase;
@@ -136,6 +138,7 @@
 import java.util.stream.Collectors;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assume.assumeTrue;
 
 /**
  * A base class for SQL query testing. It sets up query execution environment, provides useful helper methods,
@@ -738,6 +741,13 @@ public void finalizePlanner(PlannerFixture plannerFixture)
     basePlannerComponentSupplier.finalizePlanner(plannerFixture);
   }
 
+  public void assumeFeatureAvailable(EngineFeature feature)
+  {
+    boolean featureAvailable = queryFramework().engine()
+        .featureAvailable(feature, ExpressionTestHelper.PLANNER_CONTEXT);
+    assumeTrue(StringUtils.format("test disabled; feature [%s] is not available!", feature), featureAvailable);
+  }
+
   public void assertQueryIsUnplannable(final String sql, String expectedError)
   {
     assertQueryIsUnplannable(PLANNER_CONFIG_DEFAULT, sql, expectedError);

sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java

@@ -118,6 +118,7 @@
 import org.apache.druid.sql.calcite.planner.PlannerConfig;
 import org.apache.druid.sql.calcite.planner.PlannerContext;
 import org.apache.druid.sql.calcite.rel.CannotBuildQueryException;
+import org.apache.druid.sql.calcite.run.EngineFeature;
 import org.apache.druid.sql.calcite.util.CalciteTests;
 import org.apache.druid.sql.calcite.util.TestDataBuilder;
 import org.hamcrest.CoreMatchers;
@@ -14813,6 +14814,51 @@ public void testUnSupportedNullsLast()
     assertThat(e, invalidSqlIs("ASCENDING ordering with NULLS LAST is not supported! (line [1], column [41])"));
   }
 
+  @Test
+  public void testUnSupportedRangeBounds()
+  {
+    assumeFeatureAvailable(EngineFeature.WINDOW_FUNCTIONS);
+
+    DruidException e = assertThrows(DruidException.class, () -> testBuilder()
+        .queryContext(ImmutableMap.of(PlannerContext.CTX_ENABLE_WINDOW_FNS, true))
+        .sql("SELECT dim1,ROW_NUMBER() OVER (ORDER BY dim1 RANGE BETWEEN 3 PRECEDING AND 2 FOLLOWING) from druid.foo")
+        .run());
+    assertThat(e, invalidSqlIs("The query contains a window frame which may return incorrect results. To disregard this warning, set [windowingStrictValidation] to false in the query context. (line [1], column [31])"));
+  }
+
+  @Test
+  public void testUnSupportedWindowBoundExpressions()
+  {
+    assumeFeatureAvailable(EngineFeature.WINDOW_FUNCTIONS);
+
+    DruidException e = assertThrows(DruidException.class, () -> testBuilder()
+        .queryContext(ImmutableMap.of(PlannerContext.CTX_ENABLE_WINDOW_FNS, true))
+        .sql("SELECT dim1,ROW_NUMBER() OVER (ORDER BY dim1 ROWS BETWEEN dim1 PRECEDING AND dim1 FOLLOWING) from druid.foo")
+        .run());
+    assertThat(e, invalidSqlIs("Window frames with expression based lower/upper bounds are not supported. (line [1], column [31])"));
+  }
+
+
+  @Test
+  public void testUnSupportedWindowBoundTypes()
+  {
+    assumeFeatureAvailable(EngineFeature.WINDOW_FUNCTIONS);
+
+    DruidException e;
+    e = assertThrows(DruidException.class, () -> testBuilder()
+        .queryContext(ImmutableMap.of(PlannerContext.CTX_ENABLE_WINDOW_FNS, true))
+        .sql("SELECT dim1,ROW_NUMBER() OVER (ORDER BY dim1 ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING) from druid.foo")
+        .run());
+    assertThat(e, invalidSqlIs("Query bounds with both lower and upper bounds as PRECEDING or FOLLOWING is not supported. (line [1], column [31])"));
+
+    e = assertThrows(DruidException.class, () -> testBuilder()
+        .queryContext(ImmutableMap.of(PlannerContext.CTX_ENABLE_WINDOW_FNS, true))
+        .sql("SELECT dim1,ROW_NUMBER() OVER (ORDER BY dim1 ROWS BETWEEN 1 FOLLOWING AND 1 FOLLOWING) from druid.foo")
+        .run());
+    assertThat(e, invalidSqlIs("Query bounds with both lower and upper bounds as PRECEDING or FOLLOWING is not supported. (line [1], column [31])"));
+  }
+
+
   @Test
   public void testWindowingErrorWithoutFeatureFlag()
   {

sql/src/test/java/org/apache/druid/sql/calcite/CalciteWindowQueryTest.java

@@ -212,8 +212,11 @@ public void windowQueryTest() throws Exception
       testBuilder()
           .skipVectorize(true)
           .sql(testCase.getSql())
-          .queryContext(ImmutableMap.of(PlannerContext.CTX_ENABLE_WINDOW_FNS, true,
-              QueryContexts.ENABLE_DEBUG, true))
+          .queryContext(ImmutableMap.of(
+              PlannerContext.CTX_ENABLE_WINDOW_FNS, true,
+              QueryContexts.ENABLE_DEBUG, true,
+              QueryContexts.WINDOWING_STRICT_VALIDATION, false
+              ))
           .addCustomVerification(QueryVerification.ofResults(testCase))
           .run();
     }

sql/src/test/java/org/apache/druid/sql/calcite/DrillWindowQueryTest.java

@@ -5593,8 +5593,8 @@ public void test_lag_func_lag_Fn_66()
   public void test_lag_func_lag_Fn_67()
   {
     windowQueryTest();
-  }
 
+  }
   @NotYetSupported(Modes.UNSUPPORTED_NULL_ORDERING)
   @DrillTest("lag_func/lag_Fn_68")
   @Test

sql/src/test/java/org/apache/druid/sql/calcite/expression/ExpressionTestHelper.java

@@ -77,7 +77,7 @@
 import java.util.Objects;
 import java.util.stream.Collectors;
 
-class ExpressionTestHelper
+public class ExpressionTestHelper
 {
   private static final JoinableFactoryWrapper JOINABLE_FACTORY_WRAPPER = CalciteTests.createJoinableFactoryWrapper();
   private static final PlannerToolbox PLANNER_TOOLBOX = new PlannerToolbox(
@@ -99,7 +99,7 @@ NamedViewSchema.NAME, new NamedViewSchema(EasyMock.createMock(ViewSchema.class))
       CalciteTests.TEST_AUTHORIZER_MAPPER,
       AuthConfig.newBuilder().build()
   );
-  private static final PlannerContext PLANNER_CONTEXT = PlannerContext.create(
+  public static final PlannerContext PLANNER_CONTEXT = PlannerContext.create(
       PLANNER_TOOLBOX,
       "SELECT 1", // The actual query isn't important for this test
       null, /* Don't need engine */