Handle timestamp_ntz in delta conversion target

Author: vinishjail97

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

@@ -73,9 +73,9 @@
 
 @Log4j2
 public class DeltaConversionTarget implements ConversionTarget {
-  private static final String MIN_READER_VERSION = String.valueOf(1);
+  private static final int MIN_READER_VERSION = 1;
   // gets access to generated columns.
-  private static final String MIN_WRITER_VERSION = String.valueOf(4);
+  private static final int MIN_WRITER_VERSION = 4;
 
   private DeltaLog deltaLog;
   private DeltaSchemaExtractor schemaExtractor;
@@ -329,8 +329,14 @@ private void commitTransaction() {
 
     private Map<String, String> getConfigurationsForDeltaSync() {
       Map<String, String> configMap = new HashMap<>();
-      configMap.put(DeltaConfigs.MIN_READER_VERSION().key(), MIN_READER_VERSION);
-      configMap.put(DeltaConfigs.MIN_WRITER_VERSION().key(), MIN_WRITER_VERSION);
+      configMap.put(
+          DeltaConfigs.MIN_READER_VERSION().key(),
+          String.valueOf(
+              Math.max(deltaLog.snapshot().protocol().minReaderVersion(), MIN_READER_VERSION)));
+      configMap.put(
+          DeltaConfigs.MIN_WRITER_VERSION().key(),
+          String.valueOf(
+              Math.max(deltaLog.snapshot().protocol().minWriterVersion(), MIN_WRITER_VERSION)));
       configMap.put(TableSyncMetadata.XTABLE_METADATA, metadata.toJson());
       // Sets retention for the Delta Log, does not impact underlying files in the table
       configMap.put(

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

@@ -56,6 +56,11 @@
 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(
+              InternalSchema.MetadataKey.TIMESTAMP_PRECISION, InternalSchema.MetadataValue.MICROS);
 
   public static DeltaSchemaExtractor getInstance() {
     return INSTANCE;
@@ -110,11 +115,11 @@ private InternalSchema toInternalSchema(
         break;
       case "timestamp":
         type = InternalType.TIMESTAMP;
-        // Timestamps in Delta are microsecond precision by default
-        metadata =
-            Collections.singletonMap(
-                InternalSchema.MetadataKey.TIMESTAMP_PRECISION,
-                InternalSchema.MetadataValue.MICROS);
+        metadata = DEFAULT_TIMESTAMP_PRECISION_METADATA;
+        break;
+      case "timestamp_ntz":
+        type = InternalType.TIMESTAMP_NTZ;
+        metadata = DEFAULT_TIMESTAMP_PRECISION_METADATA;
         break;
       case "struct":
         StructType structType = (StructType) dataType;

xtable-core/src/main/java/org/apache/xtable/schema/SparkSchemaExtractor.java

@@ -62,7 +62,6 @@ private DataType convertFieldType(InternalField field) {
       case INT:
         return DataTypes.IntegerType;
       case LONG:
-      case TIMESTAMP_NTZ:
         return DataTypes.LongType;
       case BYTES:
       case FIXED:
@@ -76,6 +75,8 @@ private DataType convertFieldType(InternalField field) {
         return DataTypes.DateType;
       case TIMESTAMP:
         return DataTypes.TimestampType;
+      case TIMESTAMP_NTZ:
+        return DataTypes.TimestampNTZType;
       case DOUBLE:
         return DataTypes.DoubleType;
       case DECIMAL:

xtable-core/src/test/java/org/apache/xtable/delta/TestDeltaSync.java

@@ -53,6 +53,7 @@
 import org.apache.spark.sql.SparkSession;
 import org.apache.spark.sql.types.DataTypes;
 import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -77,6 +78,7 @@
 import io.delta.standalone.expressions.EqualTo;
 import io.delta.standalone.expressions.Expression;
 import io.delta.standalone.expressions.Literal;
+import io.delta.standalone.internal.exception.DeltaErrors;
 import io.delta.standalone.types.IntegerType;
 import io.delta.standalone.types.StringType;
 
@@ -431,6 +433,46 @@ public void testGetTargetCommitIdentifierWithNullSourceIdentifier() throws Excep
     assertFalse(unmappedTargetId.isPresent());
   }
 
+  @Test
+  public void testTimestampNtz() {
+    InternalSchema schema1 = getInternalSchemaWithTimestampNtz();
+    List<InternalField> fields2 = new ArrayList<>(schema1.getFields());
+    fields2.add(
+        InternalField.builder()
+            .name("float_field")
+            .schema(
+                InternalSchema.builder()
+                    .name("float")
+                    .dataType(InternalType.FLOAT)
+                    .isNullable(true)
+                    .build())
+            .build());
+    InternalSchema schema2 = getInternalSchema().toBuilder().fields(fields2).build();
+    InternalTable table1 = getInternalTable(tableName, basePath, schema1, null, LAST_COMMIT_TIME);
+    InternalTable table2 = getInternalTable(tableName, basePath, schema2, null, LAST_COMMIT_TIME);
+
+    InternalDataFile dataFile1 = getDataFile(1, Collections.emptyList(), basePath);
+    InternalDataFile dataFile2 = getDataFile(2, Collections.emptyList(), basePath);
+    InternalDataFile dataFile3 = getDataFile(3, Collections.emptyList(), basePath);
+
+    InternalSnapshot snapshot1 = buildSnapshot(table1, "0", dataFile1, dataFile2);
+    InternalSnapshot snapshot2 = buildSnapshot(table2, "1", dataFile2, dataFile3);
+
+    TableFormatSync.getInstance()
+        .syncSnapshot(Collections.singletonList(conversionTarget), snapshot1);
+    // Delta standalone library can't read versions (3,7) and needs delta kernel dependency.
+    Assertions.assertThrows(
+        DeltaErrors.InvalidProtocolVersionException.class,
+        () ->
+            validateDeltaTable(basePath, new HashSet<>(Arrays.asList(dataFile1, dataFile2)), null));
+    TableFormatSync.getInstance()
+        .syncSnapshot(Collections.singletonList(conversionTarget), snapshot2);
+    Assertions.assertThrows(
+        DeltaErrors.InvalidProtocolVersionException.class,
+        () ->
+            validateDeltaTable(basePath, new HashSet<>(Arrays.asList(dataFile1, dataFile2)), null));
+  }
+
   private static Stream<Arguments> timestampPartitionTestingArgs() {
     return Stream.of(
         Arguments.of(PartitionTransformType.YEAR),
@@ -563,6 +605,25 @@ private InternalSchema getInternalSchema() {
         .build();
   }
 
+  private InternalSchema getInternalSchemaWithTimestampNtz() {
+    Map<InternalSchema.MetadataKey, Object> timestampMetadata = new HashMap<>();
+    timestampMetadata.put(
+        InternalSchema.MetadataKey.TIMESTAMP_PRECISION, InternalSchema.MetadataValue.MICROS);
+    List<InternalField> fields = new ArrayList<>(getInternalSchema().getFields());
+    fields.add(
+        InternalField.builder()
+            .name("timestamp_ntz_field")
+            .schema(
+                InternalSchema.builder()
+                    .name("time_ntz")
+                    .dataType(InternalType.TIMESTAMP_NTZ)
+                    .isNullable(true)
+                    .metadata(timestampMetadata)
+                    .build())
+            .build());
+    return getInternalSchema().toBuilder().fields(fields).build();
+  }
+
   private static SparkSession buildSparkSession() {
     SparkConf sparkConf =
         new SparkConf()

xtable-core/src/test/java/org/apache/xtable/schema/TestSparkSchemaExtractor.java

@@ -385,8 +385,8 @@ public void testTimestamps() {
 
     StructType structRepresentationTimestampNtz =
         new StructType()
-            .add("requiredTimestampNtz", DataTypes.LongType, false)
-            .add("optionalTimestampNtz", DataTypes.LongType, true);
+            .add("requiredTimestampNtz", DataTypes.TimestampNTZType, false)
+            .add("optionalTimestampNtz", DataTypes.TimestampNTZType, true);
 
     Assertions.assertEquals(
         structRepresentationTimestamp,