Address comments

Author: vinishjail97

pom.xml

@@ -87,7 +87,7 @@
         <spotless.version>2.43.0</spotless.version>
         <apache.rat.version>0.16.1</apache.rat.version>
         <google.java.format.version>1.8</google.java.format.version>
-        <delta.standalone.version>0.5.0</delta.standalone.version>
+        <delta.standalone.version>3.3.0</delta.standalone.version>
         <delta.hive.version>3.0.0</delta.hive.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <target.dir.pattern>**/target/**</target.dir.pattern>

xtable-core/src/main/java/org/apache/xtable/avro/AvroSchemaConverter.java

@@ -160,12 +160,14 @@ private InternalSchema toInternalSchema(
           metadata.put(
               InternalSchema.MetadataKey.TIMESTAMP_PRECISION, InternalSchema.MetadataValue.MICROS);
         } else if (logicalType instanceof LogicalTypes.LocalTimestampMillis) {
-          // TODO: Hudi 0.x writes INT64 in parquet, TimestampNTZType support added in 1.x
+          // TODO: https://github.com/apache/incubator-xtable/issues/672
+          // Hudi 0.x writes INT64 in parquet, TimestampNTZType support added in 1.x
           newDataType = InternalType.LONG;
           metadata.put(
               InternalSchema.MetadataKey.TIMESTAMP_PRECISION, InternalSchema.MetadataValue.MILLIS);
         } else if (logicalType instanceof LogicalTypes.LocalTimestampMicros) {
-          // TODO: Hudi 0.x writes INT64 in parquet, TimestampNTZType support added in 1.x
+          // TODO: https://github.com/apache/incubator-xtable/issues/672
+          // Hudi 0.x writes INT64 in parquet, TimestampNTZType support added in 1.x
           newDataType = InternalType.LONG;
           metadata.put(
               InternalSchema.MetadataKey.TIMESTAMP_PRECISION, InternalSchema.MetadataValue.MICROS);

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

@@ -322,13 +322,36 @@ public void testTimestamps() {
                                 .metadata(metadata)
                                 .build())
                         .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+                        .build(),
+                    InternalField.builder()
+                        .name("requiredTimestampNtz")
+                        .schema(
+                            InternalSchema.builder()
+                                .name("timestamp_ntz")
+                                .dataType(InternalType.TIMESTAMP_NTZ)
+                                .isNullable(false)
+                                .metadata(metadata)
+                                .build())
+                        .build(),
+                    InternalField.builder()
+                        .name("optionalTimestampNtz")
+                        .schema(
+                            InternalSchema.builder()
+                                .name("timestamp_ntz")
+                                .dataType(InternalType.TIMESTAMP_NTZ)
+                                .isNullable(true)
+                                .metadata(metadata)
+                                .build())
+                        .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
                         .build()))
             .build();
 
     StructType structRepresentationTimestamp =
         new StructType()
             .add("requiredTimestamp", DataTypes.TimestampType, false)
-            .add("optionalTimestamp", DataTypes.TimestampType, true);
+            .add("optionalTimestamp", DataTypes.TimestampType, true)
+            .add("requiredTimestampNtz", DataTypes.TimestampNTZType, false)
+            .add("optionalTimestampNtz", DataTypes.TimestampNTZType, true);
 
     Assertions.assertEquals(
         internalSchemaTimestamp,

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

@@ -78,7 +78,6 @@
 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;
 
@@ -96,6 +95,7 @@
 import org.apache.xtable.model.storage.DataLayoutStrategy;
 import org.apache.xtable.model.storage.FileFormat;
 import org.apache.xtable.model.storage.InternalDataFile;
+import org.apache.xtable.model.storage.InternalFile;
 import org.apache.xtable.model.storage.PartitionFileGroup;
 import org.apache.xtable.model.storage.TableFormat;
 import org.apache.xtable.schema.SchemaFieldFinder;
@@ -460,17 +460,10 @@ public void testTimestampNtz() {
 
     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));
+    validateDeltaTableUsingSpark(basePath, new HashSet<>(Arrays.asList(dataFile1, dataFile2)));
     TableFormatSync.getInstance()
         .syncSnapshot(Collections.singletonList(conversionTarget), snapshot2);
-    Assertions.assertThrows(
-        DeltaErrors.InvalidProtocolVersionException.class,
-        () ->
-            validateDeltaTable(basePath, new HashSet<>(Arrays.asList(dataFile1, dataFile2)), null));
+    validateDeltaTableUsingSpark(basePath, new HashSet<>(Arrays.asList(dataFile2, dataFile3)));
   }
 
   private static Stream<Arguments> timestampPartitionTestingArgs() {
@@ -514,6 +507,13 @@ private void validateDeltaTable(
         internalDataFiles.size(), count, "Number of files from DeltaScan don't match expectation");
   }
 
+  private void validateDeltaTableUsingSpark(
+      Path basePath, Set<InternalDataFile> internalDataFiles) {
+    Dataset<Row> dataset = sparkSession.read().format("delta").load(basePath.toString());
+    long countFromFiles = internalDataFiles.stream().mapToLong(InternalFile::getRecordCount).sum();
+    Assertions.assertEquals(countFromFiles, dataset.count());
+  }
+
   private InternalSnapshot buildSnapshot(
       InternalTable table, String sourceIdentifier, InternalDataFile... dataFiles) {
     return InternalSnapshot.builder()