SQL version of unnest native druid function

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

@@ -125,7 +125,7 @@ private static boolean rowsEqual(final Iterable<Object[]> rowsA, final Iterable<
         final Object[] rowA = listA.get(i);
         final Object[] rowB = listB.get(i);
 
-        if (!Arrays.equals(rowA, rowB)) {
+        if (!Arrays.deepEquals(rowA, rowB)) {
           return false;
         }
       }

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

@@ -207,6 +207,18 @@ public int hashCode()
   {
     return Objects.hash(base, column, outputName);
   }
+
+  @Override
+  public String toString()
+  {
+    return "UnnestDataSource{" +
+           "base=" + base +
+           ", column='" + column + '\'' +
+           ", outputName='" + outputName + '\'' +
+           ", allowList=" + allowList +
+           '}';
+  }
+
 }
 
 

processing/src/main/java/org/apache/druid/segment/UnnestColumnValueSelectorCursor.java

@@ -100,7 +100,31 @@ public DimensionSelector makeDimensionSelector(DimensionSpec dimensionSpec)
         if (!outputName.equals(dimensionSpec.getDimension())) {
           return baseColumnSelectorFactory.makeDimensionSelector(dimensionSpec);
         }
-        throw new UOE("Unsupported dimension selector while using column value selector for column [%s]", outputName);
+        // this is done to support virtual columns
+        // In future a developer should move towards making sure that
+        // for all dictionary encoded cases we only get the dimension selector
+        return new BaseSingleValueDimensionSelector()
+        {
+          final ColumnValueSelector colSelector = makeColumnValueSelector(dimensionSpec.getDimension());
+
+          @Nullable
+          @Override
+          protected String getValue()
+          {
+            final Object returnedObj = colSelector.getObject();
+            if (returnedObj == null) {
+              return null;
+            } else {
+              return String.valueOf(returnedObj);
+            }
+          }
+
+          @Override
+          public void inspectRuntimeShape(RuntimeShapeInspector inspector)
+          {
+            colSelector.inspectRuntimeShape(inspector);
+          }
+        };
       }
 
       @Override

processing/src/main/java/org/apache/druid/segment/UnnestDimensionCursor.java

@@ -181,7 +181,10 @@ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
           @Override
           public Object getObject()
           {
-            if (indexedIntsForCurrentRow == null) {
+            if (dimSelector.getObject() == null) {
+              return null;
+            }
+            if (indexedIntsForCurrentRow.size() == 0) {
               return null;
             }
             if (allowedBitSet.isEmpty()) {
@@ -409,7 +412,12 @@ public int size()
     @Override
     public int get(int idx)
     {
-      return indexedIntsForCurrentRow.get(index);
+      // to support rows which have only null values
+      // need to check if rhe value is not null and the size is greater than 0
+      if (indexedIntsForCurrentRow != null && indexedIntsForCurrentRow.size() > 0) {
+        return indexedIntsForCurrentRow.get(index);
+      }
+      return 0;
     }
   }
 }

sql/src/main/java/org/apache/druid/sql/calcite/expression/Expressions.java

@@ -24,6 +24,7 @@
 import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
 import org.apache.calcite.rel.core.Project;
 import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexFieldAccess;
 import org.apache.calcite.rex.RexInputRef;
 import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.rex.RexNode;
@@ -34,6 +35,7 @@
 import org.apache.druid.common.config.NullHandling;
 import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.granularity.Granularity;
 import org.apache.druid.math.expr.Expr;
 import org.apache.druid.math.expr.ExprMacroTable;
@@ -58,6 +60,7 @@
 import org.apache.druid.sql.calcite.filtration.Filtration;
 import org.apache.druid.sql.calcite.planner.Calcites;
 import org.apache.druid.sql.calcite.planner.PlannerContext;
+import org.apache.druid.sql.calcite.rel.CannotBuildQueryException;
 import org.apache.druid.sql.calcite.rel.VirtualColumnRegistry;
 import org.apache.druid.sql.calcite.table.RowSignatures;
 import org.joda.time.Interval;
@@ -214,12 +217,43 @@ public static DruidExpression toDruidExpressionWithPostAggOperands(
       return rexCallToDruidExpression(plannerContext, rowSignature, rexNode, postAggregatorVisitor);
     } else if (kind == SqlKind.LITERAL) {
       return literalToDruidExpression(plannerContext, rexNode);
+    } else if (kind == SqlKind.FIELD_ACCESS) {
+      return fieldAccessToDruidExpression(rowSignature, rexNode);
     } else {
       // Can't translate.
       return null;
     }
   }
 
+  private static DruidExpression fieldAccessToDruidExpression(
+      final RowSignature rowSignature,
+      final RexNode rexNode
+  )
+  {
+    // Translate field references.
+    final RexFieldAccess ref = (RexFieldAccess) rexNode;
+    // This case arises in the case of a correlation where the rexNode points to a table from the left subtree
+    // while the underlying datasource is the scan stub created from LogicalValuesRule
+    // In such a case we throw a CannotBuildQueryException so that Calcite does not go ahead with this path
+    // This exception is caught while returning false from isValidDruidQuery() method
+
+    //use this index to return
+    final int index = rowSignature.getColumnNames().indexOf(ref.getField().getName());
+    if (ref.getField().getIndex() > rowSignature.size()) {
+      throw new CannotBuildQueryException(StringUtils.format(
+          "Cannot build query as column name [%s] does not exist in row [%s]", ref.getField().getName(), rowSignature)
+      );
+    }
+
+    final String columnName = rowSignature.getColumnName(index);
+    final Optional<ColumnType> columnType = rowSignature.getColumnType(index);
+    if (columnName == null) {
+      throw new ISE("Expression referred to nonexistent index [%d] in row [%s]", index, rowSignature);
+    }
+
+    return DruidExpression.ofColumn(columnType.orElse(null), columnName);
+  }
+
   private static DruidExpression inputRefToDruidExpression(
       final RowSignature rowSignature,
       final RexNode rexNode