cleanup

Author: richardc-db

build.sbt

@@ -153,7 +153,7 @@ lazy val commonSettings = Seq(
     2) delta-spark unidoc fails to compile. spark 3.5 is on its classpath. likely due to iceberg
        issue above.
  */
-def crossSparkProjectSettings(): Seq[Setting[_]] = getSparkVersion() match {
+def crossSparkUniDocSettings(): Seq[Setting[_]] = getSparkVersion() match {
   case LATEST_RELEASED_SPARK_VERSION => Seq(
     // Java-/Scala-/Uni-Doc Settings
     scalacOptions ++= Seq(
@@ -248,6 +248,7 @@ lazy val connectCommon = (project in file("spark-connect/common"))
     name := "delta-connect-common",
     commonSettings,
     crossSparkSettings(),
+    crossSparkUniDocSettings(),
     releaseSettings,
     Compile / compile := runTaskOnlyOnSparkMaster(
       task = Compile / compile,
@@ -307,6 +308,7 @@ lazy val connectServer = (project in file("spark-connect/server"))
       emptyValue = ()
     ).value,
     crossSparkSettings(),
+    crossSparkUniDocSettings(),
     libraryDependencies ++= Seq(
       "com.google.protobuf" % "protobuf-java" % protoVersion % "protobuf",
 
@@ -334,7 +336,7 @@ lazy val spark = (project in file("spark"))
     sparkMimaSettings,
     releaseSettings,
     crossSparkSettings(),
-    crossDeltaSparkProjectSettings(),
+    crossSparkUniDocSettings(),
     libraryDependencies ++= Seq(
       // Adding test classifier seems to break transitive resolution of the core dependencies
       "org.apache.spark" %% "spark-hive" % sparkVersion.value % "provided",

kernel/kernel-api/src/main/java/io/delta/kernel/Scan.java

@@ -205,10 +205,12 @@ public FilteredColumnarBatch next() {
 
                 // Transform physical variant columns (struct of binaries) into logical variant
                 // columns.
-                nextDataBatch = VariantUtils.withVariantColumns(
-                    engine.getExpressionHandler(),
-                    nextDataBatch
-                );
+                if (ScanStateRow.getVariantFeatureEnabled(scanState)) {
+                    nextDataBatch = VariantUtils.withVariantColumns(
+                        engine.getExpressionHandler(),
+                        nextDataBatch
+                    );
+                }
 
                 // Add partition columns
                 nextDataBatch =

kernel/kernel-api/src/main/java/io/delta/kernel/data/ColumnarBatch.java

@@ -104,6 +104,22 @@ default ColumnarBatch slice(int start, int end) {
         throw new UnsupportedOperationException("Not yet implemented!");
     }
 
+    /**
+     * Return a copy of this {@link ColumnarBatch} with the column at given {@code ordinal}
+     * replaced with {@code newVector} and the schema field at given {@code ordinal} replaced
+     * with {@code newColumnSchema}.
+     *
+     * @param ordinal Ordinal of the column vector to replace.
+     * @param newColumnSchema The schema field of the new column.
+     * @param newVector New column vector that will replace the column vector at the given
+     *                  {@code ordinal}.
+     * @return {@link ColumnarBatch} with a new column vector at the given ordinal.
+     */
+    default ColumnarBatch withReplacedColumnVector(int ordinal, StructField newColumnSchema,
+                                                   ColumnVector newVector) {
+        throw new UnsupportedOperationException("Not yet implemented!");
+    }
+
     /**
      * @return iterator of {@link Row}s in this batch
      */

kernel/kernel-api/src/main/java/io/delta/kernel/internal/data/ScanStateRow.java

@@ -42,6 +42,7 @@ public class ScanStateRow extends GenericRow {
         .add("partitionColumns", new ArrayType(StringType.STRING, false))
         .add("minReaderVersion", IntegerType.INTEGER)
         .add("minWriterVersion", IntegerType.INTEGER)
+        .add("variantFeatureEnabled", BooleanType.BOOLEAN)
         .add("tablePath", StringType.STRING);
 
     private static final Map<String, Integer> COL_NAME_TO_ORDINAL =
@@ -65,6 +66,10 @@ public static ScanStateRow of(
         valueMap.put(COL_NAME_TO_ORDINAL.get("partitionColumns"), metadata.getPartitionColumns());
         valueMap.put(COL_NAME_TO_ORDINAL.get("minReaderVersion"), protocol.getMinReaderVersion());
         valueMap.put(COL_NAME_TO_ORDINAL.get("minWriterVersion"), protocol.getMinWriterVersion());
+        valueMap.put(
+            COL_NAME_TO_ORDINAL.get("variantFeatureEnabled"),
+            protocol.getReaderFeatures().contains("variantType-preview")
+        );
         valueMap.put(COL_NAME_TO_ORDINAL.get("tablePath"), tablePath);
         return new ScanStateRow(valueMap);
     }
@@ -156,4 +161,15 @@ private static StructType parseSchema(Engine engine, String serializedSchema) {
             serializedSchema
         );
     }
+
+     /**
+     * Get whether the "variantType" table feature is enabled from scan state {@link Row} returned
+     * by {@link Scan#getScanState(Engine)}
+     *
+     * @param scanState Scan state {@link Row}
+     * @return Boolean indicating whether "variantType" is enabled.
+     */
+    public static Boolean getVariantFeatureEnabled(Row scanState) {
+        return scanState.getBoolean(COL_NAME_TO_ORDINAL.get("variantFeatureEnabled"));
+    }
 }

kernel/kernel-api/src/main/java/io/delta/kernel/internal/util/VariantUtils.java

@@ -40,7 +40,7 @@ public static ColumnarBatch withVariantColumns(
 
             ExpressionEvaluator evaluator = expressionHandler.getEvaluator(
                 // Field here is variant type if its actually a variant.
-                // TODO: probably better to pass in the schema as an argument
+                // TODO: probably better to pass in the schema as an expression argument
                 // so the schema is enforced at the expression level. Need to pass in a literal
                 // schema
                 new StructType().add(field),
@@ -51,11 +51,8 @@ public static ColumnarBatch withVariantColumns(
                 VariantType.VARIANT
             );
 
-            // TODO: don't need to pass in the entire batch.
             ColumnVector variantCol = evaluator.eval(dataBatch);
-            // TODO: make a more efficient way to do this.
-            dataBatch =
-                dataBatch.withDeletedColumnAt(i).withNewColumn(i, field, variantCol);
+            dataBatch = dataBatch.withReplacedColumnVector(i, field, variantCol);
         }
         return dataBatch;
     }

kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/data/DefaultColumnarBatch.java

@@ -109,6 +109,19 @@ public ColumnarBatch withNewSchema(StructType newSchema) {
             size, newSchema, columnVectors.toArray(new ColumnVector[0]));
     }
 
+    @Override
+    public ColumnarBatch withReplacedColumnVector(int ordinal, StructField newColumnSchema,
+                                                   ColumnVector newVector) {
+        ArrayList<StructField> newStructFields = new ArrayList<>(schema.fields());
+        newStructFields.set(ordinal, newColumnSchema);
+        StructType newSchema = new StructType(newStructFields);
+
+        ArrayList<ColumnVector> newColumnVectors = new ArrayList<>(columnVectors);
+        newColumnVectors.set(ordinal, newVector);
+        return new DefaultColumnarBatch(
+            size, newSchema, newColumnVectors.toArray(new ColumnVector[0]));
+    }
+
     @Override
     public int getSize() {
         return size;

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/internal/parquet/ParquetFileReaderSuite.scala

@@ -16,7 +16,6 @@
 package io.delta.kernel.defaults.internal.parquet
 
 import java.math.BigDecimal
-
 import io.delta.golden.GoldenTableUtils.goldenTableFile
 import io.delta.kernel.defaults.utils.{ExpressionTestUtils, TestRow}
 import io.delta.kernel.test.VectorTestUtils