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>
@@ -713,7 +713,7 @@
                 </executions>
                 <configuration>
                     <skip>${skipUTs}</skip>
-                    <redirectTestOutputToFile>true</redirectTestOutputToFile>
+                    <redirectTestOutputToFile>false</redirectTestOutputToFile>
                     <trimStackTrace>false</trimStackTrace>
                     <forkedProcessExitTimeoutInSeconds>120</forkedProcessExitTimeoutInSeconds>
                 </configuration>

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/DeltaPartitionExtractor.java

@@ -79,7 +79,7 @@ public class DeltaPartitionExtractor {
   // For timestamp partition fields, actual partition column names in delta format will be of type
   // generated & and with a name like `delta_partition_col_{transform_type}_{source_field_name}`.
   private static final String DELTA_PARTITION_COL_NAME_FORMAT = "xtable_partition_col_%s_%s";
-  static final String DELTA_GENERATION_EXPRESSION = "delta.generationExpression";
+  public static final String DELTA_GENERATION_EXPRESSION = "delta.generationExpression";
   private static final List<ParsedGeneratedExpr.GeneratedExprType> GRANULARITIES =
       Arrays.asList(
           ParsedGeneratedExpr.GeneratedExprType.YEAR,

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/kernel/DeltaKernelActionsConverter.java

@@ -0,0 +1,123 @@
+/*
+ * 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.kernel;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+
+import scala.collection.JavaConverters;
+
+import io.delta.kernel.Table;
+import io.delta.kernel.defaults.engine.DefaultEngine;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.internal.actions.AddFile;
+import io.delta.kernel.internal.actions.RemoveFile;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.stat.ColumnStat;
+import org.apache.xtable.model.stat.FileStats;
+import org.apache.xtable.model.storage.FileFormat;
+import org.apache.xtable.model.storage.InternalDataFile;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class DeltaKernelActionsConverter {
+  private static final DeltaKernelActionsConverter INSTANCE = new DeltaKernelActionsConverter();
+
+  public static DeltaKernelActionsConverter getInstance() {
+    return INSTANCE;
+  }
+
+  public InternalDataFile convertAddActionToInternalDataFile(
+      AddFile addFile,
+      Table table,
+      FileFormat fileFormat,
+      List<InternalPartitionField> partitionFields,
+      List<InternalField> fields,
+      boolean includeColumnStats,
+      DeltaKernelPartitionExtractor partitionExtractor,
+      DeltaKernelStatsExtractor fileStatsExtractor,
+      Map<String, String> partitionValues) {
+    FileStats fileStats = fileStatsExtractor.getColumnStatsForFile(addFile, fields);
+    List<ColumnStat> columnStats =
+        includeColumnStats ? fileStats.getColumnStats() : Collections.emptyList();
+    long recordCount = fileStats.getNumRecords();
+    // The immutable map from Java to Scala is not working, need to
+
+    scala.collection.mutable.Map<String, String> scalaMap =
+        JavaConverters.mapAsScalaMap(partitionValues);
+
+    return InternalDataFile.builder()
+        .physicalPath(getFullPathToFile(addFile.getPath(), table))
+        .fileFormat(fileFormat)
+        .fileSizeBytes(addFile.getSize())
+        .lastModified(addFile.getModificationTime())
+        .partitionValues(partitionExtractor.partitionValueExtraction(scalaMap, partitionFields))
+        .columnStats(columnStats)
+        .recordCount(recordCount)
+        .build();
+  }
+
+  public InternalDataFile convertRemoveActionToInternalDataFile(
+      RemoveFile removeFile,
+      Table table,
+      FileFormat fileFormat,
+      List<InternalPartitionField> partitionFields,
+      DeltaKernelPartitionExtractor partitionExtractor,
+      Map<String, String> partitionValues) {
+    scala.collection.mutable.Map<String, String> scalaMap =
+        JavaConverters.mapAsScalaMap(partitionValues);
+
+    return InternalDataFile.builder()
+        .physicalPath(getFullPathToFile(removeFile.getPath(), table))
+        .fileFormat(fileFormat)
+        .partitionValues(partitionExtractor.partitionValueExtraction(scalaMap, partitionFields))
+        .build();
+  }
+
+  public FileFormat convertToFileFormat(String provider) {
+    if (provider.equals("parquet")) {
+      return FileFormat.APACHE_PARQUET;
+    } else if (provider.equals("orc")) {
+      return FileFormat.APACHE_ORC;
+    }
+    throw new NotSupportedException(
+        String.format("delta file format %s is not recognized", provider));
+  }
+
+  static String getFullPathToFile(String dataFilePath, Table table) {
+    Configuration hadoopConf = new Configuration();
+    Engine myEngine = DefaultEngine.create(hadoopConf);
+    String tableBasePath = table.getPath(myEngine);
+    ;
+    //            String tableBasePath = snapshot.dataPath().toUri().toString();
+    if (dataFilePath.startsWith(tableBasePath)) {
+      return dataFilePath;
+    }
+    return tableBasePath + Path.SEPARATOR + dataFilePath;
+  }
+}