Delta Kernel Draft PR

pom.xml

@@ -53,7 +53,7 @@
         <module>xtable-utilities</module>
         <module>xtable-aws</module>
         <module>xtable-hive-metastore</module>
-        <module>xtable-service</module>
+<!--        <module>xtable-service</module>-->
     </modules>
 
     <properties>

xtable-core/pom.xml

@@ -110,6 +110,19 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>io.delta</groupId>
+            <artifactId>delta-kernel-api</artifactId>
+            <version>4.0.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>io.delta</groupId>
+            <artifactId>delta-kernel-defaults</artifactId>
+            <version>4.0.0</version>
+        </dependency>
+
+
         <!-- Hadoop dependencies -->
         <dependency>
             <groupId>org.apache.hadoop</groupId>

xtable-core/src/main/java/org/apache/xtable/delta/DeltaKernelConversionSourceProvider.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.xtable.delta;
+
+import org.apache.hadoop.conf.Configuration;
+
+import io.delta.kernel.defaults.engine.DefaultEngine;
+import io.delta.kernel.engine.Engine;
+
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.kernel.DeltaKernelConversionSource;
+
+public class DeltaKernelConversionSourceProvider extends ConversionSourceProvider<Long> {
+  @Override
+  public DeltaKernelConversionSource getConversionSourceInstance(SourceTable sourceTable) {
+    Configuration hadoopConf = new Configuration();
+    Engine engine = DefaultEngine.create(hadoopConf);
+    //    DeltaTable deltaTable = DeltaT/able.forPath(sourceTable.getBasePath());
+    return DeltaKernelConversionSource.builder()
+        .tableName(sourceTable.getName())
+        .basePath(sourceTable.getBasePath())
+        .engine(engine)
+        .build();
+  }
+}

xtable-core/src/main/java/org/apache/xtable/delta/DeltaKernelSchemaExtractor.java

@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.xtable.delta;
+
+import java.util.*;
+
+import io.delta.kernel.types.DataType;
+import io.delta.kernel.types.IntegerType;
+import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.StructType;
+
+import org.apache.xtable.collectors.CustomCollectors;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+public class DeltaKernelSchemaExtractor {
+
+  private static final String DELTA_COLUMN_MAPPING_ID = "delta.columnMapping.id";
+  private static final DeltaKernelSchemaExtractor INSTANCE = new DeltaKernelSchemaExtractor();
+  private static final Map<InternalSchema.MetadataKey, Object>
+      DEFAULT_TIMESTAMP_PRECISION_METADATA =
+          Collections.singletonMap(
+              InternalSchema.MetadataKey.TIMESTAMP_PRECISION, InternalSchema.MetadataValue.MICROS);
+
+  public static DeltaKernelSchemaExtractor getInstance() {
+    return INSTANCE;
+  }
+
+  public InternalSchema toInternalSchema_v2(StructType structType) {
+    return toInternalSchema_v2(structType, null, false, null);
+  }
+
+  String trimmedTypeName = "";
+
+  private InternalSchema toInternalSchema_v2(
+      DataType dataType, String parentPath, boolean nullable, String comment) {
+
+    Map<InternalSchema.MetadataKey, Object> metadata = null;
+    List<InternalField> fields = null;
+    InternalType type = null;
+    if (dataType instanceof IntegerType) {
+      type = InternalType.INT;
+      trimmedTypeName = "integer";
+    }
+    if (dataType instanceof StringType) {
+      type = InternalType.STRING;
+      trimmedTypeName = "string";
+    }
+    if (dataType instanceof StructType) {
+      // Handle StructType
+      StructType structType = (StructType) dataType;
+      // your logic here
+
+      fields =
+          structType.fields().stream()
+              .filter(
+                  field ->
+                      !field
+                          .getMetadata()
+                          .contains(DeltaPartitionExtractor.DELTA_GENERATION_EXPRESSION))
+              .map(
+                  field -> {
+                    Integer fieldId =
+                        field.getMetadata().contains(DELTA_COLUMN_MAPPING_ID)
+                            ? Long.valueOf(field.getMetadata().getLong(DELTA_COLUMN_MAPPING_ID))
+                                .intValue()
+                            : null;
+                    String fieldComment =
+                        field.getMetadata().contains("comment")
+                            ? field.getMetadata().getString("comment")
+                            : null;
+                    InternalSchema schema =
+                        toInternalSchema_v2(
+                            field.getDataType(),
+                            SchemaUtils.getFullyQualifiedPath(parentPath, field.getName()),
+                            field.isNullable(),
+                            fieldComment);
+                    return InternalField.builder()
+                        .name(field.getName())
+                        .fieldId(fieldId)
+                        .parentPath(parentPath)
+                        .schema(schema)
+                        .defaultValue(
+                            field.isNullable() ? InternalField.Constants.NULL_DEFAULT_VALUE : null)
+                        .build();
+                  })
+              .collect(CustomCollectors.toList(structType.fields().size()));
+      type = InternalType.RECORD;
+      trimmedTypeName = "struct";
+    }
+
+    return InternalSchema.builder()
+        .name(trimmedTypeName)
+        .dataType(type)
+        .comment(comment)
+        .isNullable(nullable)
+        .metadata(metadata)
+        .fields(fields)
+        .build();
+  }
+}

xtable-core/src/main/java/org/apache/xtable/delta/DeltaKernelTableExtractor.java

@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.xtable.delta;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+
+import lombok.Builder;
+
+import io.delta.kernel.*;
+import io.delta.kernel.engine.Engine;
+
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.model.storage.DataLayoutStrategy;
+import org.apache.xtable.model.storage.TableFormat;
+
+/**
+ * Extracts {@link InternalTable} canonical representation of a table at a point in time for Delta.
+ */
+@Builder
+public class DeltaKernelTableExtractor {
+  @Builder.Default
+  private static final DeltaKernelSchemaExtractor schemaExtractor =
+      DeltaKernelSchemaExtractor.getInstance();
+
+  private final String basePath;
+
+  public InternalTable table(
+      Table deltaKernelTable, Snapshot snapshot, Engine engine, String tableName, String basePath) {
+    try {
+      // Get schema from Delta Kernel's snapshot
+      io.delta.kernel.types.StructType schema = snapshot.getSchema();
+
+      System.out.println("Kernelschema: " + schema);
+
+      InternalSchema internalSchema = schemaExtractor.toInternalSchema_v2(schema);
+      //      io.delta.kernel.types.StructType schema = snapshot.getSchema();
+      ////      InternalSchema internalSchema = schemaExtractor.toInternalSchema_v2(schema);
+      //      InternalSchema internalSchema =
+      // schemaExtractor.toInternalSchema(snapshot.getSchema());
+
+      // Get partition columns
+      System.out.println("Partition columns: " + internalSchema);
+      List<String> partitionColumnNames = snapshot.getPartitionColumnNames();
+      List<InternalPartitionField> partitionFields = new ArrayList<>();
+      for (String columnName : partitionColumnNames) {
+        InternalField sourceField =
+            InternalField.builder()
+                .name(columnName)
+                .schema(
+                    InternalSchema.builder()
+                        .name(columnName)
+                        .dataType(InternalType.STRING) // Assuming string type for partition columns
+                        .build())
+                .build();
+
+        // Create the partition field with the source field
+        partitionFields.add(InternalPartitionField.builder().sourceField(sourceField).build());
+      }
+
+      DataLayoutStrategy dataLayoutStrategy =
+          partitionFields.isEmpty()
+              ? DataLayoutStrategy.FLAT
+              : DataLayoutStrategy.HIVE_STYLE_PARTITION;
+
+      // Get the timestamp
+      long timestamp = snapshot.getTimestamp(engine) * 1000; // Convert to milliseconds
+      System.out.println("InternalTable basepath" + basePath);
+      return InternalTable.builder()
+          .tableFormat(TableFormat.DELTA)
+          .basePath(basePath)
+          .name(tableName)
+          .layoutStrategy(dataLayoutStrategy)
+          .partitioningFields(partitionFields)
+          .readSchema(internalSchema)
+          .latestCommitTime(Instant.ofEpochMilli(timestamp))
+          .latestMetadataPath(basePath + "/_delta_log")
+          .build();
+    } catch (Exception e) {
+      throw new RuntimeException("Failed to extract table information using Delta Kernel", e);
+    }
+  }
+}

xtable-core/src/main/java/org/apache/xtable/delta/DeltaSchemaExtractor.java

@@ -18,11 +18,7 @@
 
 package org.apache.xtable.delta;
 
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
@@ -41,22 +37,10 @@
 import org.apache.xtable.model.schema.InternalType;
 import org.apache.xtable.schema.SchemaUtils;
 
-/**
- * Converts between Delta and InternalTable schemas. Some items to be aware of:
- *
- * <ul>
- *   <li>Delta schemas are represented as Spark StructTypes which do not have enums so the enum
- *       types are lost when converting from XTable to Delta Lake representations
- *   <li>Delta does not have a fixed length byte array option so {@link InternalType#FIXED} is
- *       simply translated to a {@link org.apache.spark.sql.types.BinaryType}
- *   <li>Similarly, {@link InternalType#TIMESTAMP_NTZ} is translated to a long in Delta Lake
- * </ul>
- */
 @NoArgsConstructor(access = AccessLevel.PRIVATE)
 public class DeltaSchemaExtractor {
   private static final String DELTA_COLUMN_MAPPING_ID = "delta.columnMapping.id";
   private static final DeltaSchemaExtractor INSTANCE = new DeltaSchemaExtractor();
-  // Timestamps in Delta are microsecond precision by default
   private static final Map<InternalSchema.MetadataKey, Object>
       DEFAULT_TIMESTAMP_PRECISION_METADATA =
           Collections.singletonMap(

xtable-core/src/main/java/org/apache/xtable/delta/DeltaTableExtractor.java

@@ -62,7 +62,7 @@ public InternalTable table(DeltaLog deltaLog, String tableName, Long version) {
         .partitioningFields(partitionFields)
         .readSchema(schema)
         .latestCommitTime(Instant.ofEpochMilli(snapshot.timestamp()))
-        .latestMetdataPath(snapshot.deltaLog().logPath().toString())
+        .latestMetadataPath(snapshot.deltaLog().logPath().toString())
         .build();
   }
 }

xtable-core/src/main/java/org/apache/xtable/hudi/HudiTableExtractor.java

@@ -87,7 +87,7 @@ public InternalTable table(HoodieTableMetaClient metaClient, HoodieInstant commi
         .partitioningFields(partitionFields)
         .readSchema(canonicalSchema)
         .latestCommitTime(HudiInstantUtils.parseFromInstantTime(commit.getTimestamp()))
-        .latestMetdataPath(metaClient.getMetaPath().toString())
+        .latestMetadataPath(metaClient.getMetaPath().toString())
         .build();
   }
 

xtable-core/src/main/java/org/apache/xtable/iceberg/IcebergConversionSource.java

@@ -131,7 +131,7 @@ public InternalTable getTable(Snapshot snapshot) {
         .latestCommitTime(Instant.ofEpochMilli(snapshot.timestampMillis()))
         .readSchema(irSchema)
         .layoutStrategy(dataLayoutStrategy)
-        .latestMetdataPath(iceOps.current().metadataFileLocation())
+        .latestMetadataPath(iceOps.current().metadataFileLocation())
         .build();
   }