[Kernel] [CC Refactor] Make Metadata implement AbstractMetadata; fix Metadata partition col APIs

connectors/flink/src/main/scala/io/delta/flink/internal/KernelSnapshotWrapper.java

@@ -140,7 +140,7 @@ private Metadata convertMetadata() {
         );
 
         // Convert the partition columns from a ColumnVector to a List<String>
-        ColumnVector partitionsVec = kernelMetadata.getPartitionColumns().getElements();
+        ColumnVector partitionsVec = kernelMetadata.getPartitionColumnsArrayValue().getElements();
         ArrayList<String> partitionColumns = new ArrayList<String>(partitionsVec.getSize());
         for(int i = 0; i < partitionsVec.getSize(); i++) {
             partitionColumns.add(partitionsVec.getString(i));
@@ -186,12 +186,12 @@ private Metadata convertMetadata() {
 
         return new Metadata(
             kernelMetadata.getId(),
-            kernelMetadata.getName().orElse(null),
-            kernelMetadata.getDescription().orElse(null),
+            kernelMetadata.getName(),
+            kernelMetadata.getDescription(),
             format,
             partitionColumns,
             kernelMetadata.getConfiguration(),
-            kernelMetadata.getCreatedTime(),
+            Optional.ofNullable(kernelMetadata.getCreatedTime()),
             schema
         );
     }

kernel/kernel-api/src/main/java/io/delta/kernel/annotation/Nullable.java

@@ -0,0 +1,27 @@
+/*
+ * 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.annotation;
+
+import java.lang.annotation.*;
+
+/**
+ * Annotation to indicate that a field, method parameter, method return value, or local variable may
+ * be null.
+ */
+@Documented
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.FIELD, ElementType.METHOD, ElementType.PARAMETER, ElementType.LOCAL_VARIABLE})
+public @interface Nullable {}

kernel/kernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/actions/AbstractMetadata.java

@@ -17,9 +17,9 @@
 package io.delta.kernel.engine.coordinatedcommits.actions;
 
 import io.delta.kernel.annotation.Evolving;
+import io.delta.kernel.annotation.Nullable;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 
 /**
  * Interface for metadata actions in Delta. The metadata defines the metadata of the table.
@@ -33,9 +33,11 @@ public interface AbstractMetadata {
   String getId();
 
   /** User-specified table identifier. */
+  @Nullable
   String getName();
 
   /** User-specified table description. */
+  @Nullable
   String getDescription();
 
   /** The table provider format. */
@@ -54,5 +56,6 @@ public interface AbstractMetadata {
   Map<String, String> getConfiguration();
 
   /** Timestamp for the creation of this metadata. */
-  Optional<Long> getCreatedTime();
+  @Nullable
+  Long getCreatedTime();
 }

kernel/kernel-api/src/main/java/io/delta/kernel/internal/ScanImpl.java

@@ -77,7 +77,8 @@ public ScanImpl(
     this.dataPath = dataPath;
     this.partitionColToStructFieldMap =
         () -> {
-          Set<String> partitionColNames = metadata.getPartitionColNames();
+          final Set<String> partitionColNames =
+              PartitionUtils.arrayValueToLowerCaseSet(metadata.getPartitionColumnsArrayValue());
           return metadata.getSchema().fields().stream()
               .filter(field -> partitionColNames.contains(field.getName().toLowerCase(Locale.ROOT)))
               .collect(toMap(field -> field.getName().toLowerCase(Locale.ROOT), identity()));
@@ -156,7 +157,8 @@ public Row getScanState(Engine engine) {
     // Compute the physical data read schema, basically the list of columns to read
     // from a Parquet data file. It should exclude partition columns and include
     // row_index metadata columns (in case DVs are present)
-    List<String> partitionColumns = VectorUtils.toJavaList(metadata.getPartitionColumns());
+    List<String> partitionColumns =
+        VectorUtils.toJavaList(metadata.getPartitionColumnsArrayValue());
     StructType physicalDataReadSchema =
         PartitionUtils.physicalSchemaWithoutPartitionColumns(
             readSchema, /* logical read schema */
@@ -185,7 +187,8 @@ private Optional<Tuple2<Predicate, Predicate>> splitFilters(Optional<Predicate>
     return filter.map(
         predicate ->
             PartitionUtils.splitMetadataAndDataPredicates(
-                predicate, metadata.getPartitionColNames()));
+                predicate,
+                PartitionUtils.arrayValueToLowerCaseSet(metadata.getPartitionColumnsArrayValue())));
   }
 
   private Optional<Predicate> getDataFilters() {

kernel/kernel-api/src/main/java/io/delta/kernel/internal/TransactionImpl.java

@@ -112,7 +112,7 @@ public Row getTransactionState(Engine engine) {
 
   @Override
   public List<String> getPartitionColumns(Engine engine) {
-    return VectorUtils.toJavaList(metadata.getPartitionColumns());
+    return VectorUtils.toJavaList(metadata.getPartitionColumnsArrayValue());
   }
 
   @Override
@@ -303,7 +303,7 @@ private boolean isReadyForCheckpoint(long newVersion) {
 
   private Map<String, String> getOperationParameters() {
     if (isNewTable) {
-      List<String> partitionCols = VectorUtils.toJavaList(metadata.getPartitionColumns());
+      List<String> partitionCols = VectorUtils.toJavaList(metadata.getPartitionColumnsArrayValue());
       String partitionBy =
           partitionCols.stream()
               .map(col -> "\"" + col + "\"")