Fix VirtualColumn related issues in window expressions

processing/src/main/java/org/apache/druid/frame/write/columnar/FrameColumnWriters.java

@@ -19,7 +19,6 @@
 
 package org.apache.druid.frame.write.columnar;
 
-import org.apache.druid.common.config.NullHandling;
 import org.apache.druid.frame.allocation.MemoryAllocator;
 import org.apache.druid.frame.write.UnsupportedColumnTypeException;
 import org.apache.druid.java.util.common.ISE;
@@ -167,9 +166,7 @@ private static ComplexFrameColumnWriter makeComplexWriter(
 
   private static boolean hasNullsForNumericWriter(final ColumnCapabilities capabilities)
   {
-    if (NullHandling.replaceWithDefault()) {
-      return false;
-    } else if (capabilities == null) {
+    if (capabilities == null) {
       return true;
     } else if (capabilities.getType().isNumeric()) {
       return capabilities.hasNulls().isMaybeTrue();

processing/src/main/java/org/apache/druid/query/rowsandcols/LazilyDecoratedRowsAndColumns.java

@@ -102,6 +102,11 @@ public Collection<String> getColumnNames()
     return viewableColumns == null ? base.getColumnNames() : viewableColumns;
   }
 
+  public RowsAndColumns getBase()
+  {
+    return base;
+  }
+
   @Override
   public int numRows()
   {

processing/src/main/java/org/apache/druid/query/rowsandcols/semantic/DefaultColumnSelectorFactoryMaker.java

@@ -140,14 +140,18 @@ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
     public ColumnValueSelector makeColumnValueSelector(@Nonnull String columnName)
     {
       return withColumnAccessor(columnName, columnAccessor -> {
-        final ColumnType type = columnAccessor.getType();
-        switch (type.getType()) {
-          case STRING:
-            return new StringColumnValueSelector(columnAccessor);
-          case COMPLEX:
-            return new ComplexColumnValueSelector(columnAccessor);
-          default:
-            return new PassThroughColumnValueSelector(columnAccessor);
+        if (columnAccessor == null) {
+          return DimensionSelector.nilSelector();
+        } else {
+          final ColumnType type = columnAccessor.getType();
+          switch (type.getType()) {
+            case STRING:
+              return new StringColumnValueSelector(columnAccessor);
+            case COMPLEX:
+              return new ComplexColumnValueSelector(columnAccessor);
+            default:
+              return new PassThroughColumnValueSelector(columnAccessor);
+          }
         }
       });
     }
@@ -156,23 +160,26 @@ public ColumnValueSelector makeColumnValueSelector(@Nonnull String columnName)
     @Override
     public ColumnCapabilities getColumnCapabilities(String column)
     {
-      return withColumnAccessor(column, columnAccessor ->
-          new ColumnCapabilitiesImpl()
+      return withColumnAccessor(column, columnAccessor -> {
+        if (columnAccessor == null) {
+          return null;
+        } else {
+          return new ColumnCapabilitiesImpl()
               .setType(columnAccessor.getType())
               .setHasMultipleValues(false)
               .setDictionaryEncoded(false)
-              .setHasBitmapIndexes(false));
+              .setHasBitmapIndexes(false);
+        }
+      });
     }
 
     private <T> T withColumnAccessor(String column, Function<ColumnAccessor, T> fn)
     {
+      @Nullable
       ColumnAccessor retVal = accessorCache.get(column);
       if (retVal == null) {
         Column racColumn = rac.findColumn(column);
-        if (racColumn == null) {
-          throw DruidException.defensive("Column[%s] not found!", column);
-        }
-        retVal = racColumn.toAccessor();
+        retVal = racColumn == null ? null : racColumn.toAccessor();
         accessorCache.put(column, retVal);
       }
       return fn.apply(retVal);

processing/src/test/java/org/apache/druid/query/rowsandcols/concrete/FrameRowsAndColumnsTest.java

@@ -22,17 +22,13 @@
 import org.apache.druid.query.rowsandcols.LazilyDecoratedRowsAndColumns;
 import org.apache.druid.query.rowsandcols.MapOfColumnsRowsAndColumns;
 import org.apache.druid.query.rowsandcols.RowsAndColumnsTestBase;
-import java.lang.reflect.Field;
 import java.util.function.Function;
 
-import static org.junit.Assert.assertTrue;
-
 public class FrameRowsAndColumnsTest extends RowsAndColumnsTestBase
 {
-
-  public FrameRowsAndColumnsTest(Class<?> expectedClass)
+  public FrameRowsAndColumnsTest()
   {
-    super(expectedClass);
+    super(FrameRowsAndColumns.class);
   }
 
   public static Function<MapOfColumnsRowsAndColumns, FrameRowsAndColumns> MAKER = input -> {
@@ -46,15 +42,6 @@ private static FrameRowsAndColumns buildFrame(MapOfColumnsRowsAndColumns input)
 
     rac.numRows(); // materialize
 
-    try {
-      Field baseField = rac.getClass().getDeclaredField("base");
-      baseField.setAccessible(true);
-      Object o = baseField.get(rac);
-      assertTrue(o instanceof FrameRowsAndColumns);
-      return (FrameRowsAndColumns) o;
-    }
-    catch (Exception e) {
-      throw new RuntimeException(e);
-    }
+    return (FrameRowsAndColumns) rac.getBase();
   }
 }

processing/src/test/java/org/apache/druid/query/rowsandcols/semantic/CombinedSemanticInterfacesTest.java

@@ -20,8 +20,6 @@
 package org.apache.druid.query.rowsandcols.semantic;
 
 import com.google.common.collect.ImmutableMap;
-import org.apache.druid.error.DruidException;
-import org.apache.druid.error.DruidExceptionMatcher;
 import org.apache.druid.query.aggregation.AggregatorFactory;
 import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
 import org.apache.druid.query.operator.ColumnWithDirection;
@@ -33,6 +31,7 @@
 import org.apache.druid.query.rowsandcols.column.IntArrayColumn;
 import org.apache.druid.query.rowsandcols.column.ObjectArrayColumn;
 import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.DimensionSelector;
 import org.apache.druid.segment.column.ColumnType;
 import org.junit.Assert;
 import org.junit.Test;
@@ -44,8 +43,8 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
 
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 
 /**
  * Place where tests can live that are testing the interactions of multiple semantic interfaces
@@ -62,15 +61,27 @@ public CombinedSemanticInterfacesTest(
   }
 
   @Test
-  public void testColumnSelectorFactoryNonExistentColumn()
+  public void testColumnSelectorFactoryMakeColumnValueSelectorNonExistentColumn()
   {
     RowsAndColumns rac = make(MapOfColumnsRowsAndColumns.fromMap(
         ImmutableMap.of(
             "some", new IntArrayColumn(new int[] {3, 54, 21, 1, 5, 54, 2, 3, 92}))));
     AtomicInteger currRow = new AtomicInteger();
     ColumnSelectorFactory csfm = ColumnSelectorFactoryMaker.fromRAC(rac).make(currRow);
-    DruidException e = assertThrows(DruidException.class, () -> csfm.makeColumnValueSelector("nonexistent"));
-    assertThat(e, DruidExceptionMatcher.defensive().expectMessageIs("Column[nonexistent] not found!"));
+
+    assertEquals(DimensionSelector.nilSelector(), csfm.makeColumnValueSelector("nonexistent"));
+  }
+
+  @Test
+  public void testColumnSelectorFactoryGetColumnCapabilitiesNonExistentColumn()
+  {
+    RowsAndColumns rac = make(MapOfColumnsRowsAndColumns.fromMap(
+        ImmutableMap.of(
+            "some", new IntArrayColumn(new int[] {3, 54, 21, 1, 5, 54, 2, 3, 92}))));
+    AtomicInteger currRow = new AtomicInteger();
+    ColumnSelectorFactory csfm = ColumnSelectorFactoryMaker.fromRAC(rac).make(currRow);
+
+    assertNull(csfm.getColumnCapabilities("nonexistent"));
   }
 
   /**

sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidQuery.java

@@ -1449,7 +1449,9 @@ private WindowOperatorQuery toWindowQuery()
     VirtualColumns virtualColumns = virtualColumnRegistry.build(Collections.emptySet());
     final List<OperatorFactory> operators;
 
-    if (!virtualColumns.isEmpty()) {
+    if (virtualColumns.isEmpty()) {
+      operators = windowing.getOperators();
+    } else {
       operators = ImmutableList.<OperatorFactory>builder()
           .add(new ScanOperatorFactory(
               null,
@@ -1460,8 +1462,6 @@ private WindowOperatorQuery toWindowQuery()
               null))
           .addAll(windowing.getOperators())
           .build();
-    } else {
-      operators = windowing.getOperators();
     }
     return new WindowOperatorQuery(
         dataSource,