Handle ntz in other formats as well

Author: vinishjail97

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

@@ -160,11 +160,13 @@ private InternalSchema toInternalSchema(
           metadata.put(
               InternalSchema.MetadataKey.TIMESTAMP_PRECISION, InternalSchema.MetadataValue.MICROS);
         } else if (logicalType instanceof LogicalTypes.LocalTimestampMillis) {
-          newDataType = InternalType.TIMESTAMP_NTZ;
+          // TODO: 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) {
-          newDataType = InternalType.TIMESTAMP_NTZ;
+          // TODO: 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);
         } else {
@@ -350,6 +352,22 @@ private Schema fromInternalSchema(InternalSchema internalSchema, String currentP
       case INT:
         return finalizeSchema(Schema.create(Schema.Type.INT), internalSchema);
       case LONG:
+        if (internalSchema.getMetadata() != null
+            && internalSchema
+                .getMetadata()
+                .containsKey(InternalSchema.MetadataKey.TIMESTAMP_PRECISION)) {
+          if (internalSchema.getMetadata().get(InternalSchema.MetadataKey.TIMESTAMP_PRECISION)
+              == InternalSchema.MetadataValue.MILLIS) {
+            return finalizeSchema(
+                LogicalTypes.localTimestampMillis().addToSchema(Schema.create(Schema.Type.LONG)),
+                internalSchema);
+          }
+          {
+            return finalizeSchema(
+                LogicalTypes.localTimestampMicros().addToSchema(Schema.create(Schema.Type.LONG)),
+                internalSchema);
+          }
+        }
         return finalizeSchema(Schema.create(Schema.Type.LONG), internalSchema);
       case STRING:
         return finalizeSchema(Schema.create(Schema.Type.STRING), internalSchema);

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

@@ -173,7 +173,6 @@ Type toIcebergType(InternalField field, AtomicInteger fieldIdTracker) {
       case INT:
         return Types.IntegerType.get();
       case LONG:
-      case TIMESTAMP_NTZ: // TODO - revisit this
         return Types.LongType.get();
       case BYTES:
         return Types.BinaryType.get();
@@ -189,6 +188,8 @@ Type toIcebergType(InternalField field, AtomicInteger fieldIdTracker) {
         return Types.DateType.get();
       case TIMESTAMP:
         return Types.TimestampType.withZone();
+      case TIMESTAMP_NTZ:
+        return Types.TimestampType.withoutZone();
       case DOUBLE:
         return Types.DoubleType.get();
       case DECIMAL:

xtable-core/src/test/java/org/apache/xtable/avro/TestAvroSchemaConverter.java

@@ -576,7 +576,7 @@ public void testAvroLogicalTypes() {
                         .schema(
                             InternalSchema.builder()
                                 .name("long")
-                                .dataType(InternalType.TIMESTAMP_NTZ)
+                                .dataType(InternalType.LONG)
                                 .isNullable(false)
                                 .metadata(millisMetadata)
                                 .build())
@@ -586,7 +586,7 @@ public void testAvroLogicalTypes() {
                         .schema(
                             InternalSchema.builder()
                                 .name("long")
-                                .dataType(InternalType.TIMESTAMP_NTZ)
+                                .dataType(InternalType.LONG)
                                 .isNullable(false)
                                 .metadata(microsMetadata)
                                 .build())

xtable-core/src/test/java/org/apache/xtable/iceberg/TestIcebergSchemaExtractor.java

@@ -501,14 +501,18 @@ public void testTimestamps() {
                 1, "requiredTimestampMillis", Types.TimestampType.withZone()),
             Types.NestedField.optional(
                 2, "optionalTimestampMillis", Types.TimestampType.withZone()),
-            Types.NestedField.required(3, "requiredTimestampNtzMillis", Types.LongType.get()),
-            Types.NestedField.optional(4, "optionalTimestampNtzMillis", Types.LongType.get()),
+            Types.NestedField.required(
+                3, "requiredTimestampNtzMillis", Types.TimestampType.withoutZone()),
+            Types.NestedField.optional(
+                4, "optionalTimestampNtzMillis", Types.TimestampType.withoutZone()),
             Types.NestedField.required(
                 5, "requiredTimestampMicros", Types.TimestampType.withZone()),
             Types.NestedField.optional(
                 6, "optionalTimestampMicros", Types.TimestampType.withZone()),
-            Types.NestedField.required(7, "requiredTimestampNtzMicros", Types.LongType.get()),
-            Types.NestedField.optional(8, "optionalTimestampNtzMicros", Types.LongType.get()));
+            Types.NestedField.required(
+                7, "requiredTimestampNtzMicros", Types.TimestampType.withoutZone()),
+            Types.NestedField.optional(
+                8, "optionalTimestampNtzMicros", Types.TimestampType.withoutZone()));
     assertTrue(expectedTargetSchema.sameSchema(SCHEMA_EXTRACTOR.toIceberg(irSchema)));
 
     Schema sourceSchema =