fix(ingest): Fix Timestamp Conversions, Add legacy api support (#14076)

Co-authored-by: Jonathan Vexler <=>
Co-authored-by: Y Ethan Guo <yihua@apache.org>

Author: jonvex

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/stats/SparkValueMetadataUtils.java

@@ -144,7 +144,7 @@ public static Comparable convertSparkToJava(ValueMetadata valueMetadata, Object
    * we need to return java.sql.Timestamp and java.sql.Date
    *
    */
-  public static Object convertJavaTypeToSparkType(Comparable<?> javaVal, boolean useJava8api) {
+  public static Object convertJavaTypeToSparkType(Object javaVal, boolean useJava8api) {
     if (!useJava8api) {
       if (javaVal instanceof Instant) {
         return Timestamp.from((Instant) javaVal);

hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java

@@ -179,6 +179,12 @@ public class HoodieTestDataGenerator implements AutoCloseable {
       + "{\"name\":\"dec_fixed_small\",\"type\":{\"type\":\"fixed\",\"name\":\"decFixedSmall\",\"size\":3,\"logicalType\":\"decimal\",\"precision\":5,\"scale\":2}},"
       + "{\"name\":\"dec_fixed_large\",\"type\":{\"type\":\"fixed\",\"name\":\"decFixedLarge\",\"size\":8,\"logicalType\":\"decimal\",\"precision\":18,\"scale\":9}},";
 
+  public static final String EXTENDED_LOGICAL_TYPES_SCHEMA_NO_LTS_V6 = "{\"name\":\"ts_millis\",\"type\":{\"type\":\"long\",\"logicalType\":\"timestamp-millis\"}},"
+      + "{\"name\":\"ts_micros\",\"type\":{\"type\":\"long\",\"logicalType\":\"timestamp-micros\"}},"
+      + "{\"name\":\"event_date\",\"type\":{\"type\":\"int\",\"logicalType\":\"date\"}},"
+      + "{\"name\":\"dec_fixed_small\",\"type\":{\"type\":\"fixed\",\"name\":\"decFixedSmall\",\"size\":3,\"logicalType\":\"decimal\",\"precision\":5,\"scale\":2}},"
+      + "{\"name\":\"dec_fixed_large\",\"type\":{\"type\":\"fixed\",\"name\":\"decFixedLarge\",\"size\":8,\"logicalType\":\"decimal\",\"precision\":18,\"scale\":9}},";
+
   public static final String EXTRA_COL_SCHEMA1 = "{\"name\": \"extra_column1\", \"type\": [\"null\", \"string\"], \"default\": null },";
   public static final String EXTRA_COL_SCHEMA2 = "{\"name\": \"extra_column2\", \"type\": [\"null\", \"string\"], \"default\": null},";
   public static final String EXTRA_COL_SCHEMA_FOR_AWS_DMS_PAYLOAD = "{\"name\": \"Op\", \"type\": [\"null\", \"string\"], \"default\": null},";
@@ -202,6 +208,9 @@ public class HoodieTestDataGenerator implements AutoCloseable {
   public static final String TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS =
       TRIP_SCHEMA_PREFIX + EXTENDED_LOGICAL_TYPES_SCHEMA_NO_LTS + TRIP_SCHEMA_SUFFIX;
 
+  public static final String TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS_V6 =
+      TRIP_SCHEMA_PREFIX + EXTENDED_LOGICAL_TYPES_SCHEMA_NO_LTS_V6 + TRIP_SCHEMA_SUFFIX;
+
 
   public static final String TRIP_NESTED_EXAMPLE_SCHEMA =
       TRIP_SCHEMA_PREFIX + FARE_NESTED_SCHEMA + TRIP_SCHEMA_SUFFIX;
@@ -233,6 +242,7 @@ public class HoodieTestDataGenerator implements AutoCloseable {
   public static final Schema AVRO_TRIP_LOGICAL_TYPES_SCHEMA = new Schema.Parser().parse(TRIP_LOGICAL_TYPES_SCHEMA);
   public static final Schema AVRO_TRIP_LOGICAL_TYPES_SCHEMA_V6 = new Schema.Parser().parse(TRIP_LOGICAL_TYPES_SCHEMA_V6);
   public static final Schema AVRO_TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS = new Schema.Parser().parse(TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS);
+  public static final Schema AVRO_TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS_V6 = new Schema.Parser().parse(TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS_V6);
   public static final Schema AVRO_TRIP_SCHEMA = new Schema.Parser().parse(TRIP_SCHEMA);
   public static final Schema FLATTENED_AVRO_SCHEMA = new Schema.Parser().parse(TRIP_FLATTENED_SCHEMA);
 
@@ -380,6 +390,8 @@ public IndexedRecord generateRandomValueAsPerSchema(String schemaStr, HoodieKey
         return generatePayloadForLogicalTypesSchemaV6(key, commitTime, false, timestamp);
       } else if (TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS.equals(schemaStr)) {
         return generatePayloadForLogicalTypesSchemaNoLTS(key, commitTime, false, timestamp);
+      } else if (TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS_V6.equals(schemaStr)) {
+        return generatePayloadForLogicalTypesSchemaNoLTSV6(key, commitTime, false, timestamp);
       }
     } else {
       if (TRIP_EXAMPLE_SCHEMA.equals(schemaStr)) {
@@ -390,6 +402,8 @@ public IndexedRecord generateRandomValueAsPerSchema(String schemaStr, HoodieKey
         return generatePayloadForLogicalTypesSchemaV6(key, commitTime, true, timestamp);
       } else if (TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS.equals(schemaStr)) {
         return generatePayloadForLogicalTypesSchemaNoLTS(key, commitTime, true, timestamp);
+      } else if (TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS_V6.equals(schemaStr)) {
+        return generatePayloadForLogicalTypesSchemaNoLTSV6(key, commitTime, true, timestamp);
       }
     }
 
@@ -451,6 +465,10 @@ public IndexedRecord generatePayloadForTripEncodedDecimalSchema(HoodieKey key, S
   public IndexedRecord generatePayloadForLogicalTypesSchemaNoLTS(HoodieKey key, String commitTime, boolean isDelete, long timestamp) {
     return generateRecordForTripLogicalTypesSchema(key, "rider-" + commitTime, "driver-" + commitTime, timestamp, isDelete, false, false);
   }
+  
+  public IndexedRecord generatePayloadForLogicalTypesSchemaNoLTSV6(HoodieKey key, String commitTime, boolean isDelete, long timestamp) {
+    return generateRecordForTripLogicalTypesSchema(key, "rider-" + commitTime, "driver-" + commitTime, timestamp, isDelete, true, false);
+  }
 
   public IndexedRecord generatePayloadForLogicalTypesSchema(HoodieKey key, String commitTime, boolean isDelete, long timestamp) {
     return generateRecordForTripLogicalTypesSchema(key, "rider-" + commitTime, "driver-" + commitTime, timestamp, isDelete, false, true);
@@ -679,8 +697,11 @@ public GenericRecord generateRecordForTripLogicalTypesSchema(HoodieKey key, Stri
                                                                long timestamp, boolean isDeleteRecord, boolean v6, boolean hasLTS) {
     GenericRecord rec;
     if (!hasLTS) {
-      // LTS = Local Timestamp
-      rec = new GenericData.Record(AVRO_TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS);
+      if (v6) {
+        rec = new GenericData.Record(AVRO_TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS_V6);
+      } else {
+        rec = new GenericData.Record(AVRO_TRIP_LOGICAL_TYPES_SCHEMA_NO_LTS);
+      }
     } else if (v6) {
       rec = new GenericData.Record(AVRO_TRIP_LOGICAL_TYPES_SCHEMA_V6);
     } else {

hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/MercifulJsonToRowConverter.java

@@ -30,8 +30,10 @@
 import org.apache.hudi.avro.processors.Parser;
 import org.apache.hudi.avro.processors.TimestampMicroLogicalTypeProcessor;
 import org.apache.hudi.avro.processors.TimestampMilliLogicalTypeProcessor;
+import org.apache.hudi.common.util.DateTimeUtils;
 import org.apache.hudi.common.util.collection.Pair;
 import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.stats.SparkValueMetadataUtils;
 import org.apache.hudi.stats.ValueType;
 import org.apache.hudi.utilities.exception.HoodieJsonToRowConversionException;
 
@@ -47,7 +49,6 @@
 
 import java.io.IOException;
 import java.math.BigDecimal;
-import java.sql.Timestamp;
 import java.time.Instant;
 import java.time.LocalDateTime;
 import java.time.temporal.ChronoField;
@@ -63,19 +64,21 @@
  * Converts Json record to Row Record.
  */
 public class MercifulJsonToRowConverter extends MercifulJsonConverter {
+  private final boolean useJava8api;
 
   /**
    * Allows enabling sanitization and allows choice of invalidCharMask for sanitization
    */
-  public MercifulJsonToRowConverter(boolean shouldSanitize, String invalidCharMask) {
-    this(new ObjectMapper().enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS), shouldSanitize, invalidCharMask);
+  public MercifulJsonToRowConverter(boolean shouldSanitize, String invalidCharMask, boolean useJava8api) {
+    this(new ObjectMapper().enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS), shouldSanitize, invalidCharMask, useJava8api);
   }
 
   /**
    * Allows a configured ObjectMapper to be passed for converting json records to row.
    */
-  public MercifulJsonToRowConverter(ObjectMapper mapper, boolean shouldSanitize, String invalidCharMask) {
+  public MercifulJsonToRowConverter(ObjectMapper mapper, boolean shouldSanitize, String invalidCharMask, boolean useJava8api) {
     super(mapper, shouldSanitize, invalidCharMask);
+    this.useJava8api = useJava8api;
   }
 
   /**
@@ -102,7 +105,7 @@ private Row convertJsonToRow(Map<String, Object> inputJson, Schema schema) {
     for (Schema.Field f : fields) {
       Object val = shouldSanitize ? getFieldFromJson(f, inputJson, schema.getFullName(), invalidCharMask) : inputJson.get(f.name());
       if (val != null) {
-        values.set(f.pos(), convertJsonField(val, f.name(), f.schema()));
+        values.set(f.pos(), SparkValueMetadataUtils.convertJavaTypeToSparkType(convertJsonField(val, f.name(), f.schema()), useJava8api));
       }
     }
     return RowFactory.create(values.toArray());
@@ -238,7 +241,7 @@ public Pair<Boolean, Object> handleStringValue(String value) {
           },
           value, schema);
       if (result.getLeft()) {
-        return Pair.of(true, new Timestamp((Long) result.getRight()));
+        return Pair.of(true, Instant.ofEpochMilli((Long) result.getRight()));
       }
       return Pair.of(false, null);
     }
@@ -275,8 +278,7 @@ public Pair<Boolean, Object> handleStringValue(String value) {
           },
           value, schema);
       if (result.getLeft()) {
-        // timestamp in spark sql doesn't support precision to the micro.
-        return Pair.of(true, new Timestamp(((Long) result.getRight()) / 1000));
+        return Pair.of(true, DateTimeUtils.microsToInstant((Long) result.getRight()));
       }
       return Pair.of(false, null);
     }

hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/RowConverter.java

@@ -43,17 +43,19 @@ public class RowConverter implements Serializable {
   private final String schemaStr;
   private final String invalidCharMask;
   private final boolean shouldSanitize;
+  private final boolean useJava8api;
 
   /**
    * To be lazily initialized on executors.
    */
   private transient MercifulJsonToRowConverter jsonConverter;
 
-  public RowConverter(Schema schema, boolean shouldSanitize, String invalidCharMask) {
+  public RowConverter(Schema schema, boolean shouldSanitize, String invalidCharMask, boolean useJava8api) {
     this.schemaStr = schema.toString();
     this.schema = schema;
     this.shouldSanitize = shouldSanitize;
     this.invalidCharMask = invalidCharMask;
+    this.useJava8api = useJava8api;
   }
 
   private void initSchema() {
@@ -65,7 +67,7 @@ private void initSchema() {
 
   private void initJsonConvertor() {
     if (jsonConverter == null) {
-      jsonConverter = new MercifulJsonToRowConverter(this.shouldSanitize, this.invalidCharMask);
+      jsonConverter = new MercifulJsonToRowConverter(this.shouldSanitize, this.invalidCharMask, this.useJava8api);
     }
   }
 

hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SourceFormatAdapter.java

@@ -47,6 +47,7 @@
 import org.apache.spark.sql.Column;
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
+import org.apache.spark.sql.internal.SQLConf;
 import org.apache.spark.sql.types.DataTypes;
 import org.apache.spark.sql.types.Metadata;
 import org.apache.spark.sql.types.StructField;
@@ -77,6 +78,9 @@ public class SourceFormatAdapter implements Closeable {
   private  boolean wrapWithException = ROW_THROW_EXPLICIT_EXCEPTIONS.defaultValue();
   private String invalidCharMask = SCHEMA_FIELD_NAME_INVALID_CHAR_MASK.defaultValue();
 
+  private boolean useJava8api = (boolean) SQLConf.DATETIME_JAVA8API_ENABLED().defaultValue().get();
+
+
   private Option<BaseErrorTableWriter> errorTableWriter = Option.empty();
 
   public SourceFormatAdapter(Source source) {
@@ -90,6 +94,7 @@ public SourceFormatAdapter(Source source, Option<BaseErrorTableWriter> errorTabl
       this.shouldSanitize = SanitizationUtils.shouldSanitize(props.get());
       this.invalidCharMask = SanitizationUtils.getInvalidCharMask(props.get());
       this.wrapWithException = ConfigUtils.getBooleanWithAltKeys(props.get(), ROW_THROW_EXPLICIT_EXCEPTIONS);
+      this.useJava8api = (boolean) getSource().getSparkSession().conf().get(SQLConf.DATETIME_JAVA8API_ENABLED());
     }
     if (this.shouldSanitize && source.getSourceType() == Source.SourceType.PROTO) {
       throw new IllegalArgumentException("PROTO cannot be sanitized");
@@ -112,6 +117,10 @@ private String getInvalidCharMask() {
     return invalidCharMask;
   }
 
+  private boolean getUseJava8api() {
+    return useJava8api;
+  }
+
   /**
    * transform input rdd of json string to generic records with support for adding error events to error table
    * @param inputBatch
@@ -134,7 +143,7 @@ private JavaRDD<GenericRecord> transformJsonToGenericRdd(InputBatch<JavaRDD<Stri
 
   private JavaRDD<Row> transformJsonToRowRdd(InputBatch<JavaRDD<String>> inputBatch) {
     MercifulJsonConverter.clearCache(inputBatch.getSchemaProvider().getSourceSchema().getFullName());
-    RowConverter convertor = new RowConverter(inputBatch.getSchemaProvider().getSourceSchema(), isFieldNameSanitizingEnabled(), getInvalidCharMask());
+    RowConverter convertor = new RowConverter(inputBatch.getSchemaProvider().getSourceSchema(), isFieldNameSanitizingEnabled(), getInvalidCharMask(), getUseJava8api());
     return inputBatch.getBatch().map(rdd -> {
       if (errorTableWriter.isPresent()) {
         JavaRDD<Either<Row, String>> javaRDD = rdd.map(convertor::fromJsonToRowWithError);