init no tests yet

richardc-db · richardc-db · commit 62f310c086f9 · 2024-03-11T23:14:53.000-07:00
diff --git a/kernel/kernel-api/src/main/java/io/delta/kernel/data/ColumnVector.java b/kernel/kernel-api/src/main/java/io/delta/kernel/data/ColumnVector.java
@@ -175,6 +175,14 @@ default ArrayValue getArray(int rowId) {
         throw new UnsupportedOperationException("Invalid value request for data type");
     }
 
+    /**
+     * Return the variant value located at {@code rowId}. Returns null if the slot for {@code rowId}
+     * is null
+     */
+    default VariantValue getVariant(int rowId) {
+        throw new UnsupportedOperationException("Invalid value request for data type");
+    }
+
     /**
      * Get the child vector associated with the given ordinal. This method is applicable only to the
      * {@code struct} type columns.
diff --git a/kernel/kernel-api/src/main/java/io/delta/kernel/data/Row.java b/kernel/kernel-api/src/main/java/io/delta/kernel/data/Row.java
@@ -117,4 +117,10 @@ public interface Row {
      * Throws error if the column at given ordinal is not of map type,
      */
     MapValue getMap(int ordinal);
+
+    /**
+     * Return variant value of the column located at the given ordinal.
+     * Throws error if the column at given ordinal is not of variant type.
+     */
+    VariantValue getVariant(int ordinal);
 }
diff --git a/kernel/kernel-api/src/main/java/io/delta/kernel/data/VariantValue.java b/kernel/kernel-api/src/main/java/io/delta/kernel/data/VariantValue.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright (2024) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.kernel.data;
+
+/**
+ * Abstraction to represent a single Variant value in a {@link ColumnVector}.
+ */
+public interface VariantValue {
+    byte[] getValue();
+
+    byte[] getMetadata();
+}
diff --git a/kernel/kernel-api/src/main/java/io/delta/kernel/internal/data/ChildVectorBasedRow.java b/kernel/kernel-api/src/main/java/io/delta/kernel/internal/data/ChildVectorBasedRow.java
@@ -21,6 +21,7 @@
 import io.delta.kernel.data.ColumnVector;
 import io.delta.kernel.data.MapValue;
 import io.delta.kernel.data.Row;
+import io.delta.kernel.data.VariantValue;
 import io.delta.kernel.types.StructType;
 
 /**
@@ -111,5 +112,10 @@ public MapValue getMap(int ordinal) {
         return getChild(ordinal).getMap(rowId);
     }
 
+    @Override
+    public VariantValue getVariant(int ordinal) {
+        return getChild(ordinal).getVariant(rowId);
+    }
+
     protected abstract ColumnVector getChild(int ordinal);
 }
diff --git a/kernel/kernel-api/src/main/java/io/delta/kernel/internal/data/GenericRow.java b/kernel/kernel-api/src/main/java/io/delta/kernel/internal/data/GenericRow.java
@@ -23,6 +23,7 @@
 import io.delta.kernel.data.ArrayValue;
 import io.delta.kernel.data.MapValue;
 import io.delta.kernel.data.Row;
+import io.delta.kernel.data.VariantValue;
 import io.delta.kernel.types.*;
 
 /**
@@ -134,6 +135,13 @@ public MapValue getMap(int ordinal) {
         return (MapValue) getValue(ordinal);
     }
 
+    @Override
+    public VariantValue getVariant(int ordinal) {
+        // TODO(r.chen): test this path somehow?
+        throwIfUnsafeAccess(ordinal, VariantType.class, "variant");
+        return (VariantValue) getValue(ordinal);
+    }
+
     private Object getValue(int ordinal) {
         return ordinalToValue.get(ordinal);
     }
diff --git a/kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/LogReplay.java b/kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/LogReplay.java
@@ -313,6 +313,8 @@ private void validateSupportedTable(Protocol protocol, Metadata metadata) {
                         case "columnMapping":
                             ColumnMapping.throwOnUnsupportedColumnMappingMode(metadata);
                             break;
+                        case "variantType-dev":
+                            break;
                         default:
                             throw new UnsupportedOperationException(
                                 "Unsupported table feature: " + readerFeature);
diff --git a/kernel/kernel-api/src/main/java/io/delta/kernel/internal/util/VectorUtils.java b/kernel/kernel-api/src/main/java/io/delta/kernel/internal/util/VectorUtils.java
@@ -110,6 +110,8 @@ private static Object getValueAsObject(
             return toJavaList(columnVector.getArray(rowId));
         } else if (dataType instanceof MapType) {
             return toJavaMap(columnVector.getMap(rowId));
+        } else if (dataType instanceof VariantType) {
+            return columnVector.getVariant(rowId);
         } else {
             throw new UnsupportedOperationException("unsupported data type");
         }
diff --git a/kernel/kernel-api/src/main/java/io/delta/kernel/types/BasePrimitiveType.java b/kernel/kernel-api/src/main/java/io/delta/kernel/types/BasePrimitiveType.java
@@ -63,6 +63,7 @@ public static List<DataType> getAllPrimitiveTypes() {
                 put("timestamp", TimestampType.TIMESTAMP);
                 put("binary", BinaryType.BINARY);
                 put("string", StringType.STRING);
+                put("variant", VariantType.VARIANT);
             }
         });
 
diff --git a/kernel/kernel-api/src/main/java/io/delta/kernel/types/VariantType.java b/kernel/kernel-api/src/main/java/io/delta/kernel/types/VariantType.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright (2024) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.kernel.types;
+
+import io.delta.kernel.annotation.Evolving;
+
+/**
+ * A variant type.
+ * <p>
+ * todo: more comments
+ * @since 4.0.0
+ */
+@Evolving
+public class VariantType extends BasePrimitiveType {
+    public static final VariantType VARIANT = new VariantType();
+
+    private VariantType() {
+        super("variant");
+    }
+}
diff --git a/kernel/kernel-api/src/test/scala/io/delta/kernel/internal/SnapshotManagerSuite.scala b/kernel/kernel-api/src/test/scala/io/delta/kernel/internal/SnapshotManagerSuite.scala
@@ -618,7 +618,7 @@ class SnapshotManagerSuite extends AnyFunSuite with MockFileSystemClientUtils {
     // corrupt incomplete multi-part checkpoint
     val corruptedCheckpointStatuses = FileNames.checkpointFileWithParts(logPath, 10, 5).asScala
       .map(p => FileStatus.of(p.toString, 10, 10))
-      .take(4)
+      .take(4).toSeq
     val deltas = deltaFileStatuses(10L to 13L)
     testExpectedError[RuntimeException](
       corruptedCheckpointStatuses ++ deltas,
@@ -667,7 +667,7 @@ class SnapshotManagerSuite extends AnyFunSuite with MockFileSystemClientUtils {
     // _last_checkpoint refers to incomplete multi-part checkpoint
     val corruptedCheckpointStatuses = FileNames.checkpointFileWithParts(logPath, 20, 5).asScala
       .map(p => FileStatus.of(p.toString, 10, 10))
-      .take(4)
+      .take(4).toSeq
     testExpectedError[RuntimeException](
       files = corruptedCheckpointStatuses ++ deltaFileStatuses(10L to 20L) ++
         singularCheckpointFileStatuses(Seq(10L)),
diff --git a/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/data/DefaultJsonRow.java b/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/data/DefaultJsonRow.java
@@ -33,6 +33,7 @@
 import io.delta.kernel.data.ColumnVector;
 import io.delta.kernel.data.MapValue;
 import io.delta.kernel.data.Row;
+import io.delta.kernel.data.VariantValue;
 import io.delta.kernel.types.*;
 import io.delta.kernel.internal.util.InternalUtils;
 
@@ -128,6 +129,11 @@ public MapValue getMap(int ordinal) {
         return (MapValue) parsedValues[ordinal];
     }
 
+    @Override
+    public VariantValue getVariant(int ordinal) {
+        throw new UnsupportedOperationException("not yet implemented");
+    }
+
     private static void throwIfTypeMismatch(String expType, boolean hasExpType, JsonNode jsonNode) {
         if (!hasExpType) {
             throw new RuntimeException(
diff --git a/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/data/vector/AbstractColumnVector.java b/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/data/vector/AbstractColumnVector.java
@@ -22,6 +22,7 @@
 import io.delta.kernel.data.ArrayValue;
 import io.delta.kernel.data.ColumnVector;
 import io.delta.kernel.data.MapValue;
+import io.delta.kernel.data.VariantValue;
 import io.delta.kernel.types.DataType;
 
 import static io.delta.kernel.internal.util.Preconditions.checkArgument;
@@ -138,6 +139,11 @@ public ArrayValue getArray(int rowId) {
         throw unsupportedDataAccessException("array");
     }
 
+    @Override
+    public VariantValue getVariant(int rowId) {
+        throw unsupportedDataAccessException("variant");
+    }
+
     // TODO no need to override these here; update default implementations in `ColumnVector`
     //   to have a more informative exception message
     protected UnsupportedOperationException unsupportedDataAccessException(String accessType) {
diff --git a/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/data/vector/DefaultGenericVector.java b/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/data/vector/DefaultGenericVector.java
@@ -23,6 +23,7 @@
 import io.delta.kernel.data.ColumnVector;
 import io.delta.kernel.data.MapValue;
 import io.delta.kernel.data.Row;
+import io.delta.kernel.data.VariantValue;
 import io.delta.kernel.types.*;
 import static io.delta.kernel.internal.util.Preconditions.checkArgument;
 
@@ -170,6 +171,13 @@ public ColumnVector getChild(int ordinal) {
                 (rowId) -> (Row) rowIdToValueAccessor.apply(rowId));
     }
 
+    @Override
+    public VariantValue getVariant(int rowId) {
+        assertValidRowId(rowId);
+        throwIfUnsafeAccess(VariantType.class, "variant");
+        return (VariantValue) rowIdToValueAccessor.apply(rowId);
+    }
+
     private void throwIfUnsafeAccess( Class<? extends DataType> expDataType, String accessType) {
         if (!expDataType.isAssignableFrom(dataType.getClass())) {
             String msg = String.format(
diff --git a/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/data/vector/DefaultSubFieldVector.java b/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/data/vector/DefaultSubFieldVector.java
@@ -23,6 +23,7 @@
 import io.delta.kernel.data.ColumnVector;
 import io.delta.kernel.data.MapValue;
 import io.delta.kernel.data.Row;
+import io.delta.kernel.data.VariantValue;
 import io.delta.kernel.types.DataType;
 import io.delta.kernel.types.StructField;
 import io.delta.kernel.types.StructType;
@@ -155,6 +156,12 @@ public ArrayValue getArray(int rowId) {
         return rowIdToRowAccessor.apply(rowId).getArray(columnOrdinal);
     }
 
+    @Override
+    public VariantValue getVariant(int rowId) {
+        assertValidRowId(rowId);
+        return rowIdToRowAccessor.apply(rowId).getVariant(columnOrdinal);
+    }
+
     @Override
     public ColumnVector getChild(int childOrdinal) {
         StructType structType = (StructType) dataType;
diff --git a/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/data/vector/DefaultVariantVector.java b/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/data/vector/DefaultVariantVector.java
@@ -0,0 +1,93 @@
+/*
+ * Copyright (2023) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.kernel.defaults.internal.data.vector;
+
+import java.util.Optional;
+import static java.util.Objects.requireNonNull;
+
+import io.delta.kernel.data.ColumnVector;
+import io.delta.kernel.data.VariantValue;
+import io.delta.kernel.types.DataType;
+
+/**
+ * {@link io.delta.kernel.data.ColumnVector} implementation for variant type data.
+ */
+public class DefaultVariantVector
+    extends AbstractColumnVector {
+    private final ColumnVector valueVector;
+    private final ColumnVector metadataVector;
+
+
+    /**
+     * Create an instance of {@link io.delta.kernel.data.ColumnVector} for array type.
+     *
+     * @param size          number of elements in the vector.
+     * @param nullability   Optional array of nullability value for each element in the vector.
+     *                      All values in the vector are considered non-null when parameter is
+     *                      empty.
+     * @param offsets       Offsets into element vector on where the index of particular row
+     *                      values start and end.
+     * @param elementVector Vector containing the array elements.
+     */
+    public DefaultVariantVector(
+        int size,
+        DataType type,
+        Optional<boolean[]> nullability,
+        ColumnVector value,
+        ColumnVector metadata) {
+        super(size, type, nullability);
+        // checkArgument(offsets.length >= size + 1, "invalid offset array size");
+        this.valueVector = requireNonNull(value, "value is null");
+        this.metadataVector = requireNonNull(metadata, "metadata is null");
+    }
+
+    /**
+     * Get the value at given {@code rowId}. The return value is undefined and can be
+     * anything, if the slot for {@code rowId} is null.
+     *
+     * @param rowId
+     * @return
+     */
+    @Override
+    public VariantValue getVariant(int rowId) {
+        checkValidRowId(rowId);
+        if (isNullAt(rowId)) {
+            return null;
+        }
+        return new VariantValue() {
+            private final byte[] value = valueVector.getBinary(rowId);
+            private final byte[] metadata = metadataVector.getBinary(rowId);
+
+            @Override
+            public byte[] getValue() {
+                return value;
+            }
+
+            @Override
+            public byte[] getMetadata() {
+                return metadata;
+            }
+        };
+    }
+
+    public ColumnVector getValueVector() {
+        return valueVector;
+    }
+
+    public ColumnVector getMetadataVector() {
+        return metadataVector;
+    }
+}
diff --git a/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/data/vector/DefaultViewVector.java b/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/data/vector/DefaultViewVector.java
@@ -20,6 +20,7 @@
 import io.delta.kernel.data.ArrayValue;
 import io.delta.kernel.data.ColumnVector;
 import io.delta.kernel.data.MapValue;
+import io.delta.kernel.data.VariantValue;
 import io.delta.kernel.types.DataType;
 import static io.delta.kernel.internal.util.Preconditions.checkArgument;
 
@@ -137,6 +138,12 @@ public ArrayValue getArray(int rowId) {
         return underlyingVector.getArray(offset + rowId);
     }
 
+    @Override
+    public VariantValue getVariant(int rowId) {
+        checkValidRowId(rowId);
+        return underlyingVector.getVariant(offset + rowId);
+    }
+
     @Override
     public ColumnVector getChild(int ordinal) {
         return new DefaultViewVector(underlyingVector.getChild(ordinal), offset, offset + size);
diff --git a/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/parquet/ParquetConverters.java b/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/parquet/ParquetConverters.java
@@ -74,6 +74,8 @@ public static Converter createConverter(
                 initialBatchSize, (DecimalType) typeFromClient, typeFromFile);
         } else if (typeFromClient instanceof TimestampType) {
             return TimestampConverters.createTimestampConverter(initialBatchSize, typeFromFile);
+        } else if (typeFromClient instanceof VariantType) {
+            return new VariantConverter(initialBatchSize);
         }
 
         throw new UnsupportedOperationException(typeFromClient + " is not supported");
diff --git a/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/parquet/ParquetSchemaUtils.java b/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/parquet/ParquetSchemaUtils.java
diff --git a/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/parquet/VariantConverter.java b/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/parquet/VariantConverter.java
diff --git a/kernel/kernel-defaults/src/test/java/io/delta/kernel/defaults/integration/DataBuilderUtils.java b/kernel/kernel-defaults/src/test/java/io/delta/kernel/defaults/integration/DataBuilderUtils.java
diff --git a/kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableReadsSuite.scala b/kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableReadsSuite.scala
diff --git a/kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/LogReplayMetricsSuite.scala b/kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/LogReplayMetricsSuite.scala
diff --git a/kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/utils/TestRow.scala b/kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/utils/TestRow.scala
diff --git a/kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/utils/TestUtils.scala b/kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/utils/TestUtils.scala

Original file line number	Diff line number	Diff line change
`@@ -110,6 +110,8 @@ private static Object getValueAsObject(`
`110`	`110`	`return toJavaList(columnVector.getArray(rowId));`
`111`	`111`	`} else if (dataType instanceof MapType) {`
`112`	`112`	`return toJavaMap(columnVector.getMap(rowId));`
	`113`	`+ } else if (dataType instanceof VariantType) {`
	`114`	`+ return columnVector.getVariant(rowId);`
`113`	`115`	`} else {`
`114`	`116`	`throw new UnsupportedOperationException("unsupported data type");`
`115`	`117`	`}`
Original file line number	Diff line number	Diff line change
`@@ -63,6 +63,7 @@ public static List<DataType> getAllPrimitiveTypes() {`
`63`	`63`	`put("timestamp", TimestampType.TIMESTAMP);`
`64`	`64`	`put("binary", BinaryType.BINARY);`
`65`	`65`	`put("string", StringType.STRING);`
	`66`	`+ put("variant", VariantType.VARIANT);`
`66`	`67`	`}`
`67`	`68`	`});`
`68`	`69`