[Backport] Fix columns with null values in windowing expressions

processing/src/main/java/org/apache/druid/query/rowsandcols/column/accessor/DoubleColumnAccessorBase.java

@@ -36,7 +36,11 @@ public ColumnType getType()
   @Override
   public Object getObject(int rowNum)
   {
-    return getDouble(rowNum);
+    if (isNull(rowNum)) {
+      return null;
+    } else {
+      return getDouble(rowNum);
+    }
   }
 
   @Override

processing/src/main/java/org/apache/druid/query/rowsandcols/column/accessor/FloatColumnAccessorBase.java

@@ -36,7 +36,11 @@ public ColumnType getType()
   @Override
   public Object getObject(int rowNum)
   {
-    return getFloat(rowNum);
+    if (isNull(rowNum)) {
+      return null;
+    } else {
+      return getFloat(rowNum);
+    }
   }
 
   @Override

processing/src/main/java/org/apache/druid/query/rowsandcols/column/accessor/LongColumnAccessorBase.java

@@ -34,9 +34,13 @@ public ColumnType getType()
 
   @Nullable
   @Override
-  public Object getObject(int rowNum)
+  public final Object getObject(int rowNum)
   {
-    return getLong(rowNum);
+    if (isNull(rowNum)) {
+      return null;
+    } else {
+      return getLong(rowNum);
+    }
   }
 
   @Override

processing/src/test/java/org/apache/druid/query/rowsandcols/column/ColumnAccessorsTest.java

@@ -0,0 +1,188 @@
+/*
+ * 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.query.rowsandcols.column;
+
+import org.apache.druid.query.rowsandcols.column.accessor.DoubleColumnAccessorBase;
+import org.apache.druid.query.rowsandcols.column.accessor.FloatColumnAccessorBase;
+import org.apache.druid.query.rowsandcols.column.accessor.LongColumnAccessorBase;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+@RunWith(Parameterized.class)
+public class ColumnAccessorsTest
+{
+  private TestAccessorShim testAccessor;
+
+  @Parameters
+  public static List<Object[]> getParameters()
+  {
+    List<Object[]> ret = new ArrayList<>();
+
+    for (TestAccessorShim accessor : TestAccessorShim.values()) {
+      ret.add(new Object[] {accessor});
+    }
+
+    return ret;
+  }
+
+  enum TestAccessorShim
+  {
+    LONG {
+      @Override
+      ColumnAccessor getColumnAccessor(Object value)
+      {
+        Long val = (Long) value;
+        return new LongColumnAccessorBase()
+        {
+          @Override
+          public int numRows()
+          {
+            return 1;
+          }
+
+          @Override
+          public boolean isNull(int rowNum)
+          {
+            return val == null;
+          }
+
+          @Override
+          public long getLong(int rowNum)
+          {
+            return val;
+          }
+        };
+      }
+
+      @Override
+      protected Object getSomeValue()
+      {
+        return 42L;
+      }
+
+
+    },
+    FLOAT {
+      @Override
+      ColumnAccessor getColumnAccessor(Object value)
+      {
+        Float val = (Float) value;
+        return new FloatColumnAccessorBase()
+        {
+
+          @Override
+          public int numRows()
+          {
+            return 1;
+          }
+
+          @Override
+          public boolean isNull(int rowNum)
+          {
+            return val == null;
+          }
+
+          @Override
+          public float getFloat(int rowNum)
+          {
+            return val;
+          }
+        };
+      }
+
+      @Override
+      protected Object getSomeValue()
+      {
+        return 42.1F;
+      }
+    },
+    DOUBLE {
+      @Override
+      ColumnAccessor getColumnAccessor(Object value)
+      {
+        Double val = (Double) value;
+        return new DoubleColumnAccessorBase()
+        {
+
+          @Override
+          public int numRows()
+          {
+            return 1;
+          }
+
+          @Override
+          public boolean isNull(int rowNum)
+          {
+            return val == null;
+          }
+
+          @Override
+          public double getDouble(int rowNum)
+          {
+            return val;
+          }
+        };
+      }
+
+      @Override
+      protected Object getSomeValue()
+      {
+        return 42.1D;
+      }
+    };
+
+    abstract ColumnAccessor getColumnAccessor(Object val);
+
+    protected abstract Object getSomeValue();
+  }
+
+  public ColumnAccessorsTest(TestAccessorShim accessor)
+  {
+    this.testAccessor = accessor;
+  }
+
+  @Test
+  public void testSomeValue()
+  {
+    Object expectedValue = testAccessor.getSomeValue();
+    ColumnAccessor acc = testAccessor.getColumnAccessor(expectedValue);
+
+    assertFalse(acc.isNull(0));
+    assertEquals(expectedValue, acc.getObject(0));
+  }
+
+  @Test
+  public void testNull()
+  {
+    ColumnAccessor acc = testAccessor.getColumnAccessor(null);
+
+    assertTrue(acc.isNull(0));
+    assertEquals(null, acc.getObject(0));
+  }
+}

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

@@ -127,6 +127,7 @@ public void verifyResults(QueryResults results) throws Exception
       }
       Assert.assertEquals(1, results.recordedQueries.size());
 
+      maybeDumpActualResults(results.results);
       final WindowOperatorQuery query = getWindowOperatorQuery(results.recordedQueries);
       for (int i = 0; i < input.expectedOperators.size(); ++i) {
         final OperatorFactory expectedOperator = input.expectedOperators.get(i);
@@ -145,7 +146,6 @@ public void verifyResults(QueryResults results) throws Exception
         Assert.assertEquals(types[i], results.signature.getColumnType(i).get());
       }
 
-      maybeDumpActualResults(results.results);
       for (Object[] result : input.expectedResults) {
         for (int i = 0; i < result.length; i++) {
           // Jackson deserializes numbers as the minimum size required to

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

@@ -6208,7 +6208,6 @@ public void test_aggregates_aggOWnFn_15()
     windowQueryTest();
   }
 
-  @NotYetSupported(Modes.RESULT_MISMATCH)
   @DrillTest("aggregates/aggOWnFn_17")
   @Test
   public void test_aggregates_aggOWnFn_17()
@@ -6727,7 +6726,6 @@ public void test_frameclause_defaultFrame_RBUPACR_bgint_4()
     windowQueryTest();
   }
 
-  @NotYetSupported(Modes.RESULT_MISMATCH)
   @DrillTest("frameclause/defaultFrame/RBUPACR_bgint_6")
   @Test
   public void test_frameclause_defaultFrame_RBUPACR_bgint_6()
@@ -6783,7 +6781,6 @@ public void test_frameclause_defaultFrame_RBUPACR_dbl_3()
     windowQueryTest();
   }
 
-  @NotYetSupported(Modes.RESULT_MISMATCH)
   @DrillTest("frameclause/defaultFrame/RBUPACR_dbl_6")
   @Test
   public void test_frameclause_defaultFrame_RBUPACR_dbl_6()
@@ -6839,7 +6836,6 @@ public void test_frameclause_defaultFrame_RBUPACR_int10()
     windowQueryTest();
   }
 
-  @NotYetSupported(Modes.RESULT_MISMATCH)
   @DrillTest("frameclause/defaultFrame/RBUPACR_int13")
   @Test
   public void test_frameclause_defaultFrame_RBUPACR_int13()
@@ -7119,7 +7115,6 @@ public void test_frameclause_RBUPACR_RBUPACR_bgint_4()
     windowQueryTest();
   }
 
-  @NotYetSupported(Modes.RESULT_MISMATCH)
   @DrillTest("frameclause/RBUPACR/RBUPACR_bgint_6")
   @Test
   public void test_frameclause_RBUPACR_RBUPACR_bgint_6()
@@ -7175,7 +7170,6 @@ public void test_frameclause_RBUPACR_RBUPACR_dbl_3()
     windowQueryTest();
   }
 
-  @NotYetSupported(Modes.RESULT_MISMATCH)
   @DrillTest("frameclause/RBUPACR/RBUPACR_dbl_6")
   @Test
   public void test_frameclause_RBUPACR_RBUPACR_dbl_6()
@@ -7199,7 +7193,6 @@ public void test_frameclause_RBUPACR_RBUPACR_int10()
     windowQueryTest();
   }
 
-  @NotYetSupported(Modes.RESULT_MISMATCH)
   @DrillTest("frameclause/RBUPACR/RBUPACR_int13")
   @Test
   public void test_frameclause_RBUPACR_RBUPACR_int13()
@@ -7239,7 +7232,6 @@ public void test_frameclause_RBUPAUF_RBUPAUF_bgint_4()
     windowQueryTest();
   }
 
-  @NotYetSupported(Modes.RESULT_MISMATCH)
   @DrillTest("frameclause/RBUPAUF/RBUPAUF_bgint_6")
   @Test
   public void test_frameclause_RBUPAUF_RBUPAUF_bgint_6()
@@ -7255,15 +7247,13 @@ public void test_frameclause_RBUPAUF_RBUPAUF_char_3()
     windowQueryTest();
   }
 
-  @NotYetSupported(Modes.RESULT_MISMATCH)
   @DrillTest("frameclause/RBUPAUF/RBUPAUF_dbl_6")
   @Test
   public void test_frameclause_RBUPAUF_RBUPAUF_dbl_6()
   {
     windowQueryTest();
   }
 
-  @NotYetSupported(Modes.RESULT_MISMATCH)
   @DrillTest("frameclause/RBUPAUF/RBUPAUF_dbl_7")
   @Test
   public void test_frameclause_RBUPAUF_RBUPAUF_dbl_7()
@@ -7303,7 +7293,6 @@ public void test_frameclause_RBUPAUF_RBUPAUF_dt_5()
     windowQueryTest();
   }
 
-  @NotYetSupported(Modes.RESULT_MISMATCH)
   @DrillTest("frameclause/RBUPAUF/RBUPAUF_int_13")
   @Test
   public void test_frameclause_RBUPAUF_RBUPAUF_int_13()

sql/src/test/resources/calcite/tests/window/windowed_long_null.sqlTest

@@ -0,0 +1,30 @@
+type: "operatorValidation"
+
+sql: |
+   SELECT
+     l2,
+     MIN(l2) OVER(partition by l2)
+   FROM druid.numfoo
+   WHERE l2 is null or l2 = -1111 or l2 = 0
+
+
+expectedOperators:
+  - type: "naiveSort"
+    columns:
+      - column: "l2"
+        direction: "ASC"
+  - type: "naivePartition"
+    partitionColumns: [ "l2" ]
+  - type: "window"
+    processor:
+      type: "framedAgg"
+      frame: { peerType: "ROWS", lowUnbounded: true, lowOffset: 0, uppUnbounded: true, uppOffset: 0 }
+      aggregations:
+        - { type: "longMin", name: "w0", fieldName: "l2" }
+
+expectedResults:
+  - [null,null]
+  - [null,null]
+  - [null,null]
+  - [null,null]
+  - [0,0]