Done refactor

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

@@ -19,11 +19,9 @@
 package io.delta.standalone.internal;
 
 import java.lang.reflect.Constructor;
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 
-import io.delta.kernel.data.ColumnVector;
 import io.delta.kernel.internal.types.DataTypeJsonSerDe;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -140,11 +138,7 @@ private Metadata convertMetadata() {
         );
 
         // Convert the partition columns from a ColumnVector to a List<String>
-        ColumnVector partitionsVec = kernelMetadata.getPartitionColumns().getElements();
-        ArrayList<String> partitionColumns = new ArrayList<String>(partitionsVec.getSize());
-        for(int i = 0; i < partitionsVec.getSize(); i++) {
-            partitionColumns.add(partitionsVec.getString(i));
-        }
+        final List<String> partitionColumns = kernelMetadata.getPartitionColumns();
 
         // Convert over the schema StructType
         List<io.delta.kernel.types.StructField> kernelFields = kernelMetadata.getSchema().fields();

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

@@ -0,0 +1,60 @@
+/*
+ * 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.actions;
+
+import io.delta.kernel.annotation.Evolving;
+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.
+ *
+ * @since 3.4.0
+ */
+@Evolving
+public interface AbstractMetadata {
+
+  /** A unique table identifier. */
+  String getId();
+
+  /** User-specified table identifier. */
+  // TODO: this should be Optional<String>. Delta protocol defines 'name' as optional.
+  String getName();
+
+  /** User-specified table description. */
+  // TODO: this should be Optional<String>. Delta protocol defines 'description' as optional.
+  String getDescription();
+
+  /** The table provider format. */
+  String getProvider();
+
+  /** The format options */
+  Map<String, String> getFormatOptions();
+
+  /** The table schema in string representation. */
+  String getSchemaString();
+
+  /** List of partition columns. */
+  List<String> getPartitionColumns();
+
+  /** The table properties defined on the table. */
+  Map<String, String> getConfiguration();
+
+  /** Timestamp for the creation of this metadata. */
+  Optional<Long> getCreatedTime();
+}

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

@@ -77,7 +77,7 @@ public ScanImpl(
     this.dataPath = dataPath;
     this.partitionColToStructFieldMap =
         () -> {
-          Set<String> partitionColNames = metadata.getPartitionColNames();
+          Set<String> partitionColNames = metadata.getPartitionColumnsLowercase();
           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 +156,7 @@ 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 = metadata.getPartitionColumns();
     StructType physicalDataReadSchema =
         PartitionUtils.physicalSchemaWithoutPartitionColumns(
             readSchema, /* logical read schema */
@@ -185,7 +185,7 @@ private Optional<Tuple2<Predicate, Predicate>> splitFilters(Optional<Predicate>
     return filter.map(
         predicate ->
             PartitionUtils.splitMetadataAndDataPredicates(
-                predicate, metadata.getPartitionColNames()));
+                predicate, metadata.getPartitionColumnsLowercase()));
   }
 
   private Optional<Predicate> getDataFilters() {

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

@@ -31,7 +31,6 @@
 import io.delta.kernel.internal.replay.CreateCheckpointIterator;
 import io.delta.kernel.internal.replay.LogReplay;
 import io.delta.kernel.internal.snapshot.LogSegment;
-import io.delta.kernel.internal.util.VectorUtils;
 import io.delta.kernel.metrics.SnapshotReport;
 import io.delta.kernel.types.StructType;
 import java.util.List;
@@ -134,7 +133,7 @@ public Protocol getProtocol() {
   }
 
   public List<String> getPartitionColumnNames(Engine engine) {
-    return VectorUtils.toJavaList(getMetadata().getPartitionColumns());
+    return getMetadata().getPartitionColumns();
   }
 
   public SnapshotReport getSnapshotReport() {

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

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