You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "Davis-Zhang-Onehouse (via GitHub)" <gi...@apache.org> on 2024/05/20 16:14:02 UTC
[PR] [HUDI-7774] Extend json convertor to support Avro logical type [hudi]
Davis-Zhang-Onehouse opened a new pull request, #11265:
URL: https://github.com/apache/hudi/pull/11265
Extended Merciful json converter to support Time, Date, UUID, Timestamp, LocalTimestamp and Duration. For supported and unsupported converter input please refer unit tests.
Test:
unit test with exhaustive branch coverage.
### Change Logs
The merciful json convertor now is able to convert raw json input to avro logical type data that are compliant to the avro schema specifications
### Impact
As an extension of component functionality, it has no impact on existing behavior. For use cases the extension addressed, it should work automatically
### Risk level (write none, low medium or high below)
Low given the code coverage.
### Documentation Update
None
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ ] Change Logs and Impact were stated clearly
- [ ] Adequate tests were added if applicable
- [ ] CI passed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2248779321
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24007",
"triggerID" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"triggerType" : "PUSH"
}, {
"hash" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24017",
"triggerID" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"triggerType" : "PUSH"
}, {
"hash" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24101",
"triggerID" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"triggerType" : "PUSH"
}, {
"hash" : "5cdb05c7d909edebbc74ccf75e26c07820272090",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24774",
"triggerID" : "5cdb05c7d909edebbc74ccf75e26c07820272090",
"triggerType" : "PUSH"
}, {
"hash" : "c0f09441fc00593d5e6c5f99402247a1c1dbda7a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24981",
"triggerID" : "c0f09441fc00593d5e6c5f99402247a1c1dbda7a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5cdb05c7d909edebbc74ccf75e26c07820272090 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24774)
* c0f09441fc00593d5e6c5f99402247a1c1dbda7a Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24981)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2248860491
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24007",
"triggerID" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"triggerType" : "PUSH"
}, {
"hash" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24017",
"triggerID" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"triggerType" : "PUSH"
}, {
"hash" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24101",
"triggerID" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"triggerType" : "PUSH"
}, {
"hash" : "5cdb05c7d909edebbc74ccf75e26c07820272090",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24774",
"triggerID" : "5cdb05c7d909edebbc74ccf75e26c07820272090",
"triggerType" : "PUSH"
}, {
"hash" : "c0f09441fc00593d5e6c5f99402247a1c1dbda7a",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24981",
"triggerID" : "c0f09441fc00593d5e6c5f99402247a1c1dbda7a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c0f09441fc00593d5e6c5f99402247a1c1dbda7a Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24981)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2136102419
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24007",
"triggerID" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"triggerType" : "PUSH"
}, {
"hash" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24017",
"triggerID" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"triggerType" : "PUSH"
}, {
"hash" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24017)
* 5699a637e30e3fafd056ab25d668640c5c8700c0 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "Davis-Zhang-Onehouse (via GitHub)" <gi...@apache.org>.
Davis-Zhang-Onehouse commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1690309755
##########
hudi-common/src/test/java/org/apache/hudi/avro/TestMercifulJsonConverter.java:
##########
@@ -55,6 +70,649 @@ public void basicConversion() throws IOException {
Assertions.assertEquals(rec, CONVERTER.convert(json, simpleSchema));
}
+ private static final String DECIMAL_AVRO_FILE_INVALID_PATH = "/decimal-logical-type-invalid.avsc";
+ private static final String DECIMAL_AVRO_FILE_PATH = "/decimal-logical-type.avsc";
+ private static final String DECIMAL_FIXED_AVRO_FILE_PATH = "/decimal-logical-type-fixed-type.avsc";
+ /**
+ * Covered case:
+ * Avro Logical Type: Decimal
+ * Exhaustive unsupported input coverage.
+ */
+ @ParameterizedTest
+ @MethodSource("decimalBadCases")
+ void decimalLogicalTypeInvalidCaseTest(String avroFile, String strInput, Double numInput,
+ boolean testFixedByteArray) throws IOException {
+ Schema schema = SchemaTestUtil.getSchemaFromResourceFilePath(avroFile);
+
+ Map<String, Object> data = new HashMap<>();
+ if (strInput != null) {
+ data.put("decimalField", strInput);
+ } else if (numInput != null) {
+ data.put("decimalField", numInput);
+ } else if (testFixedByteArray) {
+ // Convert the fixed value to int array, which is used as json value literals.
+ int[] intArray = {0, 0, 48, 57};
+ data.put("decimalField", intArray);
+ }
+ String json = MAPPER.writeValueAsString(data);
+
+ // Schedule with timestamp same as that of committed instant
+ assertThrows(MercifulJsonConverter.HoodieJsonToAvroConversionException.class, () -> {
+ CONVERTER.convert(json, schema);
+ });
+ }
+
+ static Stream<Object> decimalBadCases() {
+ return Stream.of(
+ // Invalid schema definition.
+ Arguments.of(DECIMAL_AVRO_FILE_INVALID_PATH, "123.45", null, false),
+ // Schema set precision as 5, input overwhelmed the precision.
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "123333.45", null, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, null, 123333.45, false),
+ // Schema precision set to 5, scale set to 2, so there is only 3 digit to accommodate integer part.
+ // As we do not do rounding, any input with more than 3 digit integer would fail.
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "1233", null, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, null, 1233D, false),
+ // Schema set scale as 2, input overwhelmed the scale.
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "0.222", null, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, null, 0.222, false),
+ // Invalid string which cannot be parsed as number.
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "", null, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "NotAValidString", null, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "-", null, false),
+ // Schema requires byte type while input is fixed type raw data.
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, null, null, true)
+ );
+ }
+
+ /**
+ * Covered case:
+ * Avro Logical Type: Decimal
+ * Avro type: bytes, fixed
+ * Input: Check test parameter
+ * Output: Object using Byte data type as the schema specified.
+ * */
+ @ParameterizedTest
+ @MethodSource("decimalGoodCases")
+ void decimalLogicalTypeTest(String avroFilePath, String groundTruth, String strInput,
+ Double numInput, boolean testFixedByteArray) throws IOException {
+ BigDecimal bigDecimal = new BigDecimal(groundTruth);
+ Map<String, Object> data = new HashMap<>();
+
+ Schema schema = SchemaTestUtil.getSchemaFromResourceFilePath(avroFilePath);
+ GenericRecord record = new GenericData.Record(schema);
+ Conversions.DecimalConversion conv = new Conversions.DecimalConversion();
+ Schema decimalFieldSchema = schema.getField("decimalField").schema();
+
+ // Decide the decimal field input according to the test dimension.
+ if (strInput != null) {
+ data.put("decimalField", strInput); // String number input
+ } else if (numInput != null) {
+ data.put("decimalField", numInput); // Number input
+ } else if (testFixedByteArray) {
+ // Fixed byte array input.
+ // Example: 123.45 - byte array [0, 0, 48, 57].
+ Schema fieldSchema = schema.getField("decimalField").schema();
+ GenericFixed fixedValue = new Conversions.DecimalConversion().toFixed(
+ bigDecimal, fieldSchema, fieldSchema.getLogicalType());
+ // Convert the fixed value to int array, which is used as json value literals.
+ byte[] byteArray = fixedValue.bytes();
+ int[] intArray = new int[byteArray.length];
+ for (int i = 0; i < byteArray.length; i++) {
+ // Byte is signed in Java, int is 32-bit. Convert by & 0xFF to handle negative values correctly.
+ intArray[i] = byteArray[i] & 0xFF;
+ }
+ data.put("decimalField", intArray);
+ }
+
+ // Decide the decimal field expected output according to the test dimension.
+ if (avroFilePath.equals(DECIMAL_AVRO_FILE_PATH)) {
+ record.put("decimalField", conv.toBytes(bigDecimal, decimalFieldSchema, decimalFieldSchema.getLogicalType()));
+ } else {
+ record.put("decimalField", conv.toFixed(bigDecimal, decimalFieldSchema, decimalFieldSchema.getLogicalType()));
+ }
+
+ String json = MAPPER.writeValueAsString(data);
+
+ GenericRecord real = CONVERTER.convert(json, schema);
+ Assertions.assertEquals(record, real);
+ }
+
+ static Stream<Object> decimalGoodCases() {
+ return Stream.of(
+ // The schema all set precision as 5, scale as 2.
+ // Test dimension: Schema file, Ground truth, string input, number input, fixed byte array input.
+ // Test some random numbers.
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "123.45", "123.45", null, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "123.45", null, 123.45, false),
+ // Test MIN/MAX allowed by the schema.
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "-999.99", "-999.99", null, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "999.99",null, 999.99, false),
+ // Test 0.
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "0", null, 0D, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "0", "0", null, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "0", "000.00", null, false),
+ // Same set of coverage over schame using byte/fixed type.
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "123.45", "123.45", null, false),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "123.45", null, 123.45, false),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "-999.99", "-999.99", null, false),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "999.99",null, 999.99, false),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "0", null, 0D, false),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "0", "0", null, true),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "0", "000.00", null, true),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "123.45", null, null, true),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "123.45", null, 123.45, true),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "-999.99", null, null, true),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "999.99", null, 999.99, true),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "0", null, null, true),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "0", null, null, true),
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2136082128
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24007",
"triggerID" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"triggerType" : "PUSH"
}, {
"hash" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24017",
"triggerID" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24017)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2214800910
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24007",
"triggerID" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"triggerType" : "PUSH"
}, {
"hash" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24017",
"triggerID" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"triggerType" : "PUSH"
}, {
"hash" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24101",
"triggerID" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"triggerType" : "PUSH"
}, {
"hash" : "5cdb05c7d909edebbc74ccf75e26c07820272090",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24774",
"triggerID" : "5cdb05c7d909edebbc74ccf75e26c07820272090",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5699a637e30e3fafd056ab25d668640c5c8700c0 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24101)
* 5cdb05c7d909edebbc74ccf75e26c07820272090 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24774)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1690295012
##########
hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java:
##########
@@ -187,196 +180,772 @@ private static Object convertJsonToAvroField(Object value, String name, Schema s
throw new HoodieJsonToAvroConversionException(null, name, schema, shouldSanitize, invalidCharMask);
}
- JsonToAvroFieldProcessor processor = FIELD_TYPE_PROCESSORS.get(schema.getType());
- if (null != processor) {
- return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
- }
- throw new IllegalArgumentException("JsonConverter cannot handle type: " + schema.getType());
+ return JsonToAvroFieldProcessorUtil.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
}
- /**
- * Base Class for converting json to avro fields.
- */
- private abstract static class JsonToAvroFieldProcessor implements Serializable {
+ private static class JsonToAvroFieldProcessorUtil {
+ /**
+ * Base Class for converting json to avro fields.
+ */
+ private abstract static class JsonToAvroFieldProcessor implements Serializable {
- public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
- if (!res.getLeft()) {
- throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
+ if (!res.getLeft()) {
+ throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+ return res.getRight();
}
- return res.getRight();
+
+ protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
}
- protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
- }
+ public static Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ JsonToAvroFieldProcessor processor = getProcessorForSchema(schema);
+ return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ private static JsonToAvroFieldProcessor getProcessorForSchema(Schema schema) {
+ JsonToAvroFieldProcessor processor = null;
+
+ // 3 cases to consider: customized logicalType, logicalType, and type.
+ String customizedLogicalType = schema.getProp("logicalType");
+ LogicalType logicalType = schema.getLogicalType();
+ Type type = schema.getType();
+ if (customizedLogicalType != null && !customizedLogicalType.isEmpty()) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(customizedLogicalType);
+ } else if (logicalType != null) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(logicalType.getName());
+ } else {
+ processor = AVRO_TYPE_FIELD_TYPE_PROCESSORS.get(type);
+ }
+
+ ValidationUtils.checkArgument(
+ processor != null, String.format("JsonConverter cannot handle type: %s", type));
+ return processor;
+ }
+
+ // Avro primitive and complex type processors.
+ private static final Map<Schema.Type, JsonToAvroFieldProcessor> AVRO_TYPE_FIELD_TYPE_PROCESSORS = getFieldTypeProcessors();
+ // Avro logical type processors.
+ private static final Map<String, JsonToAvroFieldProcessor> AVRO_LOGICAL_TYPE_FIELD_PROCESSORS = getLogicalFieldTypeProcessors();
+
+ /**
+ * Build type processor map for each avro type.
+ */
+ private static Map<Schema.Type, JsonToAvroFieldProcessor> getFieldTypeProcessors() {
+ Map<Schema.Type, JsonToAvroFieldProcessor> fieldTypeProcessors = new EnumMap<>(Schema.Type.class);
+ fieldTypeProcessors.put(Type.STRING, generateStringTypeHandler());
+ fieldTypeProcessors.put(Type.BOOLEAN, generateBooleanTypeHandler());
+ fieldTypeProcessors.put(Type.DOUBLE, generateDoubleTypeHandler());
+ fieldTypeProcessors.put(Type.FLOAT, generateFloatTypeHandler());
+ fieldTypeProcessors.put(Type.INT, generateIntTypeHandler());
+ fieldTypeProcessors.put(Type.LONG, generateLongTypeHandler());
+ fieldTypeProcessors.put(Type.ARRAY, generateArrayTypeHandler());
+ fieldTypeProcessors.put(Type.RECORD, generateRecordTypeHandler());
+ fieldTypeProcessors.put(Type.ENUM, generateEnumTypeHandler());
+ fieldTypeProcessors.put(Type.MAP, generateMapTypeHandler());
+ fieldTypeProcessors.put(Type.BYTES, generateBytesTypeHandler());
+ fieldTypeProcessors.put(Type.FIXED, generateFixedTypeHandler());
+ return Collections.unmodifiableMap(fieldTypeProcessors);
+ }
- private static JsonToAvroFieldProcessor generateBooleanTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ private static Map<String, JsonToAvroFieldProcessor> getLogicalFieldTypeProcessors() {
+ return CollectionUtils.createImmutableMap(
+ Pair.of(AvroLogicalTypeEnum.DECIMAL.getValue(), new DecimalLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIME_MICROS.getValue(), new TimeMicroLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIME_MILLIS.getValue(), new TimeMilliLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.DATE.getValue(), new DateLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MICROS.getValue(), new LocalTimestampMicroLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MILLIS.getValue(), new LocalTimestampMilliLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIMESTAMP_MICROS.getValue(), new TimestampMicroLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIMESTAMP_MILLIS.getValue(), new TimestampMilliLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.DURATION.getValue(), new DurationLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.UUID.getValue(), generateStringTypeHandler()));
+ }
+
+ private static class DecimalLogicalTypeProcessor extends JsonToAvroFieldProcessor {
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Boolean) {
- return Pair.of(true, value);
+
+ if (!isValidDecimalTypeConfig(schema)) {
+ return Pair.of(false, null);
+ }
+
+ // Case 1: Input is a list. It is expected to be raw Fixed byte array input, and we only support
+ // parsing it to Fixed avro type.
+ if (value instanceof List<?> && schema.getType() == Type.FIXED) {
+ JsonToAvroFieldProcessor processor = generateFixedTypeHandler();
+ return processor.convert(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ // Case 2: Input is a number or String number.
+ LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+ Pair<Boolean, BigDecimal> parseResult = parseObjectToBigDecimal(value);
+ if (Boolean.FALSE.equals(parseResult.getLeft())) {
+ return Pair.of(false, null);
+ }
+ BigDecimal bigDecimal = parseResult.getRight();
+
+ // As we don't do rounding, the validation will enforce the scale part and the integer part are all within the
+ // limit. As a result, if scale is 2 precision is 5, we only allow 3 digits for the integer.
+ // Allowed: 123.45, 123, 0.12
+ // Disallowed: 1234 (4 digit integer while the scale has already reserved 2 digit out of the 5 digit precision)
+ // 123456, 0.12345
+ if (bigDecimal.scale() > decimalType.getScale()
+ || (bigDecimal.precision() - bigDecimal.scale()) > (decimalType.getPrecision() - decimalType.getScale())) {
+ // Correspond to case
+ // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 5 as scale 2 without rounding.
+ // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 3 as scale 2 without rounding
+ return Pair.of(false, null);
+ }
+
+ switch (schema.getType()) {
+ case BYTES:
+ // Convert to primitive Arvo type that logical type Decimal uses.
+ ByteBuffer byteBuffer = new Conversions.DecimalConversion().toBytes(bigDecimal, schema, decimalType);
+ return Pair.of(true, byteBuffer);
+ case FIXED:
+ GenericFixed fixedValue = new Conversions.DecimalConversion().toFixed(bigDecimal, schema, decimalType);
+ return Pair.of(true, fixedValue);
+ default: {
+ return Pair.of(false, null);
+ }
}
- return Pair.of(false, null);
}
- };
- }
- private static JsonToAvroFieldProcessor generateIntTypeHandler() {
- return new JsonToAvroFieldProcessor() {
- @Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).intValue());
- } else if (value instanceof String) {
- return Pair.of(true, Integer.valueOf((String) value));
+ /**
+ * Check if the given schema is a valid decimal type configuration.
+ */
+ private static boolean isValidDecimalTypeConfig(Schema schema) {
+ LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+ // At the time when the schema is found not valid when it is parsed, the Avro Schema.parse will just silently
+ // set the schema to be null instead of throwing exceptions. Correspondingly, we just check if it is null here.
+ if (decimalType == null) {
+ return false;
}
- return Pair.of(false, null);
+ // Even though schema is validated at schema parsing phase, still validate here to be defensive.
+ decimalType.validate(schema);
+ return true;
}
- };
- }
- private static JsonToAvroFieldProcessor generateDoubleTypeHandler() {
- return new JsonToAvroFieldProcessor() {
- @Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).doubleValue());
- } else if (value instanceof String) {
- return Pair.of(true, Double.valueOf((String) value));
+ /**
+ * Parse the object to BigDecimal.
+ *
+ * @param obj Object to be parsed
+ * @return Pair object, with left as boolean indicating if the parsing was successful and right as the
+ * BigDecimal value.
+ */
+ private static Pair<Boolean, BigDecimal> parseObjectToBigDecimal(Object obj) {
+ // Case 1: Object is a number.
+ if (obj instanceof Number) {
+ return Pair.of(true, BigDecimal.valueOf(((Number) obj).doubleValue()));
+ }
+
+ // Case 2: Object is a number in String format.
+ if (obj instanceof String) {
+ BigDecimal bigDecimal = null;
+ try {
+ bigDecimal = new BigDecimal(((String) obj));
+ } catch (java.lang.NumberFormatException ignored) {
+ /* ignore */
+ }
+ return Pair.of(bigDecimal != null, bigDecimal);
}
return Pair.of(false, null);
}
- };
- }
+ }
- private static JsonToAvroFieldProcessor generateFloatTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ private static class DurationLogicalTypeProcessor extends JsonToAvroFieldProcessor {
+ private static final int NUM_ELEMENTS_FOR_DURATION_TYPE = 3;
+
+ /**
+ * We expect the input to be a list of 3 integers representing months, days and milliseconds.
+ */
+ private boolean isValidDurationInput(Object value) {
+ if (!(value instanceof List<?>)) {
+ return false;
+ }
+ List<?> list = (List<?>) value;
+ if (list.size() != NUM_ELEMENTS_FOR_DURATION_TYPE) {
+ return false;
+ }
+ for (Object element : list) {
+ if (!(element instanceof Integer)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Convert the given object to Avro object with schema whose logical type is duration.
+ */
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).floatValue());
- } else if (value instanceof String) {
- return Pair.of(true, Float.valueOf((String) value));
+
+ if (!isValidDurationTypeConfig(schema)) {
+ return Pair.of(false, null);
}
- return Pair.of(false, null);
+ if (!isValidDurationInput(value)) {
+ return Pair.of(false, null);
+ }
+ // After the validation the input can be safely cast to List<Integer> with 3 elements.
+ List<?> list = (List<?>) value;
+ List<Integer> converval = list.stream()
+ .filter(Integer.class::isInstance)
+ .map(Integer.class::cast)
+ .collect(Collectors.toList());
+
+ ByteBuffer buffer = ByteBuffer.allocate(schema.getFixedSize()).order(ByteOrder.LITTLE_ENDIAN);
+ for (Integer element : converval) {
+ buffer.putInt(element); // months
+ }
+ return Pair.of(true, new GenericData.Fixed(schema, buffer.array()));
+ }
+
+ /**
+ * Check if the given schema is a valid decimal type configuration.
+ */
+ private static boolean isValidDurationTypeConfig(Schema schema) {
+ String durationTypeName = AvroLogicalTypeEnum.DURATION.getValue();
+ LogicalType durationType = schema.getLogicalType();
+ String durationTypeProp = schema.getProp("logicalType");
+ // 1. The Avro type should be "Fixed".
+ // 2. Fixed size must be of 12 bytes as it hold 3 integers.
+ // 3. Logical type name should be "duration". The name might be stored in different places based on Avro version
+ // being used here.
+ return schema.getType().equals(Type.FIXED)
+ && schema.getFixedSize() == Integer.BYTES * NUM_ELEMENTS_FOR_DURATION_TYPE
+ && (durationType != null && durationType.getName().equals(durationTypeName)
+ || durationTypeProp != null && durationTypeProp.equals(durationTypeName));
+ }
+ }
+
+ /**
+ * Processor utility handling Number inputs. Consumed by TimeLogicalTypeProcessor.
+ */
+ private interface NumericParser {
+ // Convert the input number to Avro data type according to the class
+ // implementing this interface.
+ Pair<Boolean, Object> handleNumberValue(Number value);
+
+ // Convert the input number to Avro data type according to the class
+ // implementing this interface.
+ // @param value the input number in string format.
+ Pair<Boolean, Object> handleStringNumber(String value);
+
+ interface IntParser extends NumericParser {
+ @Override
+ default Pair<Boolean, Object> handleNumberValue(Number value) {
+ return Pair.of(true, value.intValue());
+ }
+
+ @Override
+ default Pair<Boolean, Object> handleStringNumber(String value) {
+ return Pair.of(true, Integer.parseInt(value));
+ }
+ }
+
+ interface LongParser extends NumericParser {
+ @Override
+ default Pair<Boolean, Object> handleNumberValue(Number value) {
+ return Pair.of(true, value.longValue());
+ }
+
+ @Override
+ default Pair<Boolean, Object> handleStringNumber(String value) {
+ return Pair.of(true, Long.parseLong(value));
+ }
+ }
+ }
+
+ /**
+ * Base Class for converting object to avro logical type TimeMilli/TimeMicro.
+ */
+ private abstract static class TimeLogicalTypeProcessor extends JsonToAvroFieldProcessor implements NumericParser {
+
+ protected static final LocalDateTime LOCAL_UNIX_EPOCH = LocalDateTime.of(1970, 1, 1, 0, 0, 0, 0);
+
+ // Logical type the processor is handling.
+ private final AvroLogicalTypeEnum logicalTypeEnum;
+
+ public TimeLogicalTypeProcessor(AvroLogicalTypeEnum logicalTypeEnum) {
+ this.logicalTypeEnum = logicalTypeEnum;
}
- };
- }
- private static JsonToAvroFieldProcessor generateLongTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ /**
+ * Main function that convert input to Object with java data type specified by schema
+ */
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ LogicalType logicalType = schema.getLogicalType();
+ if (logicalType == null) {
+ return Pair.of(false, null);
+ }
+ logicalType.validate(schema);
if (value instanceof Number) {
- return Pair.of(true, ((Number) value).longValue());
- } else if (value instanceof String) {
- return Pair.of(true, Long.valueOf((String) value));
+ return handleNumberValue((Number) value);
+ }
+ if (value instanceof String) {
+ String valStr = (String) value;
+ if (ALL_DIGITS_WITH_OPTIONAL_SIGN.matcher(valStr).matches()) {
+ return handleStringNumber(valStr);
+ } else if (isWellFormedDateTime(valStr)) {
+ return handleStringValue(valStr);
+ }
}
return Pair.of(false, null);
}
- };
- }
- private static JsonToAvroFieldProcessor generateStringTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ // Handle the case when the input is a string that may be parsed as a time.
+ protected abstract Pair<Boolean, Object> handleStringValue(String value);
+
+ protected DateTimeFormatter getDateTimeFormatter() {
+ DateTimeParseContext ctx = DATE_TIME_PARSE_CONTEXT_MAP.get(logicalTypeEnum);
+ assert ctx != null : String.format("%s should have configured date time context.", logicalTypeEnum.getValue());
+ return ctx.dateTimeFormatter;
+ }
+
+ protected Pattern getDateTimePattern() {
+ DateTimeParseContext ctx = DATE_TIME_PARSE_CONTEXT_MAP.get(logicalTypeEnum);
+ assert ctx != null : String.format("%s should have configured date time context.", logicalTypeEnum.getValue());
+ return ctx.dateTimePattern;
+ }
+
+ // Depending on the logical type the processor handles, they use different parsing context
+ // when they need to parse a timestamp string in handleStringValue.
+ private static class DateTimeParseContext {
+ public DateTimeParseContext(DateTimeFormatter dateTimeFormatter, Pattern dateTimePattern) {
+ this.dateTimeFormatter = dateTimeFormatter;
+ this.dateTimePattern = dateTimePattern;
+ }
+
+ public final Pattern dateTimePattern;
+
+ public final DateTimeFormatter dateTimeFormatter;
+ }
+
+ private static final Map<AvroLogicalTypeEnum, DateTimeParseContext> DATE_TIME_PARSE_CONTEXT_MAP = getParseContext();
+
+ private static Map<AvroLogicalTypeEnum, DateTimeParseContext> getParseContext() {
+ // Formatter for parsing local timestamp. It ignores the time zone info of the timestamp.
Review Comment:
```suggestion
// Formatter for parsing a timestamp. It assumes UTC timezone, with 'Z' as the zone ID.
```
##########
hudi-common/src/test/java/org/apache/hudi/avro/TestMercifulJsonConverter.java:
##########
@@ -55,6 +70,649 @@ public void basicConversion() throws IOException {
Assertions.assertEquals(rec, CONVERTER.convert(json, simpleSchema));
}
+ private static final String DECIMAL_AVRO_FILE_INVALID_PATH = "/decimal-logical-type-invalid.avsc";
+ private static final String DECIMAL_AVRO_FILE_PATH = "/decimal-logical-type.avsc";
+ private static final String DECIMAL_FIXED_AVRO_FILE_PATH = "/decimal-logical-type-fixed-type.avsc";
+ /**
+ * Covered case:
+ * Avro Logical Type: Decimal
+ * Exhaustive unsupported input coverage.
+ */
+ @ParameterizedTest
+ @MethodSource("decimalBadCases")
+ void decimalLogicalTypeInvalidCaseTest(String avroFile, String strInput, Double numInput,
+ boolean testFixedByteArray) throws IOException {
+ Schema schema = SchemaTestUtil.getSchemaFromResourceFilePath(avroFile);
+
+ Map<String, Object> data = new HashMap<>();
+ if (strInput != null) {
+ data.put("decimalField", strInput);
+ } else if (numInput != null) {
+ data.put("decimalField", numInput);
+ } else if (testFixedByteArray) {
+ // Convert the fixed value to int array, which is used as json value literals.
+ int[] intArray = {0, 0, 48, 57};
+ data.put("decimalField", intArray);
+ }
+ String json = MAPPER.writeValueAsString(data);
+
+ // Schedule with timestamp same as that of committed instant
+ assertThrows(MercifulJsonConverter.HoodieJsonToAvroConversionException.class, () -> {
+ CONVERTER.convert(json, schema);
+ });
+ }
+
+ static Stream<Object> decimalBadCases() {
+ return Stream.of(
+ // Invalid schema definition.
+ Arguments.of(DECIMAL_AVRO_FILE_INVALID_PATH, "123.45", null, false),
+ // Schema set precision as 5, input overwhelmed the precision.
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "123333.45", null, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, null, 123333.45, false),
+ // Schema precision set to 5, scale set to 2, so there is only 3 digit to accommodate integer part.
+ // As we do not do rounding, any input with more than 3 digit integer would fail.
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "1233", null, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, null, 1233D, false),
+ // Schema set scale as 2, input overwhelmed the scale.
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "0.222", null, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, null, 0.222, false),
+ // Invalid string which cannot be parsed as number.
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "", null, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "NotAValidString", null, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "-", null, false),
+ // Schema requires byte type while input is fixed type raw data.
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, null, null, true)
+ );
+ }
+
+ /**
+ * Covered case:
+ * Avro Logical Type: Decimal
+ * Avro type: bytes, fixed
+ * Input: Check test parameter
+ * Output: Object using Byte data type as the schema specified.
+ * */
+ @ParameterizedTest
+ @MethodSource("decimalGoodCases")
+ void decimalLogicalTypeTest(String avroFilePath, String groundTruth, String strInput,
+ Double numInput, boolean testFixedByteArray) throws IOException {
+ BigDecimal bigDecimal = new BigDecimal(groundTruth);
+ Map<String, Object> data = new HashMap<>();
+
+ Schema schema = SchemaTestUtil.getSchemaFromResourceFilePath(avroFilePath);
+ GenericRecord record = new GenericData.Record(schema);
+ Conversions.DecimalConversion conv = new Conversions.DecimalConversion();
+ Schema decimalFieldSchema = schema.getField("decimalField").schema();
+
+ // Decide the decimal field input according to the test dimension.
+ if (strInput != null) {
+ data.put("decimalField", strInput); // String number input
+ } else if (numInput != null) {
+ data.put("decimalField", numInput); // Number input
+ } else if (testFixedByteArray) {
+ // Fixed byte array input.
+ // Example: 123.45 - byte array [0, 0, 48, 57].
+ Schema fieldSchema = schema.getField("decimalField").schema();
+ GenericFixed fixedValue = new Conversions.DecimalConversion().toFixed(
+ bigDecimal, fieldSchema, fieldSchema.getLogicalType());
+ // Convert the fixed value to int array, which is used as json value literals.
+ byte[] byteArray = fixedValue.bytes();
+ int[] intArray = new int[byteArray.length];
+ for (int i = 0; i < byteArray.length; i++) {
+ // Byte is signed in Java, int is 32-bit. Convert by & 0xFF to handle negative values correctly.
+ intArray[i] = byteArray[i] & 0xFF;
+ }
+ data.put("decimalField", intArray);
+ }
+
+ // Decide the decimal field expected output according to the test dimension.
+ if (avroFilePath.equals(DECIMAL_AVRO_FILE_PATH)) {
+ record.put("decimalField", conv.toBytes(bigDecimal, decimalFieldSchema, decimalFieldSchema.getLogicalType()));
+ } else {
+ record.put("decimalField", conv.toFixed(bigDecimal, decimalFieldSchema, decimalFieldSchema.getLogicalType()));
+ }
+
+ String json = MAPPER.writeValueAsString(data);
+
+ GenericRecord real = CONVERTER.convert(json, schema);
+ Assertions.assertEquals(record, real);
+ }
+
+ static Stream<Object> decimalGoodCases() {
+ return Stream.of(
+ // The schema all set precision as 5, scale as 2.
+ // Test dimension: Schema file, Ground truth, string input, number input, fixed byte array input.
+ // Test some random numbers.
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "123.45", "123.45", null, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "123.45", null, 123.45, false),
+ // Test MIN/MAX allowed by the schema.
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "-999.99", "-999.99", null, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "999.99",null, 999.99, false),
+ // Test 0.
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "0", null, 0D, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "0", "0", null, false),
+ Arguments.of(DECIMAL_AVRO_FILE_PATH, "0", "000.00", null, false),
+ // Same set of coverage over schame using byte/fixed type.
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "123.45", "123.45", null, false),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "123.45", null, 123.45, false),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "-999.99", "-999.99", null, false),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "999.99",null, 999.99, false),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "0", null, 0D, false),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "0", "0", null, true),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "0", "000.00", null, true),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "123.45", null, null, true),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "123.45", null, 123.45, true),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "-999.99", null, null, true),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "999.99", null, 999.99, true),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "0", null, null, true),
+ Arguments.of(DECIMAL_FIXED_AVRO_FILE_PATH, "0", null, null, true),
Review Comment:
Remove duplicate entries.
##########
hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java:
##########
@@ -187,196 +180,772 @@ private static Object convertJsonToAvroField(Object value, String name, Schema s
throw new HoodieJsonToAvroConversionException(null, name, schema, shouldSanitize, invalidCharMask);
}
- JsonToAvroFieldProcessor processor = FIELD_TYPE_PROCESSORS.get(schema.getType());
- if (null != processor) {
- return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
- }
- throw new IllegalArgumentException("JsonConverter cannot handle type: " + schema.getType());
+ return JsonToAvroFieldProcessorUtil.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
}
- /**
- * Base Class for converting json to avro fields.
- */
- private abstract static class JsonToAvroFieldProcessor implements Serializable {
+ private static class JsonToAvroFieldProcessorUtil {
+ /**
+ * Base Class for converting json to avro fields.
+ */
+ private abstract static class JsonToAvroFieldProcessor implements Serializable {
- public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
- if (!res.getLeft()) {
- throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
+ if (!res.getLeft()) {
+ throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+ return res.getRight();
}
- return res.getRight();
+
+ protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
}
- protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
- }
+ public static Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ JsonToAvroFieldProcessor processor = getProcessorForSchema(schema);
+ return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ private static JsonToAvroFieldProcessor getProcessorForSchema(Schema schema) {
+ JsonToAvroFieldProcessor processor = null;
+
+ // 3 cases to consider: customized logicalType, logicalType, and type.
+ String customizedLogicalType = schema.getProp("logicalType");
+ LogicalType logicalType = schema.getLogicalType();
+ Type type = schema.getType();
+ if (customizedLogicalType != null && !customizedLogicalType.isEmpty()) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(customizedLogicalType);
+ } else if (logicalType != null) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(logicalType.getName());
+ } else {
+ processor = AVRO_TYPE_FIELD_TYPE_PROCESSORS.get(type);
+ }
+
+ ValidationUtils.checkArgument(
+ processor != null, String.format("JsonConverter cannot handle type: %s", type));
+ return processor;
+ }
+
+ // Avro primitive and complex type processors.
+ private static final Map<Schema.Type, JsonToAvroFieldProcessor> AVRO_TYPE_FIELD_TYPE_PROCESSORS = getFieldTypeProcessors();
+ // Avro logical type processors.
+ private static final Map<String, JsonToAvroFieldProcessor> AVRO_LOGICAL_TYPE_FIELD_PROCESSORS = getLogicalFieldTypeProcessors();
+
+ /**
+ * Build type processor map for each avro type.
+ */
+ private static Map<Schema.Type, JsonToAvroFieldProcessor> getFieldTypeProcessors() {
+ Map<Schema.Type, JsonToAvroFieldProcessor> fieldTypeProcessors = new EnumMap<>(Schema.Type.class);
+ fieldTypeProcessors.put(Type.STRING, generateStringTypeHandler());
+ fieldTypeProcessors.put(Type.BOOLEAN, generateBooleanTypeHandler());
+ fieldTypeProcessors.put(Type.DOUBLE, generateDoubleTypeHandler());
+ fieldTypeProcessors.put(Type.FLOAT, generateFloatTypeHandler());
+ fieldTypeProcessors.put(Type.INT, generateIntTypeHandler());
+ fieldTypeProcessors.put(Type.LONG, generateLongTypeHandler());
+ fieldTypeProcessors.put(Type.ARRAY, generateArrayTypeHandler());
+ fieldTypeProcessors.put(Type.RECORD, generateRecordTypeHandler());
+ fieldTypeProcessors.put(Type.ENUM, generateEnumTypeHandler());
+ fieldTypeProcessors.put(Type.MAP, generateMapTypeHandler());
+ fieldTypeProcessors.put(Type.BYTES, generateBytesTypeHandler());
+ fieldTypeProcessors.put(Type.FIXED, generateFixedTypeHandler());
+ return Collections.unmodifiableMap(fieldTypeProcessors);
+ }
- private static JsonToAvroFieldProcessor generateBooleanTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ private static Map<String, JsonToAvroFieldProcessor> getLogicalFieldTypeProcessors() {
+ return CollectionUtils.createImmutableMap(
+ Pair.of(AvroLogicalTypeEnum.DECIMAL.getValue(), new DecimalLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIME_MICROS.getValue(), new TimeMicroLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIME_MILLIS.getValue(), new TimeMilliLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.DATE.getValue(), new DateLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MICROS.getValue(), new LocalTimestampMicroLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MILLIS.getValue(), new LocalTimestampMilliLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIMESTAMP_MICROS.getValue(), new TimestampMicroLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIMESTAMP_MILLIS.getValue(), new TimestampMilliLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.DURATION.getValue(), new DurationLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.UUID.getValue(), generateStringTypeHandler()));
+ }
+
+ private static class DecimalLogicalTypeProcessor extends JsonToAvroFieldProcessor {
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Boolean) {
- return Pair.of(true, value);
+
+ if (!isValidDecimalTypeConfig(schema)) {
+ return Pair.of(false, null);
+ }
+
+ // Case 1: Input is a list. It is expected to be raw Fixed byte array input, and we only support
+ // parsing it to Fixed avro type.
+ if (value instanceof List<?> && schema.getType() == Type.FIXED) {
+ JsonToAvroFieldProcessor processor = generateFixedTypeHandler();
+ return processor.convert(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ // Case 2: Input is a number or String number.
+ LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+ Pair<Boolean, BigDecimal> parseResult = parseObjectToBigDecimal(value);
+ if (Boolean.FALSE.equals(parseResult.getLeft())) {
+ return Pair.of(false, null);
+ }
+ BigDecimal bigDecimal = parseResult.getRight();
+
+ // As we don't do rounding, the validation will enforce the scale part and the integer part are all within the
+ // limit. As a result, if scale is 2 precision is 5, we only allow 3 digits for the integer.
+ // Allowed: 123.45, 123, 0.12
+ // Disallowed: 1234 (4 digit integer while the scale has already reserved 2 digit out of the 5 digit precision)
+ // 123456, 0.12345
+ if (bigDecimal.scale() > decimalType.getScale()
+ || (bigDecimal.precision() - bigDecimal.scale()) > (decimalType.getPrecision() - decimalType.getScale())) {
+ // Correspond to case
+ // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 5 as scale 2 without rounding.
+ // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 3 as scale 2 without rounding
+ return Pair.of(false, null);
+ }
+
+ switch (schema.getType()) {
+ case BYTES:
+ // Convert to primitive Arvo type that logical type Decimal uses.
+ ByteBuffer byteBuffer = new Conversions.DecimalConversion().toBytes(bigDecimal, schema, decimalType);
+ return Pair.of(true, byteBuffer);
+ case FIXED:
+ GenericFixed fixedValue = new Conversions.DecimalConversion().toFixed(bigDecimal, schema, decimalType);
+ return Pair.of(true, fixedValue);
+ default: {
+ return Pair.of(false, null);
+ }
}
- return Pair.of(false, null);
}
- };
- }
- private static JsonToAvroFieldProcessor generateIntTypeHandler() {
- return new JsonToAvroFieldProcessor() {
- @Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).intValue());
- } else if (value instanceof String) {
- return Pair.of(true, Integer.valueOf((String) value));
+ /**
+ * Check if the given schema is a valid decimal type configuration.
+ */
+ private static boolean isValidDecimalTypeConfig(Schema schema) {
+ LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+ // At the time when the schema is found not valid when it is parsed, the Avro Schema.parse will just silently
+ // set the schema to be null instead of throwing exceptions. Correspondingly, we just check if it is null here.
+ if (decimalType == null) {
+ return false;
}
- return Pair.of(false, null);
+ // Even though schema is validated at schema parsing phase, still validate here to be defensive.
+ decimalType.validate(schema);
+ return true;
}
- };
- }
- private static JsonToAvroFieldProcessor generateDoubleTypeHandler() {
- return new JsonToAvroFieldProcessor() {
- @Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).doubleValue());
- } else if (value instanceof String) {
- return Pair.of(true, Double.valueOf((String) value));
+ /**
+ * Parse the object to BigDecimal.
+ *
+ * @param obj Object to be parsed
+ * @return Pair object, with left as boolean indicating if the parsing was successful and right as the
+ * BigDecimal value.
+ */
+ private static Pair<Boolean, BigDecimal> parseObjectToBigDecimal(Object obj) {
+ // Case 1: Object is a number.
+ if (obj instanceof Number) {
+ return Pair.of(true, BigDecimal.valueOf(((Number) obj).doubleValue()));
+ }
+
+ // Case 2: Object is a number in String format.
+ if (obj instanceof String) {
+ BigDecimal bigDecimal = null;
+ try {
+ bigDecimal = new BigDecimal(((String) obj));
+ } catch (java.lang.NumberFormatException ignored) {
+ /* ignore */
+ }
+ return Pair.of(bigDecimal != null, bigDecimal);
}
return Pair.of(false, null);
}
- };
- }
+ }
- private static JsonToAvroFieldProcessor generateFloatTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ private static class DurationLogicalTypeProcessor extends JsonToAvroFieldProcessor {
+ private static final int NUM_ELEMENTS_FOR_DURATION_TYPE = 3;
+
+ /**
+ * We expect the input to be a list of 3 integers representing months, days and milliseconds.
+ */
+ private boolean isValidDurationInput(Object value) {
+ if (!(value instanceof List<?>)) {
+ return false;
+ }
+ List<?> list = (List<?>) value;
+ if (list.size() != NUM_ELEMENTS_FOR_DURATION_TYPE) {
+ return false;
+ }
+ for (Object element : list) {
+ if (!(element instanceof Integer)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Convert the given object to Avro object with schema whose logical type is duration.
+ */
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).floatValue());
- } else if (value instanceof String) {
- return Pair.of(true, Float.valueOf((String) value));
+
+ if (!isValidDurationTypeConfig(schema)) {
+ return Pair.of(false, null);
}
- return Pair.of(false, null);
+ if (!isValidDurationInput(value)) {
+ return Pair.of(false, null);
+ }
+ // After the validation the input can be safely cast to List<Integer> with 3 elements.
+ List<?> list = (List<?>) value;
+ List<Integer> converval = list.stream()
+ .filter(Integer.class::isInstance)
+ .map(Integer.class::cast)
+ .collect(Collectors.toList());
+
+ ByteBuffer buffer = ByteBuffer.allocate(schema.getFixedSize()).order(ByteOrder.LITTLE_ENDIAN);
+ for (Integer element : converval) {
+ buffer.putInt(element); // months
+ }
+ return Pair.of(true, new GenericData.Fixed(schema, buffer.array()));
+ }
+
+ /**
+ * Check if the given schema is a valid decimal type configuration.
+ */
+ private static boolean isValidDurationTypeConfig(Schema schema) {
+ String durationTypeName = AvroLogicalTypeEnum.DURATION.getValue();
+ LogicalType durationType = schema.getLogicalType();
+ String durationTypeProp = schema.getProp("logicalType");
+ // 1. The Avro type should be "Fixed".
+ // 2. Fixed size must be of 12 bytes as it hold 3 integers.
+ // 3. Logical type name should be "duration". The name might be stored in different places based on Avro version
+ // being used here.
+ return schema.getType().equals(Type.FIXED)
+ && schema.getFixedSize() == Integer.BYTES * NUM_ELEMENTS_FOR_DURATION_TYPE
+ && (durationType != null && durationType.getName().equals(durationTypeName)
+ || durationTypeProp != null && durationTypeProp.equals(durationTypeName));
+ }
+ }
+
+ /**
+ * Processor utility handling Number inputs. Consumed by TimeLogicalTypeProcessor.
+ */
+ private interface NumericParser {
+ // Convert the input number to Avro data type according to the class
+ // implementing this interface.
+ Pair<Boolean, Object> handleNumberValue(Number value);
+
+ // Convert the input number to Avro data type according to the class
+ // implementing this interface.
+ // @param value the input number in string format.
+ Pair<Boolean, Object> handleStringNumber(String value);
+
+ interface IntParser extends NumericParser {
+ @Override
+ default Pair<Boolean, Object> handleNumberValue(Number value) {
+ return Pair.of(true, value.intValue());
+ }
+
+ @Override
+ default Pair<Boolean, Object> handleStringNumber(String value) {
+ return Pair.of(true, Integer.parseInt(value));
+ }
+ }
+
+ interface LongParser extends NumericParser {
+ @Override
+ default Pair<Boolean, Object> handleNumberValue(Number value) {
+ return Pair.of(true, value.longValue());
+ }
+
+ @Override
+ default Pair<Boolean, Object> handleStringNumber(String value) {
+ return Pair.of(true, Long.parseLong(value));
+ }
+ }
+ }
+
+ /**
+ * Base Class for converting object to avro logical type TimeMilli/TimeMicro.
+ */
+ private abstract static class TimeLogicalTypeProcessor extends JsonToAvroFieldProcessor implements NumericParser {
+
+ protected static final LocalDateTime LOCAL_UNIX_EPOCH = LocalDateTime.of(1970, 1, 1, 0, 0, 0, 0);
+
+ // Logical type the processor is handling.
+ private final AvroLogicalTypeEnum logicalTypeEnum;
+
+ public TimeLogicalTypeProcessor(AvroLogicalTypeEnum logicalTypeEnum) {
+ this.logicalTypeEnum = logicalTypeEnum;
}
- };
- }
- private static JsonToAvroFieldProcessor generateLongTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ /**
+ * Main function that convert input to Object with java data type specified by schema
+ */
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ LogicalType logicalType = schema.getLogicalType();
+ if (logicalType == null) {
+ return Pair.of(false, null);
+ }
+ logicalType.validate(schema);
if (value instanceof Number) {
- return Pair.of(true, ((Number) value).longValue());
- } else if (value instanceof String) {
- return Pair.of(true, Long.valueOf((String) value));
+ return handleNumberValue((Number) value);
+ }
+ if (value instanceof String) {
+ String valStr = (String) value;
+ if (ALL_DIGITS_WITH_OPTIONAL_SIGN.matcher(valStr).matches()) {
+ return handleStringNumber(valStr);
+ } else if (isWellFormedDateTime(valStr)) {
+ return handleStringValue(valStr);
+ }
}
return Pair.of(false, null);
}
- };
- }
- private static JsonToAvroFieldProcessor generateStringTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ // Handle the case when the input is a string that may be parsed as a time.
+ protected abstract Pair<Boolean, Object> handleStringValue(String value);
+
+ protected DateTimeFormatter getDateTimeFormatter() {
+ DateTimeParseContext ctx = DATE_TIME_PARSE_CONTEXT_MAP.get(logicalTypeEnum);
+ assert ctx != null : String.format("%s should have configured date time context.", logicalTypeEnum.getValue());
+ return ctx.dateTimeFormatter;
+ }
+
+ protected Pattern getDateTimePattern() {
+ DateTimeParseContext ctx = DATE_TIME_PARSE_CONTEXT_MAP.get(logicalTypeEnum);
+ assert ctx != null : String.format("%s should have configured date time context.", logicalTypeEnum.getValue());
+ return ctx.dateTimePattern;
+ }
+
+ // Depending on the logical type the processor handles, they use different parsing context
+ // when they need to parse a timestamp string in handleStringValue.
+ private static class DateTimeParseContext {
+ public DateTimeParseContext(DateTimeFormatter dateTimeFormatter, Pattern dateTimePattern) {
+ this.dateTimeFormatter = dateTimeFormatter;
+ this.dateTimePattern = dateTimePattern;
+ }
+
+ public final Pattern dateTimePattern;
+
+ public final DateTimeFormatter dateTimeFormatter;
+ }
+
+ private static final Map<AvroLogicalTypeEnum, DateTimeParseContext> DATE_TIME_PARSE_CONTEXT_MAP = getParseContext();
+
+ private static Map<AvroLogicalTypeEnum, DateTimeParseContext> getParseContext() {
+ // Formatter for parsing local timestamp. It ignores the time zone info of the timestamp.
+ // No pattern is defined as ISO_INSTANT format internal is not clear.
+ DateTimeParseContext dateTimestampParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_INSTANT,
+ null /* match everything*/);
+ // Formatter for parsing timestamp with time zone. The pattern is derived from ISO_LOCAL_TIME definition.
+ // Pattern asserts the string is
+ // <optional sign><Hour>:<Minute> + optional <second> + optional <fractional second>
Review Comment:
```suggestion
// Formatter for parsing a time of day. The pattern is derived from ISO_LOCAL_TIME definition.
// Pattern asserts the string is
// <optional sign><Hour>:<Minute> + optional <second> + optional <fractional second>
```
##########
hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java:
##########
@@ -187,196 +180,772 @@ private static Object convertJsonToAvroField(Object value, String name, Schema s
throw new HoodieJsonToAvroConversionException(null, name, schema, shouldSanitize, invalidCharMask);
}
- JsonToAvroFieldProcessor processor = FIELD_TYPE_PROCESSORS.get(schema.getType());
- if (null != processor) {
- return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
- }
- throw new IllegalArgumentException("JsonConverter cannot handle type: " + schema.getType());
+ return JsonToAvroFieldProcessorUtil.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
}
- /**
- * Base Class for converting json to avro fields.
- */
- private abstract static class JsonToAvroFieldProcessor implements Serializable {
+ private static class JsonToAvroFieldProcessorUtil {
+ /**
+ * Base Class for converting json to avro fields.
+ */
+ private abstract static class JsonToAvroFieldProcessor implements Serializable {
- public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
- if (!res.getLeft()) {
- throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
+ if (!res.getLeft()) {
+ throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+ return res.getRight();
}
- return res.getRight();
+
+ protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
}
- protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
- }
+ public static Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ JsonToAvroFieldProcessor processor = getProcessorForSchema(schema);
+ return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ private static JsonToAvroFieldProcessor getProcessorForSchema(Schema schema) {
+ JsonToAvroFieldProcessor processor = null;
+
+ // 3 cases to consider: customized logicalType, logicalType, and type.
+ String customizedLogicalType = schema.getProp("logicalType");
+ LogicalType logicalType = schema.getLogicalType();
+ Type type = schema.getType();
+ if (customizedLogicalType != null && !customizedLogicalType.isEmpty()) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(customizedLogicalType);
+ } else if (logicalType != null) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(logicalType.getName());
+ } else {
+ processor = AVRO_TYPE_FIELD_TYPE_PROCESSORS.get(type);
+ }
+
+ ValidationUtils.checkArgument(
+ processor != null, String.format("JsonConverter cannot handle type: %s", type));
+ return processor;
+ }
+
+ // Avro primitive and complex type processors.
+ private static final Map<Schema.Type, JsonToAvroFieldProcessor> AVRO_TYPE_FIELD_TYPE_PROCESSORS = getFieldTypeProcessors();
+ // Avro logical type processors.
+ private static final Map<String, JsonToAvroFieldProcessor> AVRO_LOGICAL_TYPE_FIELD_PROCESSORS = getLogicalFieldTypeProcessors();
+
+ /**
+ * Build type processor map for each avro type.
+ */
+ private static Map<Schema.Type, JsonToAvroFieldProcessor> getFieldTypeProcessors() {
+ Map<Schema.Type, JsonToAvroFieldProcessor> fieldTypeProcessors = new EnumMap<>(Schema.Type.class);
+ fieldTypeProcessors.put(Type.STRING, generateStringTypeHandler());
+ fieldTypeProcessors.put(Type.BOOLEAN, generateBooleanTypeHandler());
+ fieldTypeProcessors.put(Type.DOUBLE, generateDoubleTypeHandler());
+ fieldTypeProcessors.put(Type.FLOAT, generateFloatTypeHandler());
+ fieldTypeProcessors.put(Type.INT, generateIntTypeHandler());
+ fieldTypeProcessors.put(Type.LONG, generateLongTypeHandler());
+ fieldTypeProcessors.put(Type.ARRAY, generateArrayTypeHandler());
+ fieldTypeProcessors.put(Type.RECORD, generateRecordTypeHandler());
+ fieldTypeProcessors.put(Type.ENUM, generateEnumTypeHandler());
+ fieldTypeProcessors.put(Type.MAP, generateMapTypeHandler());
+ fieldTypeProcessors.put(Type.BYTES, generateBytesTypeHandler());
+ fieldTypeProcessors.put(Type.FIXED, generateFixedTypeHandler());
+ return Collections.unmodifiableMap(fieldTypeProcessors);
+ }
- private static JsonToAvroFieldProcessor generateBooleanTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ private static Map<String, JsonToAvroFieldProcessor> getLogicalFieldTypeProcessors() {
+ return CollectionUtils.createImmutableMap(
+ Pair.of(AvroLogicalTypeEnum.DECIMAL.getValue(), new DecimalLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIME_MICROS.getValue(), new TimeMicroLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIME_MILLIS.getValue(), new TimeMilliLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.DATE.getValue(), new DateLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MICROS.getValue(), new LocalTimestampMicroLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MILLIS.getValue(), new LocalTimestampMilliLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIMESTAMP_MICROS.getValue(), new TimestampMicroLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIMESTAMP_MILLIS.getValue(), new TimestampMilliLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.DURATION.getValue(), new DurationLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.UUID.getValue(), generateStringTypeHandler()));
+ }
+
+ private static class DecimalLogicalTypeProcessor extends JsonToAvroFieldProcessor {
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Boolean) {
- return Pair.of(true, value);
+
+ if (!isValidDecimalTypeConfig(schema)) {
+ return Pair.of(false, null);
+ }
+
+ // Case 1: Input is a list. It is expected to be raw Fixed byte array input, and we only support
+ // parsing it to Fixed avro type.
+ if (value instanceof List<?> && schema.getType() == Type.FIXED) {
+ JsonToAvroFieldProcessor processor = generateFixedTypeHandler();
+ return processor.convert(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ // Case 2: Input is a number or String number.
+ LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+ Pair<Boolean, BigDecimal> parseResult = parseObjectToBigDecimal(value);
+ if (Boolean.FALSE.equals(parseResult.getLeft())) {
+ return Pair.of(false, null);
+ }
+ BigDecimal bigDecimal = parseResult.getRight();
+
+ // As we don't do rounding, the validation will enforce the scale part and the integer part are all within the
+ // limit. As a result, if scale is 2 precision is 5, we only allow 3 digits for the integer.
+ // Allowed: 123.45, 123, 0.12
+ // Disallowed: 1234 (4 digit integer while the scale has already reserved 2 digit out of the 5 digit precision)
+ // 123456, 0.12345
+ if (bigDecimal.scale() > decimalType.getScale()
+ || (bigDecimal.precision() - bigDecimal.scale()) > (decimalType.getPrecision() - decimalType.getScale())) {
+ // Correspond to case
+ // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 5 as scale 2 without rounding.
+ // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 3 as scale 2 without rounding
+ return Pair.of(false, null);
+ }
+
+ switch (schema.getType()) {
+ case BYTES:
+ // Convert to primitive Arvo type that logical type Decimal uses.
+ ByteBuffer byteBuffer = new Conversions.DecimalConversion().toBytes(bigDecimal, schema, decimalType);
+ return Pair.of(true, byteBuffer);
+ case FIXED:
+ GenericFixed fixedValue = new Conversions.DecimalConversion().toFixed(bigDecimal, schema, decimalType);
+ return Pair.of(true, fixedValue);
+ default: {
+ return Pair.of(false, null);
+ }
}
- return Pair.of(false, null);
}
- };
- }
- private static JsonToAvroFieldProcessor generateIntTypeHandler() {
- return new JsonToAvroFieldProcessor() {
- @Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).intValue());
- } else if (value instanceof String) {
- return Pair.of(true, Integer.valueOf((String) value));
+ /**
+ * Check if the given schema is a valid decimal type configuration.
+ */
+ private static boolean isValidDecimalTypeConfig(Schema schema) {
+ LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+ // At the time when the schema is found not valid when it is parsed, the Avro Schema.parse will just silently
+ // set the schema to be null instead of throwing exceptions. Correspondingly, we just check if it is null here.
+ if (decimalType == null) {
+ return false;
}
- return Pair.of(false, null);
+ // Even though schema is validated at schema parsing phase, still validate here to be defensive.
+ decimalType.validate(schema);
+ return true;
}
- };
- }
- private static JsonToAvroFieldProcessor generateDoubleTypeHandler() {
- return new JsonToAvroFieldProcessor() {
- @Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).doubleValue());
- } else if (value instanceof String) {
- return Pair.of(true, Double.valueOf((String) value));
+ /**
+ * Parse the object to BigDecimal.
+ *
+ * @param obj Object to be parsed
+ * @return Pair object, with left as boolean indicating if the parsing was successful and right as the
+ * BigDecimal value.
+ */
+ private static Pair<Boolean, BigDecimal> parseObjectToBigDecimal(Object obj) {
+ // Case 1: Object is a number.
+ if (obj instanceof Number) {
+ return Pair.of(true, BigDecimal.valueOf(((Number) obj).doubleValue()));
+ }
+
+ // Case 2: Object is a number in String format.
+ if (obj instanceof String) {
+ BigDecimal bigDecimal = null;
+ try {
+ bigDecimal = new BigDecimal(((String) obj));
+ } catch (java.lang.NumberFormatException ignored) {
+ /* ignore */
+ }
+ return Pair.of(bigDecimal != null, bigDecimal);
}
return Pair.of(false, null);
}
- };
- }
+ }
- private static JsonToAvroFieldProcessor generateFloatTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ private static class DurationLogicalTypeProcessor extends JsonToAvroFieldProcessor {
+ private static final int NUM_ELEMENTS_FOR_DURATION_TYPE = 3;
+
+ /**
+ * We expect the input to be a list of 3 integers representing months, days and milliseconds.
+ */
+ private boolean isValidDurationInput(Object value) {
+ if (!(value instanceof List<?>)) {
+ return false;
+ }
+ List<?> list = (List<?>) value;
+ if (list.size() != NUM_ELEMENTS_FOR_DURATION_TYPE) {
+ return false;
+ }
+ for (Object element : list) {
+ if (!(element instanceof Integer)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Convert the given object to Avro object with schema whose logical type is duration.
+ */
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).floatValue());
- } else if (value instanceof String) {
- return Pair.of(true, Float.valueOf((String) value));
+
+ if (!isValidDurationTypeConfig(schema)) {
+ return Pair.of(false, null);
}
- return Pair.of(false, null);
+ if (!isValidDurationInput(value)) {
+ return Pair.of(false, null);
+ }
+ // After the validation the input can be safely cast to List<Integer> with 3 elements.
+ List<?> list = (List<?>) value;
+ List<Integer> converval = list.stream()
+ .filter(Integer.class::isInstance)
+ .map(Integer.class::cast)
+ .collect(Collectors.toList());
+
+ ByteBuffer buffer = ByteBuffer.allocate(schema.getFixedSize()).order(ByteOrder.LITTLE_ENDIAN);
+ for (Integer element : converval) {
+ buffer.putInt(element); // months
+ }
+ return Pair.of(true, new GenericData.Fixed(schema, buffer.array()));
+ }
+
+ /**
+ * Check if the given schema is a valid decimal type configuration.
+ */
+ private static boolean isValidDurationTypeConfig(Schema schema) {
+ String durationTypeName = AvroLogicalTypeEnum.DURATION.getValue();
+ LogicalType durationType = schema.getLogicalType();
+ String durationTypeProp = schema.getProp("logicalType");
+ // 1. The Avro type should be "Fixed".
+ // 2. Fixed size must be of 12 bytes as it hold 3 integers.
+ // 3. Logical type name should be "duration". The name might be stored in different places based on Avro version
+ // being used here.
+ return schema.getType().equals(Type.FIXED)
+ && schema.getFixedSize() == Integer.BYTES * NUM_ELEMENTS_FOR_DURATION_TYPE
+ && (durationType != null && durationType.getName().equals(durationTypeName)
+ || durationTypeProp != null && durationTypeProp.equals(durationTypeName));
+ }
+ }
+
+ /**
+ * Processor utility handling Number inputs. Consumed by TimeLogicalTypeProcessor.
+ */
+ private interface NumericParser {
+ // Convert the input number to Avro data type according to the class
+ // implementing this interface.
+ Pair<Boolean, Object> handleNumberValue(Number value);
+
+ // Convert the input number to Avro data type according to the class
+ // implementing this interface.
+ // @param value the input number in string format.
+ Pair<Boolean, Object> handleStringNumber(String value);
+
+ interface IntParser extends NumericParser {
+ @Override
+ default Pair<Boolean, Object> handleNumberValue(Number value) {
+ return Pair.of(true, value.intValue());
+ }
+
+ @Override
+ default Pair<Boolean, Object> handleStringNumber(String value) {
+ return Pair.of(true, Integer.parseInt(value));
+ }
+ }
+
+ interface LongParser extends NumericParser {
+ @Override
+ default Pair<Boolean, Object> handleNumberValue(Number value) {
+ return Pair.of(true, value.longValue());
+ }
+
+ @Override
+ default Pair<Boolean, Object> handleStringNumber(String value) {
+ return Pair.of(true, Long.parseLong(value));
+ }
+ }
+ }
+
+ /**
+ * Base Class for converting object to avro logical type TimeMilli/TimeMicro.
+ */
+ private abstract static class TimeLogicalTypeProcessor extends JsonToAvroFieldProcessor implements NumericParser {
+
+ protected static final LocalDateTime LOCAL_UNIX_EPOCH = LocalDateTime.of(1970, 1, 1, 0, 0, 0, 0);
+
+ // Logical type the processor is handling.
+ private final AvroLogicalTypeEnum logicalTypeEnum;
+
+ public TimeLogicalTypeProcessor(AvroLogicalTypeEnum logicalTypeEnum) {
+ this.logicalTypeEnum = logicalTypeEnum;
}
- };
- }
- private static JsonToAvroFieldProcessor generateLongTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ /**
+ * Main function that convert input to Object with java data type specified by schema
+ */
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ LogicalType logicalType = schema.getLogicalType();
+ if (logicalType == null) {
+ return Pair.of(false, null);
+ }
+ logicalType.validate(schema);
if (value instanceof Number) {
- return Pair.of(true, ((Number) value).longValue());
- } else if (value instanceof String) {
- return Pair.of(true, Long.valueOf((String) value));
+ return handleNumberValue((Number) value);
+ }
+ if (value instanceof String) {
+ String valStr = (String) value;
+ if (ALL_DIGITS_WITH_OPTIONAL_SIGN.matcher(valStr).matches()) {
+ return handleStringNumber(valStr);
+ } else if (isWellFormedDateTime(valStr)) {
+ return handleStringValue(valStr);
+ }
}
return Pair.of(false, null);
}
- };
- }
- private static JsonToAvroFieldProcessor generateStringTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ // Handle the case when the input is a string that may be parsed as a time.
+ protected abstract Pair<Boolean, Object> handleStringValue(String value);
+
+ protected DateTimeFormatter getDateTimeFormatter() {
+ DateTimeParseContext ctx = DATE_TIME_PARSE_CONTEXT_MAP.get(logicalTypeEnum);
+ assert ctx != null : String.format("%s should have configured date time context.", logicalTypeEnum.getValue());
+ return ctx.dateTimeFormatter;
+ }
+
+ protected Pattern getDateTimePattern() {
+ DateTimeParseContext ctx = DATE_TIME_PARSE_CONTEXT_MAP.get(logicalTypeEnum);
+ assert ctx != null : String.format("%s should have configured date time context.", logicalTypeEnum.getValue());
+ return ctx.dateTimePattern;
+ }
+
+ // Depending on the logical type the processor handles, they use different parsing context
+ // when they need to parse a timestamp string in handleStringValue.
+ private static class DateTimeParseContext {
+ public DateTimeParseContext(DateTimeFormatter dateTimeFormatter, Pattern dateTimePattern) {
+ this.dateTimeFormatter = dateTimeFormatter;
+ this.dateTimePattern = dateTimePattern;
+ }
+
+ public final Pattern dateTimePattern;
+
+ public final DateTimeFormatter dateTimeFormatter;
+ }
+
+ private static final Map<AvroLogicalTypeEnum, DateTimeParseContext> DATE_TIME_PARSE_CONTEXT_MAP = getParseContext();
+
+ private static Map<AvroLogicalTypeEnum, DateTimeParseContext> getParseContext() {
+ // Formatter for parsing local timestamp. It ignores the time zone info of the timestamp.
+ // No pattern is defined as ISO_INSTANT format internal is not clear.
+ DateTimeParseContext dateTimestampParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_INSTANT,
+ null /* match everything*/);
+ // Formatter for parsing timestamp with time zone. The pattern is derived from ISO_LOCAL_TIME definition.
+ // Pattern asserts the string is
+ // <optional sign><Hour>:<Minute> + optional <second> + optional <fractional second>
+ DateTimeParseContext dateTimeParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_LOCAL_TIME,
+ Pattern.compile("^[+-]?\\d{2}:\\d{2}(?::\\d{2}(?:\\.\\d{1,9})?)?"));
+ // Formatter for parsing local time. The pattern is derived from ISO_LOCAL_TIME definition.
+ // Pattern asserts the string is
+ // <optional sign><Year>-<Month>-<Day>T<Hour>:<Minute> + optional <second> + optional <fractional second>
+ DateTimeParseContext localTimestampParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_LOCAL_DATE_TIME,
+ Pattern.compile("^[+-]?\\d{4,10}-\\d{2}-\\d{2}T\\d{2}:\\d{2}(?::\\d{2}(?:\\.\\d{1,9})?)?")
+ );
+ // Formatter for parsing local date. The pattern is derived from ISO_LOCAL_DATE definition.
+ // Pattern asserts the string is
+ // <optional sign><Year>-<Month>-<Day>
Review Comment:
```suggestion
// Formatter for parsing a date. The pattern is derived from ISO_LOCAL_DATE definition.
// Pattern asserts the string is
// <optional sign><Year>-<Month>-<Day>
```
##########
hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java:
##########
@@ -187,196 +180,772 @@ private static Object convertJsonToAvroField(Object value, String name, Schema s
throw new HoodieJsonToAvroConversionException(null, name, schema, shouldSanitize, invalidCharMask);
}
- JsonToAvroFieldProcessor processor = FIELD_TYPE_PROCESSORS.get(schema.getType());
- if (null != processor) {
- return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
- }
- throw new IllegalArgumentException("JsonConverter cannot handle type: " + schema.getType());
+ return JsonToAvroFieldProcessorUtil.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
}
- /**
- * Base Class for converting json to avro fields.
- */
- private abstract static class JsonToAvroFieldProcessor implements Serializable {
+ private static class JsonToAvroFieldProcessorUtil {
+ /**
+ * Base Class for converting json to avro fields.
+ */
+ private abstract static class JsonToAvroFieldProcessor implements Serializable {
- public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
- if (!res.getLeft()) {
- throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
+ if (!res.getLeft()) {
+ throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+ return res.getRight();
}
- return res.getRight();
+
+ protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
}
- protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
- }
+ public static Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ JsonToAvroFieldProcessor processor = getProcessorForSchema(schema);
+ return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ private static JsonToAvroFieldProcessor getProcessorForSchema(Schema schema) {
+ JsonToAvroFieldProcessor processor = null;
+
+ // 3 cases to consider: customized logicalType, logicalType, and type.
+ String customizedLogicalType = schema.getProp("logicalType");
+ LogicalType logicalType = schema.getLogicalType();
+ Type type = schema.getType();
+ if (customizedLogicalType != null && !customizedLogicalType.isEmpty()) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(customizedLogicalType);
+ } else if (logicalType != null) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(logicalType.getName());
+ } else {
+ processor = AVRO_TYPE_FIELD_TYPE_PROCESSORS.get(type);
+ }
+
+ ValidationUtils.checkArgument(
+ processor != null, String.format("JsonConverter cannot handle type: %s", type));
+ return processor;
+ }
+
+ // Avro primitive and complex type processors.
+ private static final Map<Schema.Type, JsonToAvroFieldProcessor> AVRO_TYPE_FIELD_TYPE_PROCESSORS = getFieldTypeProcessors();
+ // Avro logical type processors.
+ private static final Map<String, JsonToAvroFieldProcessor> AVRO_LOGICAL_TYPE_FIELD_PROCESSORS = getLogicalFieldTypeProcessors();
+
+ /**
+ * Build type processor map for each avro type.
+ */
+ private static Map<Schema.Type, JsonToAvroFieldProcessor> getFieldTypeProcessors() {
+ Map<Schema.Type, JsonToAvroFieldProcessor> fieldTypeProcessors = new EnumMap<>(Schema.Type.class);
+ fieldTypeProcessors.put(Type.STRING, generateStringTypeHandler());
+ fieldTypeProcessors.put(Type.BOOLEAN, generateBooleanTypeHandler());
+ fieldTypeProcessors.put(Type.DOUBLE, generateDoubleTypeHandler());
+ fieldTypeProcessors.put(Type.FLOAT, generateFloatTypeHandler());
+ fieldTypeProcessors.put(Type.INT, generateIntTypeHandler());
+ fieldTypeProcessors.put(Type.LONG, generateLongTypeHandler());
+ fieldTypeProcessors.put(Type.ARRAY, generateArrayTypeHandler());
+ fieldTypeProcessors.put(Type.RECORD, generateRecordTypeHandler());
+ fieldTypeProcessors.put(Type.ENUM, generateEnumTypeHandler());
+ fieldTypeProcessors.put(Type.MAP, generateMapTypeHandler());
+ fieldTypeProcessors.put(Type.BYTES, generateBytesTypeHandler());
+ fieldTypeProcessors.put(Type.FIXED, generateFixedTypeHandler());
+ return Collections.unmodifiableMap(fieldTypeProcessors);
+ }
- private static JsonToAvroFieldProcessor generateBooleanTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ private static Map<String, JsonToAvroFieldProcessor> getLogicalFieldTypeProcessors() {
+ return CollectionUtils.createImmutableMap(
+ Pair.of(AvroLogicalTypeEnum.DECIMAL.getValue(), new DecimalLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIME_MICROS.getValue(), new TimeMicroLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIME_MILLIS.getValue(), new TimeMilliLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.DATE.getValue(), new DateLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MICROS.getValue(), new LocalTimestampMicroLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MILLIS.getValue(), new LocalTimestampMilliLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIMESTAMP_MICROS.getValue(), new TimestampMicroLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIMESTAMP_MILLIS.getValue(), new TimestampMilliLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.DURATION.getValue(), new DurationLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.UUID.getValue(), generateStringTypeHandler()));
+ }
+
+ private static class DecimalLogicalTypeProcessor extends JsonToAvroFieldProcessor {
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Boolean) {
- return Pair.of(true, value);
+
+ if (!isValidDecimalTypeConfig(schema)) {
+ return Pair.of(false, null);
+ }
+
+ // Case 1: Input is a list. It is expected to be raw Fixed byte array input, and we only support
+ // parsing it to Fixed avro type.
+ if (value instanceof List<?> && schema.getType() == Type.FIXED) {
+ JsonToAvroFieldProcessor processor = generateFixedTypeHandler();
+ return processor.convert(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ // Case 2: Input is a number or String number.
+ LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+ Pair<Boolean, BigDecimal> parseResult = parseObjectToBigDecimal(value);
+ if (Boolean.FALSE.equals(parseResult.getLeft())) {
+ return Pair.of(false, null);
+ }
+ BigDecimal bigDecimal = parseResult.getRight();
+
+ // As we don't do rounding, the validation will enforce the scale part and the integer part are all within the
+ // limit. As a result, if scale is 2 precision is 5, we only allow 3 digits for the integer.
+ // Allowed: 123.45, 123, 0.12
+ // Disallowed: 1234 (4 digit integer while the scale has already reserved 2 digit out of the 5 digit precision)
+ // 123456, 0.12345
+ if (bigDecimal.scale() > decimalType.getScale()
+ || (bigDecimal.precision() - bigDecimal.scale()) > (decimalType.getPrecision() - decimalType.getScale())) {
+ // Correspond to case
+ // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 5 as scale 2 without rounding.
+ // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 3 as scale 2 without rounding
+ return Pair.of(false, null);
+ }
+
+ switch (schema.getType()) {
+ case BYTES:
+ // Convert to primitive Arvo type that logical type Decimal uses.
+ ByteBuffer byteBuffer = new Conversions.DecimalConversion().toBytes(bigDecimal, schema, decimalType);
+ return Pair.of(true, byteBuffer);
+ case FIXED:
+ GenericFixed fixedValue = new Conversions.DecimalConversion().toFixed(bigDecimal, schema, decimalType);
+ return Pair.of(true, fixedValue);
+ default: {
+ return Pair.of(false, null);
+ }
}
- return Pair.of(false, null);
}
- };
- }
- private static JsonToAvroFieldProcessor generateIntTypeHandler() {
- return new JsonToAvroFieldProcessor() {
- @Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).intValue());
- } else if (value instanceof String) {
- return Pair.of(true, Integer.valueOf((String) value));
+ /**
+ * Check if the given schema is a valid decimal type configuration.
+ */
+ private static boolean isValidDecimalTypeConfig(Schema schema) {
+ LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+ // At the time when the schema is found not valid when it is parsed, the Avro Schema.parse will just silently
+ // set the schema to be null instead of throwing exceptions. Correspondingly, we just check if it is null here.
+ if (decimalType == null) {
+ return false;
}
- return Pair.of(false, null);
+ // Even though schema is validated at schema parsing phase, still validate here to be defensive.
+ decimalType.validate(schema);
+ return true;
}
- };
- }
- private static JsonToAvroFieldProcessor generateDoubleTypeHandler() {
- return new JsonToAvroFieldProcessor() {
- @Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).doubleValue());
- } else if (value instanceof String) {
- return Pair.of(true, Double.valueOf((String) value));
+ /**
+ * Parse the object to BigDecimal.
+ *
+ * @param obj Object to be parsed
+ * @return Pair object, with left as boolean indicating if the parsing was successful and right as the
+ * BigDecimal value.
+ */
+ private static Pair<Boolean, BigDecimal> parseObjectToBigDecimal(Object obj) {
+ // Case 1: Object is a number.
+ if (obj instanceof Number) {
+ return Pair.of(true, BigDecimal.valueOf(((Number) obj).doubleValue()));
+ }
+
+ // Case 2: Object is a number in String format.
+ if (obj instanceof String) {
+ BigDecimal bigDecimal = null;
+ try {
+ bigDecimal = new BigDecimal(((String) obj));
+ } catch (java.lang.NumberFormatException ignored) {
+ /* ignore */
+ }
+ return Pair.of(bigDecimal != null, bigDecimal);
}
return Pair.of(false, null);
}
- };
- }
+ }
- private static JsonToAvroFieldProcessor generateFloatTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ private static class DurationLogicalTypeProcessor extends JsonToAvroFieldProcessor {
+ private static final int NUM_ELEMENTS_FOR_DURATION_TYPE = 3;
+
+ /**
+ * We expect the input to be a list of 3 integers representing months, days and milliseconds.
+ */
+ private boolean isValidDurationInput(Object value) {
+ if (!(value instanceof List<?>)) {
+ return false;
+ }
+ List<?> list = (List<?>) value;
+ if (list.size() != NUM_ELEMENTS_FOR_DURATION_TYPE) {
+ return false;
+ }
+ for (Object element : list) {
+ if (!(element instanceof Integer)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Convert the given object to Avro object with schema whose logical type is duration.
+ */
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).floatValue());
- } else if (value instanceof String) {
- return Pair.of(true, Float.valueOf((String) value));
+
+ if (!isValidDurationTypeConfig(schema)) {
+ return Pair.of(false, null);
}
- return Pair.of(false, null);
+ if (!isValidDurationInput(value)) {
+ return Pair.of(false, null);
+ }
+ // After the validation the input can be safely cast to List<Integer> with 3 elements.
+ List<?> list = (List<?>) value;
+ List<Integer> converval = list.stream()
+ .filter(Integer.class::isInstance)
+ .map(Integer.class::cast)
+ .collect(Collectors.toList());
+
+ ByteBuffer buffer = ByteBuffer.allocate(schema.getFixedSize()).order(ByteOrder.LITTLE_ENDIAN);
+ for (Integer element : converval) {
+ buffer.putInt(element); // months
+ }
+ return Pair.of(true, new GenericData.Fixed(schema, buffer.array()));
+ }
+
+ /**
+ * Check if the given schema is a valid decimal type configuration.
+ */
+ private static boolean isValidDurationTypeConfig(Schema schema) {
+ String durationTypeName = AvroLogicalTypeEnum.DURATION.getValue();
+ LogicalType durationType = schema.getLogicalType();
+ String durationTypeProp = schema.getProp("logicalType");
+ // 1. The Avro type should be "Fixed".
+ // 2. Fixed size must be of 12 bytes as it hold 3 integers.
+ // 3. Logical type name should be "duration". The name might be stored in different places based on Avro version
+ // being used here.
+ return schema.getType().equals(Type.FIXED)
+ && schema.getFixedSize() == Integer.BYTES * NUM_ELEMENTS_FOR_DURATION_TYPE
+ && (durationType != null && durationType.getName().equals(durationTypeName)
+ || durationTypeProp != null && durationTypeProp.equals(durationTypeName));
+ }
+ }
+
+ /**
+ * Processor utility handling Number inputs. Consumed by TimeLogicalTypeProcessor.
+ */
+ private interface NumericParser {
+ // Convert the input number to Avro data type according to the class
+ // implementing this interface.
+ Pair<Boolean, Object> handleNumberValue(Number value);
+
+ // Convert the input number to Avro data type according to the class
+ // implementing this interface.
+ // @param value the input number in string format.
+ Pair<Boolean, Object> handleStringNumber(String value);
+
+ interface IntParser extends NumericParser {
+ @Override
+ default Pair<Boolean, Object> handleNumberValue(Number value) {
+ return Pair.of(true, value.intValue());
+ }
+
+ @Override
+ default Pair<Boolean, Object> handleStringNumber(String value) {
+ return Pair.of(true, Integer.parseInt(value));
+ }
+ }
+
+ interface LongParser extends NumericParser {
+ @Override
+ default Pair<Boolean, Object> handleNumberValue(Number value) {
+ return Pair.of(true, value.longValue());
+ }
+
+ @Override
+ default Pair<Boolean, Object> handleStringNumber(String value) {
+ return Pair.of(true, Long.parseLong(value));
+ }
+ }
+ }
+
+ /**
+ * Base Class for converting object to avro logical type TimeMilli/TimeMicro.
+ */
+ private abstract static class TimeLogicalTypeProcessor extends JsonToAvroFieldProcessor implements NumericParser {
+
+ protected static final LocalDateTime LOCAL_UNIX_EPOCH = LocalDateTime.of(1970, 1, 1, 0, 0, 0, 0);
+
+ // Logical type the processor is handling.
+ private final AvroLogicalTypeEnum logicalTypeEnum;
+
+ public TimeLogicalTypeProcessor(AvroLogicalTypeEnum logicalTypeEnum) {
+ this.logicalTypeEnum = logicalTypeEnum;
}
- };
- }
- private static JsonToAvroFieldProcessor generateLongTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ /**
+ * Main function that convert input to Object with java data type specified by schema
+ */
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ LogicalType logicalType = schema.getLogicalType();
+ if (logicalType == null) {
+ return Pair.of(false, null);
+ }
+ logicalType.validate(schema);
if (value instanceof Number) {
- return Pair.of(true, ((Number) value).longValue());
- } else if (value instanceof String) {
- return Pair.of(true, Long.valueOf((String) value));
+ return handleNumberValue((Number) value);
+ }
+ if (value instanceof String) {
+ String valStr = (String) value;
+ if (ALL_DIGITS_WITH_OPTIONAL_SIGN.matcher(valStr).matches()) {
+ return handleStringNumber(valStr);
+ } else if (isWellFormedDateTime(valStr)) {
+ return handleStringValue(valStr);
+ }
}
return Pair.of(false, null);
}
- };
- }
- private static JsonToAvroFieldProcessor generateStringTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ // Handle the case when the input is a string that may be parsed as a time.
+ protected abstract Pair<Boolean, Object> handleStringValue(String value);
+
+ protected DateTimeFormatter getDateTimeFormatter() {
+ DateTimeParseContext ctx = DATE_TIME_PARSE_CONTEXT_MAP.get(logicalTypeEnum);
+ assert ctx != null : String.format("%s should have configured date time context.", logicalTypeEnum.getValue());
+ return ctx.dateTimeFormatter;
+ }
+
+ protected Pattern getDateTimePattern() {
+ DateTimeParseContext ctx = DATE_TIME_PARSE_CONTEXT_MAP.get(logicalTypeEnum);
+ assert ctx != null : String.format("%s should have configured date time context.", logicalTypeEnum.getValue());
+ return ctx.dateTimePattern;
+ }
+
+ // Depending on the logical type the processor handles, they use different parsing context
+ // when they need to parse a timestamp string in handleStringValue.
+ private static class DateTimeParseContext {
+ public DateTimeParseContext(DateTimeFormatter dateTimeFormatter, Pattern dateTimePattern) {
+ this.dateTimeFormatter = dateTimeFormatter;
+ this.dateTimePattern = dateTimePattern;
+ }
+
+ public final Pattern dateTimePattern;
+
+ public final DateTimeFormatter dateTimeFormatter;
+ }
+
+ private static final Map<AvroLogicalTypeEnum, DateTimeParseContext> DATE_TIME_PARSE_CONTEXT_MAP = getParseContext();
+
+ private static Map<AvroLogicalTypeEnum, DateTimeParseContext> getParseContext() {
+ // Formatter for parsing local timestamp. It ignores the time zone info of the timestamp.
+ // No pattern is defined as ISO_INSTANT format internal is not clear.
+ DateTimeParseContext dateTimestampParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_INSTANT,
+ null /* match everything*/);
+ // Formatter for parsing timestamp with time zone. The pattern is derived from ISO_LOCAL_TIME definition.
+ // Pattern asserts the string is
+ // <optional sign><Hour>:<Minute> + optional <second> + optional <fractional second>
+ DateTimeParseContext dateTimeParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_LOCAL_TIME,
+ Pattern.compile("^[+-]?\\d{2}:\\d{2}(?::\\d{2}(?:\\.\\d{1,9})?)?"));
+ // Formatter for parsing local time. The pattern is derived from ISO_LOCAL_TIME definition.
+ // Pattern asserts the string is
+ // <optional sign><Year>-<Month>-<Day>T<Hour>:<Minute> + optional <second> + optional <fractional second>
+ DateTimeParseContext localTimestampParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_LOCAL_DATE_TIME,
+ Pattern.compile("^[+-]?\\d{4,10}-\\d{2}-\\d{2}T\\d{2}:\\d{2}(?::\\d{2}(?:\\.\\d{1,9})?)?")
+ );
+ // Formatter for parsing local date. The pattern is derived from ISO_LOCAL_DATE definition.
+ // Pattern asserts the string is
+ // <optional sign><Year>-<Month>-<Day>
+ DateTimeParseContext localDateParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_LOCAL_DATE,
+ Pattern.compile("^[+-]?\\d{4,10}-\\d{2}-\\d{2}?")
+ );
+
+ EnumMap<AvroLogicalTypeEnum, DateTimeParseContext> ctx = new EnumMap<>(AvroLogicalTypeEnum.class);
+ ctx.put(AvroLogicalTypeEnum.TIME_MICROS, dateTimeParseContext);
+ ctx.put(AvroLogicalTypeEnum.TIME_MILLIS, dateTimeParseContext);
+ ctx.put(AvroLogicalTypeEnum.DATE, localDateParseContext);
+ ctx.put(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MICROS, localTimestampParseContext);
+ ctx.put(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MILLIS, localTimestampParseContext);
+ ctx.put(AvroLogicalTypeEnum.TIMESTAMP_MICROS, dateTimestampParseContext);
+ ctx.put(AvroLogicalTypeEnum.TIMESTAMP_MILLIS, dateTimestampParseContext);
+ return Collections.unmodifiableMap(ctx);
+ }
+
+ // Pattern validating if it is an number in string form.
+ // Only check at most 19 digits as this is the max num of digits for LONG.MAX_VALUE to contain the cost of regex matching.
+ protected static final Pattern ALL_DIGITS_WITH_OPTIONAL_SIGN = Pattern.compile("^[-+]?\\d{1,19}$");
+
+ /**
+ * Check if the given string is a well-formed date time string.
+ * If no pattern is defined, it will always return true.
+ */
+ private boolean isWellFormedDateTime(String value) {
+ Pattern pattern = getDateTimePattern();
+ return pattern == null || pattern.matcher(value).matches();
+ }
+
+ protected Pair<Boolean, Instant> convertToInstantTime(String input) {
+ // Parse the input timestamp, DateTimeFormatter.ISO_INSTANT is implied here
+ Instant time = null;
+ try {
+ time = Instant.parse(input);
+ } catch (DateTimeParseException ignore) {
+ /* ignore */
+ }
+ return Pair.of(time != null, time);
+ }
+
+ protected Pair<Boolean, LocalTime> convertToLocalTime(String input) {
+ // Parse the input timestamp, DateTimeFormatter.ISO_LOCAL_TIME is implied here
+ LocalTime time = null;
+ try {
+ // Try parsing as an ISO date
+ time = LocalTime.parse(input);
+ } catch (DateTimeParseException ignore) {
+ /* ignore */
+ }
+ return Pair.of(time != null, time);
+ }
+
+ protected Pair<Boolean, LocalDateTime> convertToLocalDateTime(String input) {
+ // Parse the input timestamp, DateTimeFormatter.ISO_LOCAL_DATE_TIME is implied here
+ LocalDateTime time = null;
+ try {
+ // Try parsing as an ISO date
+ time = LocalDateTime.parse(input, getDateTimeFormatter());
+ } catch (DateTimeParseException ignore) {
+ /* ignore */
+ }
+ return Pair.of(time != null, time);
+ }
+ }
+
+ private static class DateLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.IntParser {
+ public DateLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.DATE);
+ }
+
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- return Pair.of(true, value.toString());
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, LocalDate> result = convertToLocalDate(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
+ }
+ LocalDate date = result.getRight();
+ int daysSinceEpoch = (int) ChronoUnit.DAYS.between(LocalDate.ofEpochDay(0), date);
+ return Pair.of(true, daysSinceEpoch);
+ }
+
+ private Pair<Boolean, LocalDate> convertToLocalDate(String input) {
+ // Parse the input timestamp, DateTimeFormatter.ISO_LOCAL_TIME is implied here
Review Comment:
```suggestion
// Parse the input date string, DateTimeFormatter.ISO_LOCAL_DATE is implied here
```
##########
hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java:
##########
@@ -187,196 +180,772 @@ private static Object convertJsonToAvroField(Object value, String name, Schema s
throw new HoodieJsonToAvroConversionException(null, name, schema, shouldSanitize, invalidCharMask);
}
- JsonToAvroFieldProcessor processor = FIELD_TYPE_PROCESSORS.get(schema.getType());
- if (null != processor) {
- return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
- }
- throw new IllegalArgumentException("JsonConverter cannot handle type: " + schema.getType());
+ return JsonToAvroFieldProcessorUtil.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
}
- /**
- * Base Class for converting json to avro fields.
- */
- private abstract static class JsonToAvroFieldProcessor implements Serializable {
+ private static class JsonToAvroFieldProcessorUtil {
+ /**
+ * Base Class for converting json to avro fields.
+ */
+ private abstract static class JsonToAvroFieldProcessor implements Serializable {
- public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
- if (!res.getLeft()) {
- throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
+ if (!res.getLeft()) {
+ throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+ return res.getRight();
}
- return res.getRight();
+
+ protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
}
- protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
- }
+ public static Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ JsonToAvroFieldProcessor processor = getProcessorForSchema(schema);
+ return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ private static JsonToAvroFieldProcessor getProcessorForSchema(Schema schema) {
+ JsonToAvroFieldProcessor processor = null;
+
+ // 3 cases to consider: customized logicalType, logicalType, and type.
+ String customizedLogicalType = schema.getProp("logicalType");
+ LogicalType logicalType = schema.getLogicalType();
+ Type type = schema.getType();
+ if (customizedLogicalType != null && !customizedLogicalType.isEmpty()) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(customizedLogicalType);
+ } else if (logicalType != null) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(logicalType.getName());
+ } else {
+ processor = AVRO_TYPE_FIELD_TYPE_PROCESSORS.get(type);
+ }
+
+ ValidationUtils.checkArgument(
+ processor != null, String.format("JsonConverter cannot handle type: %s", type));
+ return processor;
+ }
+
+ // Avro primitive and complex type processors.
+ private static final Map<Schema.Type, JsonToAvroFieldProcessor> AVRO_TYPE_FIELD_TYPE_PROCESSORS = getFieldTypeProcessors();
+ // Avro logical type processors.
+ private static final Map<String, JsonToAvroFieldProcessor> AVRO_LOGICAL_TYPE_FIELD_PROCESSORS = getLogicalFieldTypeProcessors();
+
+ /**
+ * Build type processor map for each avro type.
+ */
+ private static Map<Schema.Type, JsonToAvroFieldProcessor> getFieldTypeProcessors() {
+ Map<Schema.Type, JsonToAvroFieldProcessor> fieldTypeProcessors = new EnumMap<>(Schema.Type.class);
+ fieldTypeProcessors.put(Type.STRING, generateStringTypeHandler());
+ fieldTypeProcessors.put(Type.BOOLEAN, generateBooleanTypeHandler());
+ fieldTypeProcessors.put(Type.DOUBLE, generateDoubleTypeHandler());
+ fieldTypeProcessors.put(Type.FLOAT, generateFloatTypeHandler());
+ fieldTypeProcessors.put(Type.INT, generateIntTypeHandler());
+ fieldTypeProcessors.put(Type.LONG, generateLongTypeHandler());
+ fieldTypeProcessors.put(Type.ARRAY, generateArrayTypeHandler());
+ fieldTypeProcessors.put(Type.RECORD, generateRecordTypeHandler());
+ fieldTypeProcessors.put(Type.ENUM, generateEnumTypeHandler());
+ fieldTypeProcessors.put(Type.MAP, generateMapTypeHandler());
+ fieldTypeProcessors.put(Type.BYTES, generateBytesTypeHandler());
+ fieldTypeProcessors.put(Type.FIXED, generateFixedTypeHandler());
+ return Collections.unmodifiableMap(fieldTypeProcessors);
+ }
- private static JsonToAvroFieldProcessor generateBooleanTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ private static Map<String, JsonToAvroFieldProcessor> getLogicalFieldTypeProcessors() {
+ return CollectionUtils.createImmutableMap(
+ Pair.of(AvroLogicalTypeEnum.DECIMAL.getValue(), new DecimalLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIME_MICROS.getValue(), new TimeMicroLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIME_MILLIS.getValue(), new TimeMilliLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.DATE.getValue(), new DateLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MICROS.getValue(), new LocalTimestampMicroLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MILLIS.getValue(), new LocalTimestampMilliLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIMESTAMP_MICROS.getValue(), new TimestampMicroLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.TIMESTAMP_MILLIS.getValue(), new TimestampMilliLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.DURATION.getValue(), new DurationLogicalTypeProcessor()),
+ Pair.of(AvroLogicalTypeEnum.UUID.getValue(), generateStringTypeHandler()));
+ }
+
+ private static class DecimalLogicalTypeProcessor extends JsonToAvroFieldProcessor {
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Boolean) {
- return Pair.of(true, value);
+
+ if (!isValidDecimalTypeConfig(schema)) {
+ return Pair.of(false, null);
+ }
+
+ // Case 1: Input is a list. It is expected to be raw Fixed byte array input, and we only support
+ // parsing it to Fixed avro type.
+ if (value instanceof List<?> && schema.getType() == Type.FIXED) {
+ JsonToAvroFieldProcessor processor = generateFixedTypeHandler();
+ return processor.convert(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ // Case 2: Input is a number or String number.
+ LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+ Pair<Boolean, BigDecimal> parseResult = parseObjectToBigDecimal(value);
+ if (Boolean.FALSE.equals(parseResult.getLeft())) {
+ return Pair.of(false, null);
+ }
+ BigDecimal bigDecimal = parseResult.getRight();
+
+ // As we don't do rounding, the validation will enforce the scale part and the integer part are all within the
+ // limit. As a result, if scale is 2 precision is 5, we only allow 3 digits for the integer.
+ // Allowed: 123.45, 123, 0.12
+ // Disallowed: 1234 (4 digit integer while the scale has already reserved 2 digit out of the 5 digit precision)
+ // 123456, 0.12345
+ if (bigDecimal.scale() > decimalType.getScale()
+ || (bigDecimal.precision() - bigDecimal.scale()) > (decimalType.getPrecision() - decimalType.getScale())) {
+ // Correspond to case
+ // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 5 as scale 2 without rounding.
+ // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 3 as scale 2 without rounding
+ return Pair.of(false, null);
+ }
+
+ switch (schema.getType()) {
+ case BYTES:
+ // Convert to primitive Arvo type that logical type Decimal uses.
+ ByteBuffer byteBuffer = new Conversions.DecimalConversion().toBytes(bigDecimal, schema, decimalType);
+ return Pair.of(true, byteBuffer);
+ case FIXED:
+ GenericFixed fixedValue = new Conversions.DecimalConversion().toFixed(bigDecimal, schema, decimalType);
+ return Pair.of(true, fixedValue);
+ default: {
+ return Pair.of(false, null);
+ }
}
- return Pair.of(false, null);
}
- };
- }
- private static JsonToAvroFieldProcessor generateIntTypeHandler() {
- return new JsonToAvroFieldProcessor() {
- @Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).intValue());
- } else if (value instanceof String) {
- return Pair.of(true, Integer.valueOf((String) value));
+ /**
+ * Check if the given schema is a valid decimal type configuration.
+ */
+ private static boolean isValidDecimalTypeConfig(Schema schema) {
+ LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+ // At the time when the schema is found not valid when it is parsed, the Avro Schema.parse will just silently
+ // set the schema to be null instead of throwing exceptions. Correspondingly, we just check if it is null here.
+ if (decimalType == null) {
+ return false;
}
- return Pair.of(false, null);
+ // Even though schema is validated at schema parsing phase, still validate here to be defensive.
+ decimalType.validate(schema);
+ return true;
}
- };
- }
- private static JsonToAvroFieldProcessor generateDoubleTypeHandler() {
- return new JsonToAvroFieldProcessor() {
- @Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).doubleValue());
- } else if (value instanceof String) {
- return Pair.of(true, Double.valueOf((String) value));
+ /**
+ * Parse the object to BigDecimal.
+ *
+ * @param obj Object to be parsed
+ * @return Pair object, with left as boolean indicating if the parsing was successful and right as the
+ * BigDecimal value.
+ */
+ private static Pair<Boolean, BigDecimal> parseObjectToBigDecimal(Object obj) {
+ // Case 1: Object is a number.
+ if (obj instanceof Number) {
+ return Pair.of(true, BigDecimal.valueOf(((Number) obj).doubleValue()));
+ }
+
+ // Case 2: Object is a number in String format.
+ if (obj instanceof String) {
+ BigDecimal bigDecimal = null;
+ try {
+ bigDecimal = new BigDecimal(((String) obj));
+ } catch (java.lang.NumberFormatException ignored) {
+ /* ignore */
+ }
+ return Pair.of(bigDecimal != null, bigDecimal);
}
return Pair.of(false, null);
}
- };
- }
+ }
- private static JsonToAvroFieldProcessor generateFloatTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ private static class DurationLogicalTypeProcessor extends JsonToAvroFieldProcessor {
+ private static final int NUM_ELEMENTS_FOR_DURATION_TYPE = 3;
+
+ /**
+ * We expect the input to be a list of 3 integers representing months, days and milliseconds.
+ */
+ private boolean isValidDurationInput(Object value) {
+ if (!(value instanceof List<?>)) {
+ return false;
+ }
+ List<?> list = (List<?>) value;
+ if (list.size() != NUM_ELEMENTS_FOR_DURATION_TYPE) {
+ return false;
+ }
+ for (Object element : list) {
+ if (!(element instanceof Integer)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Convert the given object to Avro object with schema whose logical type is duration.
+ */
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).floatValue());
- } else if (value instanceof String) {
- return Pair.of(true, Float.valueOf((String) value));
+
+ if (!isValidDurationTypeConfig(schema)) {
+ return Pair.of(false, null);
}
- return Pair.of(false, null);
+ if (!isValidDurationInput(value)) {
+ return Pair.of(false, null);
+ }
+ // After the validation the input can be safely cast to List<Integer> with 3 elements.
+ List<?> list = (List<?>) value;
+ List<Integer> converval = list.stream()
+ .filter(Integer.class::isInstance)
+ .map(Integer.class::cast)
+ .collect(Collectors.toList());
+
+ ByteBuffer buffer = ByteBuffer.allocate(schema.getFixedSize()).order(ByteOrder.LITTLE_ENDIAN);
+ for (Integer element : converval) {
+ buffer.putInt(element); // months
+ }
+ return Pair.of(true, new GenericData.Fixed(schema, buffer.array()));
+ }
+
+ /**
+ * Check if the given schema is a valid decimal type configuration.
+ */
+ private static boolean isValidDurationTypeConfig(Schema schema) {
+ String durationTypeName = AvroLogicalTypeEnum.DURATION.getValue();
+ LogicalType durationType = schema.getLogicalType();
+ String durationTypeProp = schema.getProp("logicalType");
+ // 1. The Avro type should be "Fixed".
+ // 2. Fixed size must be of 12 bytes as it hold 3 integers.
+ // 3. Logical type name should be "duration". The name might be stored in different places based on Avro version
+ // being used here.
+ return schema.getType().equals(Type.FIXED)
+ && schema.getFixedSize() == Integer.BYTES * NUM_ELEMENTS_FOR_DURATION_TYPE
+ && (durationType != null && durationType.getName().equals(durationTypeName)
+ || durationTypeProp != null && durationTypeProp.equals(durationTypeName));
+ }
+ }
+
+ /**
+ * Processor utility handling Number inputs. Consumed by TimeLogicalTypeProcessor.
+ */
+ private interface NumericParser {
+ // Convert the input number to Avro data type according to the class
+ // implementing this interface.
+ Pair<Boolean, Object> handleNumberValue(Number value);
+
+ // Convert the input number to Avro data type according to the class
+ // implementing this interface.
+ // @param value the input number in string format.
+ Pair<Boolean, Object> handleStringNumber(String value);
+
+ interface IntParser extends NumericParser {
+ @Override
+ default Pair<Boolean, Object> handleNumberValue(Number value) {
+ return Pair.of(true, value.intValue());
+ }
+
+ @Override
+ default Pair<Boolean, Object> handleStringNumber(String value) {
+ return Pair.of(true, Integer.parseInt(value));
+ }
+ }
+
+ interface LongParser extends NumericParser {
+ @Override
+ default Pair<Boolean, Object> handleNumberValue(Number value) {
+ return Pair.of(true, value.longValue());
+ }
+
+ @Override
+ default Pair<Boolean, Object> handleStringNumber(String value) {
+ return Pair.of(true, Long.parseLong(value));
+ }
+ }
+ }
+
+ /**
+ * Base Class for converting object to avro logical type TimeMilli/TimeMicro.
+ */
+ private abstract static class TimeLogicalTypeProcessor extends JsonToAvroFieldProcessor implements NumericParser {
+
+ protected static final LocalDateTime LOCAL_UNIX_EPOCH = LocalDateTime.of(1970, 1, 1, 0, 0, 0, 0);
+
+ // Logical type the processor is handling.
+ private final AvroLogicalTypeEnum logicalTypeEnum;
+
+ public TimeLogicalTypeProcessor(AvroLogicalTypeEnum logicalTypeEnum) {
+ this.logicalTypeEnum = logicalTypeEnum;
}
- };
- }
- private static JsonToAvroFieldProcessor generateLongTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ /**
+ * Main function that convert input to Object with java data type specified by schema
+ */
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ LogicalType logicalType = schema.getLogicalType();
+ if (logicalType == null) {
+ return Pair.of(false, null);
+ }
+ logicalType.validate(schema);
if (value instanceof Number) {
- return Pair.of(true, ((Number) value).longValue());
- } else if (value instanceof String) {
- return Pair.of(true, Long.valueOf((String) value));
+ return handleNumberValue((Number) value);
+ }
+ if (value instanceof String) {
+ String valStr = (String) value;
+ if (ALL_DIGITS_WITH_OPTIONAL_SIGN.matcher(valStr).matches()) {
+ return handleStringNumber(valStr);
+ } else if (isWellFormedDateTime(valStr)) {
+ return handleStringValue(valStr);
+ }
}
return Pair.of(false, null);
}
- };
- }
- private static JsonToAvroFieldProcessor generateStringTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ // Handle the case when the input is a string that may be parsed as a time.
+ protected abstract Pair<Boolean, Object> handleStringValue(String value);
+
+ protected DateTimeFormatter getDateTimeFormatter() {
+ DateTimeParseContext ctx = DATE_TIME_PARSE_CONTEXT_MAP.get(logicalTypeEnum);
+ assert ctx != null : String.format("%s should have configured date time context.", logicalTypeEnum.getValue());
+ return ctx.dateTimeFormatter;
+ }
+
+ protected Pattern getDateTimePattern() {
+ DateTimeParseContext ctx = DATE_TIME_PARSE_CONTEXT_MAP.get(logicalTypeEnum);
+ assert ctx != null : String.format("%s should have configured date time context.", logicalTypeEnum.getValue());
+ return ctx.dateTimePattern;
+ }
+
+ // Depending on the logical type the processor handles, they use different parsing context
+ // when they need to parse a timestamp string in handleStringValue.
+ private static class DateTimeParseContext {
+ public DateTimeParseContext(DateTimeFormatter dateTimeFormatter, Pattern dateTimePattern) {
+ this.dateTimeFormatter = dateTimeFormatter;
+ this.dateTimePattern = dateTimePattern;
+ }
+
+ public final Pattern dateTimePattern;
+
+ public final DateTimeFormatter dateTimeFormatter;
+ }
+
+ private static final Map<AvroLogicalTypeEnum, DateTimeParseContext> DATE_TIME_PARSE_CONTEXT_MAP = getParseContext();
+
+ private static Map<AvroLogicalTypeEnum, DateTimeParseContext> getParseContext() {
+ // Formatter for parsing local timestamp. It ignores the time zone info of the timestamp.
+ // No pattern is defined as ISO_INSTANT format internal is not clear.
+ DateTimeParseContext dateTimestampParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_INSTANT,
+ null /* match everything*/);
+ // Formatter for parsing timestamp with time zone. The pattern is derived from ISO_LOCAL_TIME definition.
+ // Pattern asserts the string is
+ // <optional sign><Hour>:<Minute> + optional <second> + optional <fractional second>
+ DateTimeParseContext dateTimeParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_LOCAL_TIME,
+ Pattern.compile("^[+-]?\\d{2}:\\d{2}(?::\\d{2}(?:\\.\\d{1,9})?)?"));
+ // Formatter for parsing local time. The pattern is derived from ISO_LOCAL_TIME definition.
+ // Pattern asserts the string is
+ // <optional sign><Year>-<Month>-<Day>T<Hour>:<Minute> + optional <second> + optional <fractional second>
Review Comment:
```suggestion
// Formatter for parsing a timestamp in a local timezone. The pattern is derived from ISO_LOCAL_DATE_TIME definition.
// Pattern asserts the string is
// <optional sign><Year>-<Month>-<Day>T<Hour>:<Minute> + optional <second> + optional <fractional second>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "Davis-Zhang-Onehouse (via GitHub)" <gi...@apache.org>.
Davis-Zhang-Onehouse commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1608801275
##########
hudi-common/src/test/java/org/apache/hudi/avro/TestMercifulJsonConverter.java:
##########
@@ -55,6 +70,649 @@ public void basicConversion() throws IOException {
Assertions.assertEquals(rec, CONVERTER.convert(json, simpleSchema));
}
+ private static final String DECIMAL_AVRO_FILE_INVALID_PATH = "/decimal-logical-type-invalid.avsc";
+ private static final String DECIMAL_AVRO_FILE_PATH = "/decimal-logical-type.avsc";
+ private static final String DECIMAL_FIXED_AVRO_FILE_PATH = "/decimal-logical-type-fixed-type.avsc";
+ /**
+ * Covered case:
+ * Avro Logical Type: Decimal
+ * Exhaustive unsupported input coverage.
+ */
+ @ParameterizedTest
+ @MethodSource("decimalBadCases")
+ void decimalLogicalTypeInvalidCaseTest(String avroFile, String strInput, Double numInput,
Review Comment:
Yes, @the-other-tim-brown has confirmed on the type of logical types that we support. Test achieves at least 95% of branch coverage
For the code porting, it is replied in the previous thread.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2136127128
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24007",
"triggerID" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"triggerType" : "PUSH"
}, {
"hash" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24017",
"triggerID" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"triggerType" : "PUSH"
}, {
"hash" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24101",
"triggerID" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24017)
* 5699a637e30e3fafd056ab25d668640c5c8700c0 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24101)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Extend json convertor to support Avro logical type [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2120779703
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1690288658
##########
hudi-common/src/main/java/org/apache/hudi/avro/AvroLogicalTypeEnum.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.hudi.avro;
+
+public enum AvroLogicalTypeEnum {
Review Comment:
Sg
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua merged PR #11265:
URL: https://github.com/apache/hudi/pull/11265
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2214788106
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24007",
"triggerID" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"triggerType" : "PUSH"
}, {
"hash" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24017",
"triggerID" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"triggerType" : "PUSH"
}, {
"hash" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24101",
"triggerID" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"triggerType" : "PUSH"
}, {
"hash" : "5cdb05c7d909edebbc74ccf75e26c07820272090",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5cdb05c7d909edebbc74ccf75e26c07820272090",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5699a637e30e3fafd056ab25d668640c5c8700c0 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24101)
* 5cdb05c7d909edebbc74ccf75e26c07820272090 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Extend json convertor to support Avro logical type [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2120885755
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24007",
"triggerID" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24007)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2136320701
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24007",
"triggerID" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"triggerType" : "PUSH"
}, {
"hash" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24017",
"triggerID" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"triggerType" : "PUSH"
}, {
"hash" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24101",
"triggerID" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5699a637e30e3fafd056ab25d668640c5c8700c0 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24101)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "Davis-Zhang-Onehouse (via GitHub)" <gi...@apache.org>.
Davis-Zhang-Onehouse commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1608797800
##########
hudi-common/src/main/java/org/apache/hudi/avro/AvroLogicalTypeEnum.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.hudi.avro;
+
+public enum AvroLogicalTypeEnum {
Review Comment:
added comments
It is not a good idea to use the LogicalTypes because:
1. There is no enum/const string from avro side that give us an overview of all logical types that avro support by default. The LogicalTypes are java class, which are not a good choice to use class as parameters and pass it everywhere.
2. There are some logical types that the avro official doc claims have support (for example the uuid), while the avro lib of the same version does not have such logical type defined.
3. Logical types are intended for extension from client side. In the future if new customized logical type to be added, the enum approach is more friendly to be extended.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "Davis-Zhang-Onehouse (via GitHub)" <gi...@apache.org>.
Davis-Zhang-Onehouse commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1608799784
##########
hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java:
##########
@@ -187,196 +178,774 @@ private static Object convertJsonToAvroField(Object value, String name, Schema s
throw new HoodieJsonToAvroConversionException(null, name, schema, shouldSanitize, invalidCharMask);
}
- JsonToAvroFieldProcessor processor = FIELD_TYPE_PROCESSORS.get(schema.getType());
- if (null != processor) {
+ return JsonToAvroFieldProcessorUtil.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ private static class JsonToAvroFieldProcessorUtil {
Review Comment:
All the code is from the same code change that are pending rolling out to hudi internal
hudi internal PR https://github.com/onehouseinc/hudi-internal/pull/697/commits/115dcef8d681bdfd4cc1f343619669600e912623
All code is written by myself, except some part is code refactoring of some existing class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1608917941
##########
hudi-common/src/main/java/org/apache/hudi/avro/AvroLogicalTypeEnum.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.hudi.avro;
+
+public enum AvroLogicalTypeEnum {
Review Comment:
It's still good to use contants from `LogicalTypes` here.
> There are some logical types that the avro official doc claims have support (for example the uuid), while the avro lib of the same version does not have such logical type defined.
Could you add some docs and link to the spec on what logical types are supported in which versions? Hudi uses different versions of Avro depending on Spark and Flink versions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "Davis-Zhang-Onehouse (via GitHub)" <gi...@apache.org>.
Davis-Zhang-Onehouse commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1608799784
##########
hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java:
##########
@@ -187,196 +178,774 @@ private static Object convertJsonToAvroField(Object value, String name, Schema s
throw new HoodieJsonToAvroConversionException(null, name, schema, shouldSanitize, invalidCharMask);
}
- JsonToAvroFieldProcessor processor = FIELD_TYPE_PROCESSORS.get(schema.getType());
- if (null != processor) {
+ return JsonToAvroFieldProcessorUtil.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ private static class JsonToAvroFieldProcessorUtil {
Review Comment:
All code is written by myself, except some part is code refactoring of some existing class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "codope (via GitHub)" <gi...@apache.org>.
codope commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1641999059
##########
hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java:
##########
@@ -187,196 +178,774 @@ private static Object convertJsonToAvroField(Object value, String name, Schema s
throw new HoodieJsonToAvroConversionException(null, name, schema, shouldSanitize, invalidCharMask);
}
- JsonToAvroFieldProcessor processor = FIELD_TYPE_PROCESSORS.get(schema.getType());
- if (null != processor) {
+ return JsonToAvroFieldProcessorUtil.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ private static class JsonToAvroFieldProcessorUtil {
+ /**
+ * Base Class for converting json to avro fields.
+ */
+ private abstract static class JsonToAvroFieldProcessor implements Serializable {
+
+ public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
+ if (!res.getLeft()) {
+ throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+ return res.getRight();
+ }
+
+ protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
+ }
+
+ public static Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ JsonToAvroFieldProcessor processor = getProcessorForSchema(schema);
return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
}
- throw new IllegalArgumentException("JsonConverter cannot handle type: " + schema.getType());
- }
- /**
- * Base Class for converting json to avro fields.
- */
- private abstract static class JsonToAvroFieldProcessor implements Serializable {
+ private static JsonToAvroFieldProcessor getProcessorForSchema(Schema schema) {
+ JsonToAvroFieldProcessor processor = null;
+
+ // 3 cases to consider: customized logicalType, logicalType, and type.
+ String customizedLogicalType = schema.getProp("logicalType");
+ LogicalType logicalType = schema.getLogicalType();
+ Type type = schema.getType();
+ if (customizedLogicalType != null && !customizedLogicalType.isEmpty()) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(customizedLogicalType);
+ } else if (logicalType != null) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(logicalType.getName());
+ } else {
+ processor = AVRO_TYPE_FIELD_TYPE_PROCESSORS.get(type);
+ }
- public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
- if (!res.getLeft()) {
- throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ if (processor == null) {
+ throw new IllegalArgumentException(String.format("JsonConverter cannot handle type: %s", type));
Review Comment:
Use `ValidationUtils.checkArgument` or `checkState`
##########
hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java:
##########
@@ -187,196 +178,774 @@ private static Object convertJsonToAvroField(Object value, String name, Schema s
throw new HoodieJsonToAvroConversionException(null, name, schema, shouldSanitize, invalidCharMask);
}
- JsonToAvroFieldProcessor processor = FIELD_TYPE_PROCESSORS.get(schema.getType());
- if (null != processor) {
+ return JsonToAvroFieldProcessorUtil.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ private static class JsonToAvroFieldProcessorUtil {
+ /**
+ * Base Class for converting json to avro fields.
+ */
+ private abstract static class JsonToAvroFieldProcessor implements Serializable {
+
+ public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
+ if (!res.getLeft()) {
+ throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+ return res.getRight();
+ }
+
+ protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
+ }
+
+ public static Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ JsonToAvroFieldProcessor processor = getProcessorForSchema(schema);
return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
}
- throw new IllegalArgumentException("JsonConverter cannot handle type: " + schema.getType());
- }
- /**
- * Base Class for converting json to avro fields.
- */
- private abstract static class JsonToAvroFieldProcessor implements Serializable {
+ private static JsonToAvroFieldProcessor getProcessorForSchema(Schema schema) {
+ JsonToAvroFieldProcessor processor = null;
+
+ // 3 cases to consider: customized logicalType, logicalType, and type.
+ String customizedLogicalType = schema.getProp("logicalType");
+ LogicalType logicalType = schema.getLogicalType();
+ Type type = schema.getType();
+ if (customizedLogicalType != null && !customizedLogicalType.isEmpty()) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(customizedLogicalType);
+ } else if (logicalType != null) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(logicalType.getName());
+ } else {
+ processor = AVRO_TYPE_FIELD_TYPE_PROCESSORS.get(type);
+ }
- public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
- if (!res.getLeft()) {
- throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ if (processor == null) {
+ throw new IllegalArgumentException(String.format("JsonConverter cannot handle type: %s", type));
}
- return res.getRight();
+ return processor;
}
- protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
- }
+ // Avro primitive and complex type processors.
+ private static final Map<Schema.Type, JsonToAvroFieldProcessor> AVRO_TYPE_FIELD_TYPE_PROCESSORS = getFieldTypeProcessors();
+ // Avro logical type processors.
+ private static final Map<String, JsonToAvroFieldProcessor> AVRO_LOGICAL_TYPE_FIELD_PROCESSORS = getLogicalFieldTypeProcessors();
+
+ /**
+ * Build type processor map for each avro type.
+ */
+ private static Map<Schema.Type, JsonToAvroFieldProcessor> getFieldTypeProcessors() {
+ Map<Schema.Type, JsonToAvroFieldProcessor> fieldTypeProcessors = new EnumMap<>(Schema.Type.class);
+ fieldTypeProcessors.put(Type.STRING, generateStringTypeHandler());
+ fieldTypeProcessors.put(Type.BOOLEAN, generateBooleanTypeHandler());
+ fieldTypeProcessors.put(Type.DOUBLE, generateDoubleTypeHandler());
+ fieldTypeProcessors.put(Type.FLOAT, generateFloatTypeHandler());
+ fieldTypeProcessors.put(Type.INT, generateIntTypeHandler());
+ fieldTypeProcessors.put(Type.LONG, generateLongTypeHandler());
+ fieldTypeProcessors.put(Type.ARRAY, generateArrayTypeHandler());
+ fieldTypeProcessors.put(Type.RECORD, generateRecordTypeHandler());
+ fieldTypeProcessors.put(Type.ENUM, generateEnumTypeHandler());
+ fieldTypeProcessors.put(Type.MAP, generateMapTypeHandler());
+ fieldTypeProcessors.put(Type.BYTES, generateBytesTypeHandler());
+ fieldTypeProcessors.put(Type.FIXED, generateFixedTypeHandler());
+ return Collections.unmodifiableMap(fieldTypeProcessors);
+ }
- private static JsonToAvroFieldProcessor generateBooleanTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ private static Map<String, JsonToAvroFieldProcessor> getLogicalFieldTypeProcessors() {
+ Map<String, JsonToAvroFieldProcessor> logicalFieldTypeProcessors = new HashMap<>();
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.DECIMAL.getValue(), new DecimalLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.TIME_MICROS.getValue(), new TimeMicroLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.TIME_MILLIS.getValue(), new TimeMilliLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.DATE.getValue(), new DateLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MICROS.getValue(), new LocalTimestampMicroLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MILLIS.getValue(), new LocalTimestampMilliLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.TIMESTAMP_MICROS.getValue(), new TimestampMicroLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.TIMESTAMP_MILLIS.getValue(), new TimestampMilliLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.DURATION.getValue(), new DurationLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.UUID.getValue(), generateStringTypeHandler());
+ return Collections.unmodifiableMap(logicalFieldTypeProcessors);
+ }
+
+ private static class DecimalLogicalTypeProcessor extends JsonToAvroFieldProcessor {
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Boolean) {
- return Pair.of(true, value);
+
+ if (!isValidDecimalTypeConfig(schema)) {
+ return Pair.of(false, null);
+ }
+
+ // Case 1: Input is a list. It is expected to be raw Fixed byte array input, and we only support
+ // parsing it to Fixed avro type.
+ if (value instanceof List<?> && schema.getType() == Type.FIXED) {
+ JsonToAvroFieldProcessor processor = generateFixedTypeHandler();
+ return processor.convert(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ // Case 2: Input is a number or String number.
+ LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+ Pair<Boolean, BigDecimal> parseResult = parseObjectToBigDecimal(value);
+ if (Boolean.FALSE.equals(parseResult.getLeft())) {
+ return Pair.of(false, null);
+ }
+ BigDecimal bigDecimal = parseResult.getRight();
+
+ // As we don't do rounding, the validation will enforce the scale part and the integer part are all within the
+ // limit. As a result, if scale is 2 precision is 5, we only allow 3 digits for the integer.
+ // Allowed: 123.45, 123, 0.12
+ // Disallowed: 1234 (4 digit integer while the scale has already reserved 2 digit out of the 5 digit precision)
+ // 123456, 0.12345
+ if (bigDecimal.scale() > decimalType.getScale()
+ || (bigDecimal.precision() - bigDecimal.scale()) > (decimalType.getPrecision() - decimalType.getScale())) {
+ // Correspond to case
+ // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 5 as scale 2 without rounding.
+ // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 3 as scale 2 without rounding
+ return Pair.of(false, null);
+ }
+
+ switch (schema.getType()) {
+ case BYTES:
+ // Convert to primitive Arvo type that logical type Decimal uses.
+ ByteBuffer byteBuffer = new Conversions.DecimalConversion().toBytes(bigDecimal, schema, decimalType);
+ return Pair.of(true, byteBuffer);
+ case FIXED:
+ GenericFixed fixedValue = new Conversions.DecimalConversion().toFixed(bigDecimal, schema, decimalType);
+ return Pair.of(true, fixedValue);
+ default: {
+ return Pair.of(false, null);
+ }
}
- return Pair.of(false, null);
}
- };
- }
- private static JsonToAvroFieldProcessor generateIntTypeHandler() {
- return new JsonToAvroFieldProcessor() {
- @Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).intValue());
- } else if (value instanceof String) {
- return Pair.of(true, Integer.valueOf((String) value));
+ /**
+ * Check if the given schema is a valid decimal type configuration.
+ */
+ private static boolean isValidDecimalTypeConfig(Schema schema) {
+ LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+ // At the time when the schema is found not valid when it is parsed, the Avro Schema.parse will just silently
+ // set the schema to be null instead of throwing exceptions. Correspondingly, we just check if it is null here.
+ if (decimalType == null) {
+ return false;
}
- return Pair.of(false, null);
+ // Even though schema is validated at schema parsing phase, still validate here to be defensive.
+ decimalType.validate(schema);
+ return true;
}
- };
- }
- private static JsonToAvroFieldProcessor generateDoubleTypeHandler() {
- return new JsonToAvroFieldProcessor() {
- @Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).doubleValue());
- } else if (value instanceof String) {
- return Pair.of(true, Double.valueOf((String) value));
+ /**
+ * Parse the object to BigDecimal.
+ *
+ * @param obj Object to be parsed
+ * @return Pair object, with left as boolean indicating if the parsing was successful and right as the
+ * BigDecimal value.
+ */
+ private static Pair<Boolean, BigDecimal> parseObjectToBigDecimal(Object obj) {
+ // Case 1: Object is a number.
+ if (obj instanceof Number) {
+ return Pair.of(true, BigDecimal.valueOf(((Number) obj).doubleValue()));
+ }
+
+ // Case 2: Object is a number in String format.
+ if (obj instanceof String) {
+ BigDecimal bigDecimal = null;
+ try {
+ bigDecimal = new BigDecimal(((String) obj));
+ } catch (java.lang.NumberFormatException ignored) {
+ /* ignore */
+ }
+ return Pair.of(bigDecimal != null, bigDecimal);
}
return Pair.of(false, null);
}
- };
- }
+ }
+
+ private static class DurationLogicalTypeProcessor extends JsonToAvroFieldProcessor {
+ private static final int NUM_ELEMENTS_FOR_DURATION_TYPE = 3;
+
+ /**
+ * We expect the input to be a list of 3 integers representing months, days and milliseconds.
+ */
+ private boolean isValidDurationInput(Object value) {
+ if (!(value instanceof List<?>)) {
+ return false;
+ }
+ List<?> list = (List<?>) value;
+ if (list.size() != NUM_ELEMENTS_FOR_DURATION_TYPE) {
+ return false;
+ }
+ for (Object element : list) {
+ if (!(element instanceof Integer)) {
+ return false;
+ }
+ }
+ return true;
+ }
- private static JsonToAvroFieldProcessor generateFloatTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ /**
+ * Convert the given object to Avro object with schema whose logical type is duration.
+ */
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).floatValue());
- } else if (value instanceof String) {
- return Pair.of(true, Float.valueOf((String) value));
+
+ if (!isValidDurationTypeConfig(schema)) {
+ return Pair.of(false, null);
}
- return Pair.of(false, null);
+ if (!isValidDurationInput(value)) {
+ return Pair.of(false, null);
+ }
+ // After the validation the input can be safely cast to List<Integer> with 3 elements.
+ List<?> list = (List<?>) value;
+ List<Integer> converval = list.stream()
+ .filter(Integer.class::isInstance)
+ .map(Integer.class::cast)
+ .collect(Collectors.toList());
+
+ ByteBuffer buffer = ByteBuffer.allocate(schema.getFixedSize()).order(ByteOrder.LITTLE_ENDIAN);
+ for (Integer element : converval) {
+ buffer.putInt(element); // months
+ }
+ return Pair.of(true, new GenericData.Fixed(schema, buffer.array()));
+ }
+
+ /**
+ * Check if the given schema is a valid decimal type configuration.
+ */
+ private static boolean isValidDurationTypeConfig(Schema schema) {
+ String durationTypeName = AvroLogicalTypeEnum.DURATION.getValue();
+ LogicalType durationType = schema.getLogicalType();
+ String durationTypeProp = schema.getProp("logicalType");
+ // 1. The Avro type should be "Fixed".
+ // 2. Fixed size must be of 12 bytes as it hold 3 integers.
+ // 3. Logical type name should be "duration". The name might be stored in different places based on Avro version
+ // being used here.
+ return schema.getType().equals(Type.FIXED)
+ && schema.getFixedSize() == Integer.BYTES * NUM_ELEMENTS_FOR_DURATION_TYPE
+ && (durationType != null && durationType.getName().equals(durationTypeName)
+ || durationTypeProp != null && durationTypeProp.equals(durationTypeName));
+ }
+ }
+
+ /**
+ * Processor utility handling Number inputs. Consumed by TimeLogicalTypeProcessor.
+ */
+ private interface NumericParser {
+ // Convert the input number to Avro data type according to the class
+ // implementing this interface.
+ Pair<Boolean, Object> handleNumberValue(Number value);
+
+ // Convert the input number to Avro data type according to the class
+ // implementing this interface.
+ // @param value the input number in string format.
+ Pair<Boolean, Object> handleStringNumber(String value);
+
+ interface IntParser extends NumericParser {
+ @Override
+ default Pair<Boolean, Object> handleNumberValue(Number value) {
+ return Pair.of(true, value.intValue());
+ }
+
+ @Override
+ default Pair<Boolean, Object> handleStringNumber(String value) {
+ return Pair.of(true, Integer.parseInt(value));
+ }
+ }
+
+ interface LongParser extends NumericParser {
+ @Override
+ default Pair<Boolean, Object> handleNumberValue(Number value) {
+ return Pair.of(true, value.longValue());
+ }
+
+ @Override
+ default Pair<Boolean, Object> handleStringNumber(String value) {
+ return Pair.of(true, Long.parseLong(value));
+ }
+ }
+ }
+
+ /**
+ * Base Class for converting object to avro logical type TimeMilli/TimeMicro.
+ */
+ private abstract static class TimeLogicalTypeProcessor extends JsonToAvroFieldProcessor implements NumericParser {
+
+ protected static final LocalDateTime LOCAL_UNIX_EPOCH = LocalDateTime.of(1970, 1, 1, 0, 0, 0, 0);
+
+ // Logical type the processor is handling.
+ private final AvroLogicalTypeEnum logicalTypeEnum;
+
+ public TimeLogicalTypeProcessor(AvroLogicalTypeEnum logicalTypeEnum) {
+ this.logicalTypeEnum = logicalTypeEnum;
}
- };
- }
- private static JsonToAvroFieldProcessor generateLongTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ /**
+ * Main function that convert input to Object with java data type specified by schema
+ */
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ LogicalType logicalType = schema.getLogicalType();
+ if (logicalType == null) {
+ return Pair.of(false, null);
+ }
+ logicalType.validate(schema);
if (value instanceof Number) {
- return Pair.of(true, ((Number) value).longValue());
- } else if (value instanceof String) {
- return Pair.of(true, Long.valueOf((String) value));
+ return handleNumberValue((Number) value);
+ }
+ if (value instanceof String) {
+ String valStr = (String) value;
+ if (ALL_DIGITS_WITH_OPTIONAL_SIGN.matcher(valStr).matches()) {
+ return handleStringNumber(valStr);
+ } else if (isWellFormedDateTime(valStr)) {
+ return handleStringValue(valStr);
+ }
}
return Pair.of(false, null);
}
- };
- }
- private static JsonToAvroFieldProcessor generateStringTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ // Handle the case when the input is a string that may be parsed as a time.
+ protected abstract Pair<Boolean, Object> handleStringValue(String value);
+
+ protected DateTimeFormatter getDateTimeFormatter() {
+ DateTimeParseContext ctx = DATE_TIME_PARSE_CONTEXT_MAP.get(logicalTypeEnum);
+ assert ctx != null : String.format("%s should have configured date time context.", logicalTypeEnum.getValue());
+ return ctx.dateTimeFormatter;
+ }
+
+ protected Pattern getDateTimePattern() {
+ DateTimeParseContext ctx = DATE_TIME_PARSE_CONTEXT_MAP.get(logicalTypeEnum);
+ assert ctx != null : String.format("%s should have configured date time context.", logicalTypeEnum.getValue());
+ return ctx.dateTimePattern;
+ }
+
+ // Depending on the logical type the processor handles, they use different parsing context
+ // when they need to parse a timestamp string in handleStringValue.
+ private static class DateTimeParseContext {
+ public DateTimeParseContext(DateTimeFormatter dateTimeFormatter, Pattern dateTimePattern) {
+ this.dateTimeFormatter = dateTimeFormatter;
+ this.dateTimePattern = dateTimePattern;
+ }
+
+ public final Pattern dateTimePattern;
+
+ public final DateTimeFormatter dateTimeFormatter;
+ }
+
+ private static final Map<AvroLogicalTypeEnum, DateTimeParseContext> DATE_TIME_PARSE_CONTEXT_MAP = getParseContext();
+
+ private static Map<AvroLogicalTypeEnum, DateTimeParseContext> getParseContext() {
+ // Formatter for parsing local timestamp. It ignores the time zone info of the timestamp.
+ // No pattern is defined as ISO_INSTANT format internal is not clear.
+ DateTimeParseContext dateTimestampParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_INSTANT,
+ null /* match everything*/);
+ // Formatter for parsing timestamp with time zone. The pattern is derived from ISO_LOCAL_TIME definition.
+ // Pattern asserts the string is
+ // <optional sign><Hour>:<Minute> + optional <second> + optional <fractional second>
+ DateTimeParseContext dateTimeParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_LOCAL_TIME,
+ Pattern.compile("^[+-]?\\d{2}:\\d{2}(?::\\d{2}(?:\\.\\d{1,9})?)?"));
+ // Formatter for parsing local time. The pattern is derived from ISO_LOCAL_TIME definition.
+ // Pattern asserts the string is
+ // <optional sign><Year>-<Month>-<Day>T<Hour>:<Minute> + optional <second> + optional <fractional second>
+ DateTimeParseContext localTimestampParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_LOCAL_DATE_TIME,
+ Pattern.compile("^[+-]?\\d{4,10}-\\d{2}-\\d{2}T\\d{2}:\\d{2}(?::\\d{2}(?:\\.\\d{1,9})?)?")
+ );
+ // Formatter for parsing local date. The pattern is derived from ISO_LOCAL_DATE definition.
+ // Pattern asserts the string is
+ // <optional sign><Year>-<Month>-<Day>
+ DateTimeParseContext localDateParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_LOCAL_DATE,
+ Pattern.compile("^[+-]?\\d{4,10}-\\d{2}-\\d{2}?")
+ );
+
+ EnumMap<AvroLogicalTypeEnum, DateTimeParseContext> ctx = new EnumMap<>(AvroLogicalTypeEnum.class);
+ ctx.put(AvroLogicalTypeEnum.TIME_MICROS, dateTimeParseContext);
+ ctx.put(AvroLogicalTypeEnum.TIME_MILLIS, dateTimeParseContext);
+ ctx.put(AvroLogicalTypeEnum.DATE, localDateParseContext);
+ ctx.put(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MICROS, localTimestampParseContext);
+ ctx.put(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MILLIS, localTimestampParseContext);
+ ctx.put(AvroLogicalTypeEnum.TIMESTAMP_MICROS, dateTimestampParseContext);
+ ctx.put(AvroLogicalTypeEnum.TIMESTAMP_MILLIS, dateTimestampParseContext);
+ return Collections.unmodifiableMap(ctx);
+ }
+
+ // Pattern validating if it is an number in string form.
+ // Only check at most 19 digits as this is the max num of digits for LONG.MAX_VALUE to contain the cost of regex matching.
+ protected static final Pattern ALL_DIGITS_WITH_OPTIONAL_SIGN = Pattern.compile("^[-+]?\\d{1,19}$");
+
+ /**
+ * Check if the given string is a well-formed date time string.
+ * If no pattern is defined, it will always return true.
+ */
+ private boolean isWellFormedDateTime(String value) {
+ Pattern pattern = getDateTimePattern();
+ return pattern == null || pattern.matcher(value).matches();
+ }
+
+ protected Pair<Boolean, Instant> convertToInstantTime(String input) {
+ // Parse the input timestamp, DateTimeFormatter.ISO_INSTANT is implied here
+ Instant time = null;
+ try {
+ time = Instant.parse(input);
+ } catch (DateTimeParseException ignore) {
+ /* ignore */
+ }
+ return Pair.of(time != null, time);
+ }
+
+ protected Pair<Boolean, LocalTime> convertToLocalTime(String input) {
+ // Parse the input timestamp, DateTimeFormatter.ISO_LOCAL_TIME is implied here
+ LocalTime time = null;
+ try {
+ // Try parsing as an ISO date
+ time = LocalTime.parse(input);
+ } catch (DateTimeParseException ignore) {
+ /* ignore */
+ }
+ return Pair.of(time != null, time);
+ }
+
+ protected Pair<Boolean, LocalDateTime> convertToLocalDateTime(String input) {
+ // Parse the input timestamp, DateTimeFormatter.ISO_LOCAL_DATE_TIME is implied here
+ LocalDateTime time = null;
+ try {
+ // Try parsing as an ISO date
+ time = LocalDateTime.parse(input, getDateTimeFormatter());
+ } catch (DateTimeParseException ignore) {
+ /* ignore */
+ }
+ return Pair.of(time != null, time);
+ }
+ }
+
+ private static class DateLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.IntParser {
+ public DateLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.DATE);
+ }
+
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- return Pair.of(true, value.toString());
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, LocalDate> result = convertToLocalDate(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
+ }
+ LocalDate date = result.getRight();
+ int daysSinceEpoch = (int) ChronoUnit.DAYS.between(LocalDate.ofEpochDay(0), date);
+ return Pair.of(true, daysSinceEpoch);
+ }
+
+ private Pair<Boolean, LocalDate> convertToLocalDate(String input) {
+ // Parse the input timestamp, DateTimeFormatter.ISO_LOCAL_TIME is implied here
+ LocalDate date = null;
+ try {
+ // Try parsing as an ISO date
+ date = LocalDate.parse(input);
+ } catch (DateTimeParseException ignore) {
+ /* ignore */
+ }
+ return Pair.of(date != null, date);
+ }
+ }
+
+ /**
+ * Processor for TimeMilli logical type.
+ */
+ private static class TimeMilliLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.IntParser {
+ public TimeMilliLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.TIME_MILLIS);
}
- };
- }
- private static JsonToAvroFieldProcessor generateBytesTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- // Should return ByteBuffer (see GenericData.isBytes())
- return Pair.of(true, ByteBuffer.wrap(getUTF8Bytes(value.toString())));
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, LocalTime> result = convertToLocalTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
+ }
+ LocalTime time = result.getRight();
+ Integer millisOfDay = time.toSecondOfDay() * 1000 + time.getNano() / 1000000;
+ return Pair.of(true, millisOfDay);
+ }
+ }
+
+ /**
+ * Processor for TimeMicro logical type.
+ */
+ private static class TimeMicroLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.LongParser {
+ public TimeMicroLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.TIME_MICROS);
}
- };
- }
- private static JsonToAvroFieldProcessor generateFixedTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- // The ObjectMapper use List to represent FixedType
- // eg: "decimal_val": [0, 0, 14, -63, -52] will convert to ArrayList<Integer>
- List<Integer> converval = (List<Integer>) value;
- byte[] src = new byte[converval.size()];
- for (int i = 0; i < converval.size(); i++) {
- src[i] = converval.get(i).byteValue();
- }
- byte[] dst = new byte[schema.getFixedSize()];
- System.arraycopy(src, 0, dst, 0, Math.min(schema.getFixedSize(), src.length));
- return Pair.of(true, new GenericData.Fixed(schema, dst));
- }
- };
- }
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, LocalTime> result = convertToLocalTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
+ }
+ LocalTime time = result.getRight();
+ Long microsOfDay = (long) time.toSecondOfDay() * 1000000 + time.getNano() / 1000;
+ return Pair.of(true, microsOfDay);
+ }
+ }
+
+ /**
+ * Processor for TimeMicro logical type.
+ */
+ private static class LocalTimestampMicroLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.LongParser {
+ public LocalTimestampMicroLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MICROS);
+ }
- private static JsonToAvroFieldProcessor generateEnumTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (schema.getEnumSymbols().contains(value.toString())) {
- return Pair.of(true, new GenericData.EnumSymbol(schema, value.toString()));
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, LocalDateTime> result = convertToLocalDateTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
}
- throw new HoodieJsonToAvroConversionException(String.format("Symbol %s not in enum", value.toString()),
- schema.getFullName(), schema, shouldSanitize, invalidCharMask);
+ LocalDateTime time = result.getRight();
+
+ // Calculate the difference in milliseconds
+ long diffInMicros = LOCAL_UNIX_EPOCH.until(time, ChronoField.MICRO_OF_SECOND.getBaseUnit());
+ return Pair.of(true, diffInMicros);
+ }
+ }
+
+ /**
+ * Processor for TimeMicro logical type.
+ */
+ private static class TimestampMicroLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.LongParser {
+ public TimestampMicroLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.TIMESTAMP_MICROS);
}
- };
- }
- private static JsonToAvroFieldProcessor generateRecordTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- GenericRecord result = new GenericData.Record(schema);
- return Pair.of(true, convertJsonToAvro((Map<String, Object>) value, schema, shouldSanitize, invalidCharMask));
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, Instant> result = convertToInstantTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
+ }
+ Instant time = result.getRight();
+
+ // Calculate the difference in milliseconds
+ long diffInMicro = Instant.EPOCH.until(time, ChronoField.MICRO_OF_SECOND.getBaseUnit());
+ return Pair.of(true, diffInMicro);
+ }
+ }
+
+ /**
+ * Processor for TimeMicro logical type.
+ */
+ private static class LocalTimestampMilliLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.LongParser {
+ public LocalTimestampMilliLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MILLIS);
}
- };
- }
- private static JsonToAvroFieldProcessor generateArrayTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Schema elementSchema = schema.getElementType();
- List<Object> listRes = new ArrayList<>();
- for (Object v : (List) value) {
- listRes.add(convertJsonToAvroField(v, name, elementSchema, shouldSanitize, invalidCharMask));
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, LocalDateTime> result = convertToLocalDateTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
}
- return Pair.of(true, new GenericData.Array<>(schema, listRes));
+ LocalDateTime time = result.getRight();
+
+ // Calculate the difference in milliseconds
+ long diffInMillis = LOCAL_UNIX_EPOCH.until(time, ChronoField.MILLI_OF_SECOND.getBaseUnit());
+ return Pair.of(true, diffInMillis);
+ }
+ }
+
+ /**
+ * Processor for TimeMicro logical type.
+ */
+ private static class TimestampMilliLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.LongParser {
+ public TimestampMilliLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.TIMESTAMP_MILLIS);
}
- };
- }
- private static JsonToAvroFieldProcessor generateMapTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Schema valueSchema = schema.getValueType();
- Map<String, Object> mapRes = new HashMap<>();
- for (Map.Entry<String, Object> v : ((Map<String, Object>) value).entrySet()) {
- mapRes.put(v.getKey(), convertJsonToAvroField(v.getValue(), name, valueSchema, shouldSanitize, invalidCharMask));
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, Instant> result = convertToInstantTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
}
- return Pair.of(true, mapRes);
+ Instant time = result.getRight();
+
+ // Calculate the difference in milliseconds
+ long diffInMillis = Instant.EPOCH.until(time, ChronoField.MILLI_OF_SECOND.getBaseUnit());
+ return Pair.of(true, diffInMillis);
}
- };
+ }
+
+ private static JsonToAvroFieldProcessor generateBooleanTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ if (value instanceof Boolean) {
+ return Pair.of(true, value);
+ }
+ return Pair.of(false, null);
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateIntTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ if (value instanceof Number) {
+ return Pair.of(true, ((Number) value).intValue());
+ } else if (value instanceof String) {
+ return Pair.of(true, Integer.valueOf((String) value));
+ }
+ return Pair.of(false, null);
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateDoubleTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ if (value instanceof Number) {
+ return Pair.of(true, ((Number) value).doubleValue());
+ } else if (value instanceof String) {
+ return Pair.of(true, Double.valueOf((String) value));
+ }
+ return Pair.of(false, null);
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateFloatTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ if (value instanceof Number) {
+ return Pair.of(true, ((Number) value).floatValue());
+ } else if (value instanceof String) {
+ return Pair.of(true, Float.valueOf((String) value));
+ }
+ return Pair.of(false, null);
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateLongTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ if (value instanceof Number) {
+ return Pair.of(true, ((Number) value).longValue());
+ } else if (value instanceof String) {
+ return Pair.of(true, Long.valueOf((String) value));
+ }
+ return Pair.of(false, null);
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateStringTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ return Pair.of(true, value.toString());
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateBytesTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ // Should return ByteBuffer (see GenericData.isBytes())
+ return Pair.of(true, ByteBuffer.wrap(value.toString().getBytes()));
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateFixedTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ // The ObjectMapper use List to represent FixedType
+ // eg: "decimal_val": [0, 0, 14, -63, -52] will convert to ArrayList<Integer>
Review Comment:
Interesting, is this as per specification? I am curious why not List<Long>?
##########
hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java:
##########
@@ -187,196 +178,774 @@ private static Object convertJsonToAvroField(Object value, String name, Schema s
throw new HoodieJsonToAvroConversionException(null, name, schema, shouldSanitize, invalidCharMask);
}
- JsonToAvroFieldProcessor processor = FIELD_TYPE_PROCESSORS.get(schema.getType());
- if (null != processor) {
+ return JsonToAvroFieldProcessorUtil.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ private static class JsonToAvroFieldProcessorUtil {
+ /**
+ * Base Class for converting json to avro fields.
+ */
+ private abstract static class JsonToAvroFieldProcessor implements Serializable {
+
+ public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
+ if (!res.getLeft()) {
+ throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+ return res.getRight();
+ }
+
+ protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
+ }
+
+ public static Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ JsonToAvroFieldProcessor processor = getProcessorForSchema(schema);
return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
}
- throw new IllegalArgumentException("JsonConverter cannot handle type: " + schema.getType());
- }
- /**
- * Base Class for converting json to avro fields.
- */
- private abstract static class JsonToAvroFieldProcessor implements Serializable {
+ private static JsonToAvroFieldProcessor getProcessorForSchema(Schema schema) {
+ JsonToAvroFieldProcessor processor = null;
+
+ // 3 cases to consider: customized logicalType, logicalType, and type.
+ String customizedLogicalType = schema.getProp("logicalType");
+ LogicalType logicalType = schema.getLogicalType();
+ Type type = schema.getType();
+ if (customizedLogicalType != null && !customizedLogicalType.isEmpty()) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(customizedLogicalType);
+ } else if (logicalType != null) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(logicalType.getName());
+ } else {
+ processor = AVRO_TYPE_FIELD_TYPE_PROCESSORS.get(type);
+ }
- public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
- if (!res.getLeft()) {
- throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ if (processor == null) {
+ throw new IllegalArgumentException(String.format("JsonConverter cannot handle type: %s", type));
}
- return res.getRight();
+ return processor;
}
- protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
- }
+ // Avro primitive and complex type processors.
+ private static final Map<Schema.Type, JsonToAvroFieldProcessor> AVRO_TYPE_FIELD_TYPE_PROCESSORS = getFieldTypeProcessors();
+ // Avro logical type processors.
+ private static final Map<String, JsonToAvroFieldProcessor> AVRO_LOGICAL_TYPE_FIELD_PROCESSORS = getLogicalFieldTypeProcessors();
+
+ /**
+ * Build type processor map for each avro type.
+ */
+ private static Map<Schema.Type, JsonToAvroFieldProcessor> getFieldTypeProcessors() {
+ Map<Schema.Type, JsonToAvroFieldProcessor> fieldTypeProcessors = new EnumMap<>(Schema.Type.class);
+ fieldTypeProcessors.put(Type.STRING, generateStringTypeHandler());
+ fieldTypeProcessors.put(Type.BOOLEAN, generateBooleanTypeHandler());
+ fieldTypeProcessors.put(Type.DOUBLE, generateDoubleTypeHandler());
+ fieldTypeProcessors.put(Type.FLOAT, generateFloatTypeHandler());
+ fieldTypeProcessors.put(Type.INT, generateIntTypeHandler());
+ fieldTypeProcessors.put(Type.LONG, generateLongTypeHandler());
+ fieldTypeProcessors.put(Type.ARRAY, generateArrayTypeHandler());
+ fieldTypeProcessors.put(Type.RECORD, generateRecordTypeHandler());
+ fieldTypeProcessors.put(Type.ENUM, generateEnumTypeHandler());
+ fieldTypeProcessors.put(Type.MAP, generateMapTypeHandler());
+ fieldTypeProcessors.put(Type.BYTES, generateBytesTypeHandler());
+ fieldTypeProcessors.put(Type.FIXED, generateFixedTypeHandler());
+ return Collections.unmodifiableMap(fieldTypeProcessors);
+ }
- private static JsonToAvroFieldProcessor generateBooleanTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ private static Map<String, JsonToAvroFieldProcessor> getLogicalFieldTypeProcessors() {
+ Map<String, JsonToAvroFieldProcessor> logicalFieldTypeProcessors = new HashMap<>();
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.DECIMAL.getValue(), new DecimalLogicalTypeProcessor());
Review Comment:
Consider using read-only `CollectionUtils.createaImmutableMap`
```
CollectionUtils.createImmutableMap(
Pair.of(AvroLogicalTypeEnum.DECIMAL.getValue(), new DecimalLogicalTypeProcessor()),
Pair.of(AvroLogicalTypeEnum.TIME_MICROS.getValue(), new TimeMicroLogicalTypeProcessor()),
...
);
```
##########
hudi-common/src/test/resources/date-type-invalid.avsc:
##########
Review Comment:
Given that these are small schemas for each individual types, instead of all these separate avro schema files, how about simlpy declaring the schemas as static constants in test/test util. That way tests are self-contained and do not rely on external files, which can simplify running tests and also reduce test time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2249013995
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c0f09441fc00593d5e6c5f99402247a1c1dbda7a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24981",
"triggerID" : "c0f09441fc00593d5e6c5f99402247a1c1dbda7a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c0f09441fc00593d5e6c5f99402247a1c1dbda7a Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24981)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "Davis-Zhang-Onehouse (via GitHub)" <gi...@apache.org>.
Davis-Zhang-Onehouse commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2223310964
@codope is onboarded with the change. @yihua your review is required. Please take a look when you get a chance (after you got bandwidth from your current priority)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2123301071
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24007",
"triggerID" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"triggerType" : "PUSH"
}, {
"hash" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24007)
* 134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "Davis-Zhang-Onehouse (via GitHub)" <gi...@apache.org>.
Davis-Zhang-Onehouse commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1614082202
##########
hudi-common/src/main/java/org/apache/hudi/avro/AvroLogicalTypeEnum.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.hudi.avro;
+
+public enum AvroLogicalTypeEnum {
Review Comment:
addressed offline, Ethan is on the same page that we create our own Enum as Avro does not have such thing published for consumption.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2248766058
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24007",
"triggerID" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"triggerType" : "PUSH"
}, {
"hash" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24017",
"triggerID" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"triggerType" : "PUSH"
}, {
"hash" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24101",
"triggerID" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"triggerType" : "PUSH"
}, {
"hash" : "5cdb05c7d909edebbc74ccf75e26c07820272090",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24774",
"triggerID" : "5cdb05c7d909edebbc74ccf75e26c07820272090",
"triggerType" : "PUSH"
}, {
"hash" : "c0f09441fc00593d5e6c5f99402247a1c1dbda7a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c0f09441fc00593d5e6c5f99402247a1c1dbda7a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5cdb05c7d909edebbc74ccf75e26c07820272090 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24774)
* c0f09441fc00593d5e6c5f99402247a1c1dbda7a UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "Davis-Zhang-Onehouse (via GitHub)" <gi...@apache.org>.
Davis-Zhang-Onehouse commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1668961091
##########
hudi-common/src/test/resources/date-type-invalid.avsc:
##########
Review Comment:
Thanks Sagar for the comments, it might not be a good idea given:
- Each small is small, all they added up is non trivial amount of code. If we have all those raw file content in the test code, it affects the readability.
- We can add them into a separate test util, yet it is still in the form of "external dependency". The only benefits is reducing test runtime as we don't need to read from files. This is negligible benefits which I'm not sure it really worth extra dev efforts given other high priority task we are on.
- In file format it's more readable. Java just does not tackle will to make huge raw string easy for human read.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "codope (via GitHub)" <gi...@apache.org>.
codope commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1673287677
##########
hudi-common/src/test/resources/date-type-invalid.avsc:
##########
Review Comment:
ok with separate resource files, but we should be conscious as too many files like these will bloat the bundle.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Extend json convertor to support Avro logical type [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2120792311
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24007",
"triggerID" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24007)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1608735139
##########
hudi-common/src/main/java/org/apache/hudi/avro/AvroLogicalTypeEnum.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.hudi.avro;
+
+public enum AvroLogicalTypeEnum {
Review Comment:
nit: Javadocs. Does this work with all Avro spec versions?
##########
hudi-common/src/main/java/org/apache/hudi/avro/AvroLogicalTypeEnum.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.hudi.avro;
+
+public enum AvroLogicalTypeEnum {
Review Comment:
Also, Avro library defines the `LogicalTypes`. Can we use that instead of creating own enums?
https://github.com/apache/avro/blob/main/lang/java/avro/src/main/java/org/apache/avro/LogicalTypes.java
##########
hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java:
##########
@@ -187,196 +178,774 @@ private static Object convertJsonToAvroField(Object value, String name, Schema s
throw new HoodieJsonToAvroConversionException(null, name, schema, shouldSanitize, invalidCharMask);
}
- JsonToAvroFieldProcessor processor = FIELD_TYPE_PROCESSORS.get(schema.getType());
- if (null != processor) {
+ return JsonToAvroFieldProcessorUtil.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ private static class JsonToAvroFieldProcessorUtil {
Review Comment:
Are you porting the logic from another library? If so, could you reference the source?
##########
hudi-common/src/test/java/org/apache/hudi/avro/TestMercifulJsonConverter.java:
##########
@@ -55,6 +70,649 @@ public void basicConversion() throws IOException {
Assertions.assertEquals(rec, CONVERTER.convert(json, simpleSchema));
}
+ private static final String DECIMAL_AVRO_FILE_INVALID_PATH = "/decimal-logical-type-invalid.avsc";
+ private static final String DECIMAL_AVRO_FILE_PATH = "/decimal-logical-type.avsc";
+ private static final String DECIMAL_FIXED_AVRO_FILE_PATH = "/decimal-logical-type-fixed-type.avsc";
+ /**
+ * Covered case:
+ * Avro Logical Type: Decimal
+ * Exhaustive unsupported input coverage.
+ */
+ @ParameterizedTest
+ @MethodSource("decimalBadCases")
+ void decimalLogicalTypeInvalidCaseTest(String avroFile, String strInput, Double numInput,
Review Comment:
Similar for the tests on code porting. And do these cover all cases of logical types?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1608752328
##########
hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java:
##########
@@ -187,196 +178,774 @@ private static Object convertJsonToAvroField(Object value, String name, Schema s
throw new HoodieJsonToAvroConversionException(null, name, schema, shouldSanitize, invalidCharMask);
}
- JsonToAvroFieldProcessor processor = FIELD_TYPE_PROCESSORS.get(schema.getType());
- if (null != processor) {
+ return JsonToAvroFieldProcessorUtil.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ private static class JsonToAvroFieldProcessorUtil {
Review Comment:
What's the reason of porting the code?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2249006561
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c0f09441fc00593d5e6c5f99402247a1c1dbda7a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c0f09441fc00593d5e6c5f99402247a1c1dbda7a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c0f09441fc00593d5e6c5f99402247a1c1dbda7a UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "Davis-Zhang-Onehouse (via GitHub)" <gi...@apache.org>.
Davis-Zhang-Onehouse commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1608950828
##########
hudi-common/src/main/java/org/apache/hudi/avro/AvroLogicalTypeEnum.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.hudi.avro;
+
+public enum AvroLogicalTypeEnum {
Review Comment:
> It's still good to use contants from LogicalTypes here.
The only available constant is the java class itself from Avro side. If no other alternative is suggested here,
would you please expand on the benefit of using java class as parameter, which is not suggested given all the concerns listed here https://www.perplexity.ai/search/whats-the-caveats-E1jM2RV9Rj.KlAcZjQ0ZPQ#0
Java class as variables are very inconvenient to handle, expensive and dangerous.
> Could you add some docs and link
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "Davis-Zhang-Onehouse (via GitHub)" <gi...@apache.org>.
Davis-Zhang-Onehouse commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1668966899
##########
hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java:
##########
@@ -187,196 +178,774 @@ private static Object convertJsonToAvroField(Object value, String name, Schema s
throw new HoodieJsonToAvroConversionException(null, name, schema, shouldSanitize, invalidCharMask);
}
- JsonToAvroFieldProcessor processor = FIELD_TYPE_PROCESSORS.get(schema.getType());
- if (null != processor) {
+ return JsonToAvroFieldProcessorUtil.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ private static class JsonToAvroFieldProcessorUtil {
+ /**
+ * Base Class for converting json to avro fields.
+ */
+ private abstract static class JsonToAvroFieldProcessor implements Serializable {
+
+ public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
+ if (!res.getLeft()) {
+ throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+ return res.getRight();
+ }
+
+ protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
+ }
+
+ public static Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ JsonToAvroFieldProcessor processor = getProcessorForSchema(schema);
return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
}
- throw new IllegalArgumentException("JsonConverter cannot handle type: " + schema.getType());
- }
- /**
- * Base Class for converting json to avro fields.
- */
- private abstract static class JsonToAvroFieldProcessor implements Serializable {
+ private static JsonToAvroFieldProcessor getProcessorForSchema(Schema schema) {
+ JsonToAvroFieldProcessor processor = null;
+
+ // 3 cases to consider: customized logicalType, logicalType, and type.
+ String customizedLogicalType = schema.getProp("logicalType");
+ LogicalType logicalType = schema.getLogicalType();
+ Type type = schema.getType();
+ if (customizedLogicalType != null && !customizedLogicalType.isEmpty()) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(customizedLogicalType);
+ } else if (logicalType != null) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(logicalType.getName());
+ } else {
+ processor = AVRO_TYPE_FIELD_TYPE_PROCESSORS.get(type);
+ }
- public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
- if (!res.getLeft()) {
- throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ if (processor == null) {
+ throw new IllegalArgumentException(String.format("JsonConverter cannot handle type: %s", type));
}
- return res.getRight();
+ return processor;
}
- protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
- }
+ // Avro primitive and complex type processors.
+ private static final Map<Schema.Type, JsonToAvroFieldProcessor> AVRO_TYPE_FIELD_TYPE_PROCESSORS = getFieldTypeProcessors();
+ // Avro logical type processors.
+ private static final Map<String, JsonToAvroFieldProcessor> AVRO_LOGICAL_TYPE_FIELD_PROCESSORS = getLogicalFieldTypeProcessors();
+
+ /**
+ * Build type processor map for each avro type.
+ */
+ private static Map<Schema.Type, JsonToAvroFieldProcessor> getFieldTypeProcessors() {
+ Map<Schema.Type, JsonToAvroFieldProcessor> fieldTypeProcessors = new EnumMap<>(Schema.Type.class);
+ fieldTypeProcessors.put(Type.STRING, generateStringTypeHandler());
+ fieldTypeProcessors.put(Type.BOOLEAN, generateBooleanTypeHandler());
+ fieldTypeProcessors.put(Type.DOUBLE, generateDoubleTypeHandler());
+ fieldTypeProcessors.put(Type.FLOAT, generateFloatTypeHandler());
+ fieldTypeProcessors.put(Type.INT, generateIntTypeHandler());
+ fieldTypeProcessors.put(Type.LONG, generateLongTypeHandler());
+ fieldTypeProcessors.put(Type.ARRAY, generateArrayTypeHandler());
+ fieldTypeProcessors.put(Type.RECORD, generateRecordTypeHandler());
+ fieldTypeProcessors.put(Type.ENUM, generateEnumTypeHandler());
+ fieldTypeProcessors.put(Type.MAP, generateMapTypeHandler());
+ fieldTypeProcessors.put(Type.BYTES, generateBytesTypeHandler());
+ fieldTypeProcessors.put(Type.FIXED, generateFixedTypeHandler());
+ return Collections.unmodifiableMap(fieldTypeProcessors);
+ }
- private static JsonToAvroFieldProcessor generateBooleanTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ private static Map<String, JsonToAvroFieldProcessor> getLogicalFieldTypeProcessors() {
+ Map<String, JsonToAvroFieldProcessor> logicalFieldTypeProcessors = new HashMap<>();
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.DECIMAL.getValue(), new DecimalLogicalTypeProcessor());
Review Comment:
done
##########
hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java:
##########
@@ -187,196 +178,774 @@ private static Object convertJsonToAvroField(Object value, String name, Schema s
throw new HoodieJsonToAvroConversionException(null, name, schema, shouldSanitize, invalidCharMask);
}
- JsonToAvroFieldProcessor processor = FIELD_TYPE_PROCESSORS.get(schema.getType());
- if (null != processor) {
+ return JsonToAvroFieldProcessorUtil.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ private static class JsonToAvroFieldProcessorUtil {
+ /**
+ * Base Class for converting json to avro fields.
+ */
+ private abstract static class JsonToAvroFieldProcessor implements Serializable {
+
+ public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
+ if (!res.getLeft()) {
+ throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+ return res.getRight();
+ }
+
+ protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
+ }
+
+ public static Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ JsonToAvroFieldProcessor processor = getProcessorForSchema(schema);
return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
}
- throw new IllegalArgumentException("JsonConverter cannot handle type: " + schema.getType());
- }
- /**
- * Base Class for converting json to avro fields.
- */
- private abstract static class JsonToAvroFieldProcessor implements Serializable {
+ private static JsonToAvroFieldProcessor getProcessorForSchema(Schema schema) {
+ JsonToAvroFieldProcessor processor = null;
+
+ // 3 cases to consider: customized logicalType, logicalType, and type.
+ String customizedLogicalType = schema.getProp("logicalType");
+ LogicalType logicalType = schema.getLogicalType();
+ Type type = schema.getType();
+ if (customizedLogicalType != null && !customizedLogicalType.isEmpty()) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(customizedLogicalType);
+ } else if (logicalType != null) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(logicalType.getName());
+ } else {
+ processor = AVRO_TYPE_FIELD_TYPE_PROCESSORS.get(type);
+ }
- public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
- if (!res.getLeft()) {
- throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ if (processor == null) {
+ throw new IllegalArgumentException(String.format("JsonConverter cannot handle type: %s", type));
Review Comment:
done
##########
hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java:
##########
@@ -187,196 +178,774 @@ private static Object convertJsonToAvroField(Object value, String name, Schema s
throw new HoodieJsonToAvroConversionException(null, name, schema, shouldSanitize, invalidCharMask);
}
- JsonToAvroFieldProcessor processor = FIELD_TYPE_PROCESSORS.get(schema.getType());
- if (null != processor) {
+ return JsonToAvroFieldProcessorUtil.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ private static class JsonToAvroFieldProcessorUtil {
+ /**
+ * Base Class for converting json to avro fields.
+ */
+ private abstract static class JsonToAvroFieldProcessor implements Serializable {
+
+ public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
+ if (!res.getLeft()) {
+ throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+ return res.getRight();
+ }
+
+ protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
+ }
+
+ public static Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ JsonToAvroFieldProcessor processor = getProcessorForSchema(schema);
return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
}
- throw new IllegalArgumentException("JsonConverter cannot handle type: " + schema.getType());
- }
- /**
- * Base Class for converting json to avro fields.
- */
- private abstract static class JsonToAvroFieldProcessor implements Serializable {
+ private static JsonToAvroFieldProcessor getProcessorForSchema(Schema schema) {
+ JsonToAvroFieldProcessor processor = null;
+
+ // 3 cases to consider: customized logicalType, logicalType, and type.
+ String customizedLogicalType = schema.getProp("logicalType");
+ LogicalType logicalType = schema.getLogicalType();
+ Type type = schema.getType();
+ if (customizedLogicalType != null && !customizedLogicalType.isEmpty()) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(customizedLogicalType);
+ } else if (logicalType != null) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(logicalType.getName());
+ } else {
+ processor = AVRO_TYPE_FIELD_TYPE_PROCESSORS.get(type);
+ }
- public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
- if (!res.getLeft()) {
- throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ if (processor == null) {
+ throw new IllegalArgumentException(String.format("JsonConverter cannot handle type: %s", type));
}
- return res.getRight();
+ return processor;
}
- protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
- }
+ // Avro primitive and complex type processors.
+ private static final Map<Schema.Type, JsonToAvroFieldProcessor> AVRO_TYPE_FIELD_TYPE_PROCESSORS = getFieldTypeProcessors();
+ // Avro logical type processors.
+ private static final Map<String, JsonToAvroFieldProcessor> AVRO_LOGICAL_TYPE_FIELD_PROCESSORS = getLogicalFieldTypeProcessors();
+
+ /**
+ * Build type processor map for each avro type.
+ */
+ private static Map<Schema.Type, JsonToAvroFieldProcessor> getFieldTypeProcessors() {
+ Map<Schema.Type, JsonToAvroFieldProcessor> fieldTypeProcessors = new EnumMap<>(Schema.Type.class);
+ fieldTypeProcessors.put(Type.STRING, generateStringTypeHandler());
+ fieldTypeProcessors.put(Type.BOOLEAN, generateBooleanTypeHandler());
+ fieldTypeProcessors.put(Type.DOUBLE, generateDoubleTypeHandler());
+ fieldTypeProcessors.put(Type.FLOAT, generateFloatTypeHandler());
+ fieldTypeProcessors.put(Type.INT, generateIntTypeHandler());
+ fieldTypeProcessors.put(Type.LONG, generateLongTypeHandler());
+ fieldTypeProcessors.put(Type.ARRAY, generateArrayTypeHandler());
+ fieldTypeProcessors.put(Type.RECORD, generateRecordTypeHandler());
+ fieldTypeProcessors.put(Type.ENUM, generateEnumTypeHandler());
+ fieldTypeProcessors.put(Type.MAP, generateMapTypeHandler());
+ fieldTypeProcessors.put(Type.BYTES, generateBytesTypeHandler());
+ fieldTypeProcessors.put(Type.FIXED, generateFixedTypeHandler());
+ return Collections.unmodifiableMap(fieldTypeProcessors);
+ }
- private static JsonToAvroFieldProcessor generateBooleanTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ private static Map<String, JsonToAvroFieldProcessor> getLogicalFieldTypeProcessors() {
+ Map<String, JsonToAvroFieldProcessor> logicalFieldTypeProcessors = new HashMap<>();
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.DECIMAL.getValue(), new DecimalLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.TIME_MICROS.getValue(), new TimeMicroLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.TIME_MILLIS.getValue(), new TimeMilliLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.DATE.getValue(), new DateLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MICROS.getValue(), new LocalTimestampMicroLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MILLIS.getValue(), new LocalTimestampMilliLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.TIMESTAMP_MICROS.getValue(), new TimestampMicroLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.TIMESTAMP_MILLIS.getValue(), new TimestampMilliLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.DURATION.getValue(), new DurationLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.UUID.getValue(), generateStringTypeHandler());
+ return Collections.unmodifiableMap(logicalFieldTypeProcessors);
+ }
+
+ private static class DecimalLogicalTypeProcessor extends JsonToAvroFieldProcessor {
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Boolean) {
- return Pair.of(true, value);
+
+ if (!isValidDecimalTypeConfig(schema)) {
+ return Pair.of(false, null);
+ }
+
+ // Case 1: Input is a list. It is expected to be raw Fixed byte array input, and we only support
+ // parsing it to Fixed avro type.
+ if (value instanceof List<?> && schema.getType() == Type.FIXED) {
+ JsonToAvroFieldProcessor processor = generateFixedTypeHandler();
+ return processor.convert(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ // Case 2: Input is a number or String number.
+ LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+ Pair<Boolean, BigDecimal> parseResult = parseObjectToBigDecimal(value);
+ if (Boolean.FALSE.equals(parseResult.getLeft())) {
+ return Pair.of(false, null);
+ }
+ BigDecimal bigDecimal = parseResult.getRight();
+
+ // As we don't do rounding, the validation will enforce the scale part and the integer part are all within the
+ // limit. As a result, if scale is 2 precision is 5, we only allow 3 digits for the integer.
+ // Allowed: 123.45, 123, 0.12
+ // Disallowed: 1234 (4 digit integer while the scale has already reserved 2 digit out of the 5 digit precision)
+ // 123456, 0.12345
+ if (bigDecimal.scale() > decimalType.getScale()
+ || (bigDecimal.precision() - bigDecimal.scale()) > (decimalType.getPrecision() - decimalType.getScale())) {
+ // Correspond to case
+ // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 5 as scale 2 without rounding.
+ // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 3 as scale 2 without rounding
+ return Pair.of(false, null);
+ }
+
+ switch (schema.getType()) {
+ case BYTES:
+ // Convert to primitive Arvo type that logical type Decimal uses.
+ ByteBuffer byteBuffer = new Conversions.DecimalConversion().toBytes(bigDecimal, schema, decimalType);
+ return Pair.of(true, byteBuffer);
+ case FIXED:
+ GenericFixed fixedValue = new Conversions.DecimalConversion().toFixed(bigDecimal, schema, decimalType);
+ return Pair.of(true, fixedValue);
+ default: {
+ return Pair.of(false, null);
+ }
}
- return Pair.of(false, null);
}
- };
- }
- private static JsonToAvroFieldProcessor generateIntTypeHandler() {
- return new JsonToAvroFieldProcessor() {
- @Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).intValue());
- } else if (value instanceof String) {
- return Pair.of(true, Integer.valueOf((String) value));
+ /**
+ * Check if the given schema is a valid decimal type configuration.
+ */
+ private static boolean isValidDecimalTypeConfig(Schema schema) {
+ LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+ // At the time when the schema is found not valid when it is parsed, the Avro Schema.parse will just silently
+ // set the schema to be null instead of throwing exceptions. Correspondingly, we just check if it is null here.
+ if (decimalType == null) {
+ return false;
}
- return Pair.of(false, null);
+ // Even though schema is validated at schema parsing phase, still validate here to be defensive.
+ decimalType.validate(schema);
+ return true;
}
- };
- }
- private static JsonToAvroFieldProcessor generateDoubleTypeHandler() {
- return new JsonToAvroFieldProcessor() {
- @Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).doubleValue());
- } else if (value instanceof String) {
- return Pair.of(true, Double.valueOf((String) value));
+ /**
+ * Parse the object to BigDecimal.
+ *
+ * @param obj Object to be parsed
+ * @return Pair object, with left as boolean indicating if the parsing was successful and right as the
+ * BigDecimal value.
+ */
+ private static Pair<Boolean, BigDecimal> parseObjectToBigDecimal(Object obj) {
+ // Case 1: Object is a number.
+ if (obj instanceof Number) {
+ return Pair.of(true, BigDecimal.valueOf(((Number) obj).doubleValue()));
+ }
+
+ // Case 2: Object is a number in String format.
+ if (obj instanceof String) {
+ BigDecimal bigDecimal = null;
+ try {
+ bigDecimal = new BigDecimal(((String) obj));
+ } catch (java.lang.NumberFormatException ignored) {
+ /* ignore */
+ }
+ return Pair.of(bigDecimal != null, bigDecimal);
}
return Pair.of(false, null);
}
- };
- }
+ }
+
+ private static class DurationLogicalTypeProcessor extends JsonToAvroFieldProcessor {
+ private static final int NUM_ELEMENTS_FOR_DURATION_TYPE = 3;
+
+ /**
+ * We expect the input to be a list of 3 integers representing months, days and milliseconds.
+ */
+ private boolean isValidDurationInput(Object value) {
+ if (!(value instanceof List<?>)) {
+ return false;
+ }
+ List<?> list = (List<?>) value;
+ if (list.size() != NUM_ELEMENTS_FOR_DURATION_TYPE) {
+ return false;
+ }
+ for (Object element : list) {
+ if (!(element instanceof Integer)) {
+ return false;
+ }
+ }
+ return true;
+ }
- private static JsonToAvroFieldProcessor generateFloatTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ /**
+ * Convert the given object to Avro object with schema whose logical type is duration.
+ */
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).floatValue());
- } else if (value instanceof String) {
- return Pair.of(true, Float.valueOf((String) value));
+
+ if (!isValidDurationTypeConfig(schema)) {
+ return Pair.of(false, null);
}
- return Pair.of(false, null);
+ if (!isValidDurationInput(value)) {
+ return Pair.of(false, null);
+ }
+ // After the validation the input can be safely cast to List<Integer> with 3 elements.
+ List<?> list = (List<?>) value;
+ List<Integer> converval = list.stream()
+ .filter(Integer.class::isInstance)
+ .map(Integer.class::cast)
+ .collect(Collectors.toList());
+
+ ByteBuffer buffer = ByteBuffer.allocate(schema.getFixedSize()).order(ByteOrder.LITTLE_ENDIAN);
+ for (Integer element : converval) {
+ buffer.putInt(element); // months
+ }
+ return Pair.of(true, new GenericData.Fixed(schema, buffer.array()));
+ }
+
+ /**
+ * Check if the given schema is a valid decimal type configuration.
+ */
+ private static boolean isValidDurationTypeConfig(Schema schema) {
+ String durationTypeName = AvroLogicalTypeEnum.DURATION.getValue();
+ LogicalType durationType = schema.getLogicalType();
+ String durationTypeProp = schema.getProp("logicalType");
+ // 1. The Avro type should be "Fixed".
+ // 2. Fixed size must be of 12 bytes as it hold 3 integers.
+ // 3. Logical type name should be "duration". The name might be stored in different places based on Avro version
+ // being used here.
+ return schema.getType().equals(Type.FIXED)
+ && schema.getFixedSize() == Integer.BYTES * NUM_ELEMENTS_FOR_DURATION_TYPE
+ && (durationType != null && durationType.getName().equals(durationTypeName)
+ || durationTypeProp != null && durationTypeProp.equals(durationTypeName));
+ }
+ }
+
+ /**
+ * Processor utility handling Number inputs. Consumed by TimeLogicalTypeProcessor.
+ */
+ private interface NumericParser {
+ // Convert the input number to Avro data type according to the class
+ // implementing this interface.
+ Pair<Boolean, Object> handleNumberValue(Number value);
+
+ // Convert the input number to Avro data type according to the class
+ // implementing this interface.
+ // @param value the input number in string format.
+ Pair<Boolean, Object> handleStringNumber(String value);
+
+ interface IntParser extends NumericParser {
+ @Override
+ default Pair<Boolean, Object> handleNumberValue(Number value) {
+ return Pair.of(true, value.intValue());
+ }
+
+ @Override
+ default Pair<Boolean, Object> handleStringNumber(String value) {
+ return Pair.of(true, Integer.parseInt(value));
+ }
+ }
+
+ interface LongParser extends NumericParser {
+ @Override
+ default Pair<Boolean, Object> handleNumberValue(Number value) {
+ return Pair.of(true, value.longValue());
+ }
+
+ @Override
+ default Pair<Boolean, Object> handleStringNumber(String value) {
+ return Pair.of(true, Long.parseLong(value));
+ }
+ }
+ }
+
+ /**
+ * Base Class for converting object to avro logical type TimeMilli/TimeMicro.
+ */
+ private abstract static class TimeLogicalTypeProcessor extends JsonToAvroFieldProcessor implements NumericParser {
+
+ protected static final LocalDateTime LOCAL_UNIX_EPOCH = LocalDateTime.of(1970, 1, 1, 0, 0, 0, 0);
+
+ // Logical type the processor is handling.
+ private final AvroLogicalTypeEnum logicalTypeEnum;
+
+ public TimeLogicalTypeProcessor(AvroLogicalTypeEnum logicalTypeEnum) {
+ this.logicalTypeEnum = logicalTypeEnum;
}
- };
- }
- private static JsonToAvroFieldProcessor generateLongTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ /**
+ * Main function that convert input to Object with java data type specified by schema
+ */
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ LogicalType logicalType = schema.getLogicalType();
+ if (logicalType == null) {
+ return Pair.of(false, null);
+ }
+ logicalType.validate(schema);
if (value instanceof Number) {
- return Pair.of(true, ((Number) value).longValue());
- } else if (value instanceof String) {
- return Pair.of(true, Long.valueOf((String) value));
+ return handleNumberValue((Number) value);
+ }
+ if (value instanceof String) {
+ String valStr = (String) value;
+ if (ALL_DIGITS_WITH_OPTIONAL_SIGN.matcher(valStr).matches()) {
+ return handleStringNumber(valStr);
+ } else if (isWellFormedDateTime(valStr)) {
+ return handleStringValue(valStr);
+ }
}
return Pair.of(false, null);
}
- };
- }
- private static JsonToAvroFieldProcessor generateStringTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ // Handle the case when the input is a string that may be parsed as a time.
+ protected abstract Pair<Boolean, Object> handleStringValue(String value);
+
+ protected DateTimeFormatter getDateTimeFormatter() {
+ DateTimeParseContext ctx = DATE_TIME_PARSE_CONTEXT_MAP.get(logicalTypeEnum);
+ assert ctx != null : String.format("%s should have configured date time context.", logicalTypeEnum.getValue());
+ return ctx.dateTimeFormatter;
+ }
+
+ protected Pattern getDateTimePattern() {
+ DateTimeParseContext ctx = DATE_TIME_PARSE_CONTEXT_MAP.get(logicalTypeEnum);
+ assert ctx != null : String.format("%s should have configured date time context.", logicalTypeEnum.getValue());
+ return ctx.dateTimePattern;
+ }
+
+ // Depending on the logical type the processor handles, they use different parsing context
+ // when they need to parse a timestamp string in handleStringValue.
+ private static class DateTimeParseContext {
+ public DateTimeParseContext(DateTimeFormatter dateTimeFormatter, Pattern dateTimePattern) {
+ this.dateTimeFormatter = dateTimeFormatter;
+ this.dateTimePattern = dateTimePattern;
+ }
+
+ public final Pattern dateTimePattern;
+
+ public final DateTimeFormatter dateTimeFormatter;
+ }
+
+ private static final Map<AvroLogicalTypeEnum, DateTimeParseContext> DATE_TIME_PARSE_CONTEXT_MAP = getParseContext();
+
+ private static Map<AvroLogicalTypeEnum, DateTimeParseContext> getParseContext() {
+ // Formatter for parsing local timestamp. It ignores the time zone info of the timestamp.
+ // No pattern is defined as ISO_INSTANT format internal is not clear.
+ DateTimeParseContext dateTimestampParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_INSTANT,
+ null /* match everything*/);
+ // Formatter for parsing timestamp with time zone. The pattern is derived from ISO_LOCAL_TIME definition.
+ // Pattern asserts the string is
+ // <optional sign><Hour>:<Minute> + optional <second> + optional <fractional second>
+ DateTimeParseContext dateTimeParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_LOCAL_TIME,
+ Pattern.compile("^[+-]?\\d{2}:\\d{2}(?::\\d{2}(?:\\.\\d{1,9})?)?"));
+ // Formatter for parsing local time. The pattern is derived from ISO_LOCAL_TIME definition.
+ // Pattern asserts the string is
+ // <optional sign><Year>-<Month>-<Day>T<Hour>:<Minute> + optional <second> + optional <fractional second>
+ DateTimeParseContext localTimestampParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_LOCAL_DATE_TIME,
+ Pattern.compile("^[+-]?\\d{4,10}-\\d{2}-\\d{2}T\\d{2}:\\d{2}(?::\\d{2}(?:\\.\\d{1,9})?)?")
+ );
+ // Formatter for parsing local date. The pattern is derived from ISO_LOCAL_DATE definition.
+ // Pattern asserts the string is
+ // <optional sign><Year>-<Month>-<Day>
+ DateTimeParseContext localDateParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_LOCAL_DATE,
+ Pattern.compile("^[+-]?\\d{4,10}-\\d{2}-\\d{2}?")
+ );
+
+ EnumMap<AvroLogicalTypeEnum, DateTimeParseContext> ctx = new EnumMap<>(AvroLogicalTypeEnum.class);
+ ctx.put(AvroLogicalTypeEnum.TIME_MICROS, dateTimeParseContext);
+ ctx.put(AvroLogicalTypeEnum.TIME_MILLIS, dateTimeParseContext);
+ ctx.put(AvroLogicalTypeEnum.DATE, localDateParseContext);
+ ctx.put(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MICROS, localTimestampParseContext);
+ ctx.put(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MILLIS, localTimestampParseContext);
+ ctx.put(AvroLogicalTypeEnum.TIMESTAMP_MICROS, dateTimestampParseContext);
+ ctx.put(AvroLogicalTypeEnum.TIMESTAMP_MILLIS, dateTimestampParseContext);
+ return Collections.unmodifiableMap(ctx);
+ }
+
+ // Pattern validating if it is an number in string form.
+ // Only check at most 19 digits as this is the max num of digits for LONG.MAX_VALUE to contain the cost of regex matching.
+ protected static final Pattern ALL_DIGITS_WITH_OPTIONAL_SIGN = Pattern.compile("^[-+]?\\d{1,19}$");
+
+ /**
+ * Check if the given string is a well-formed date time string.
+ * If no pattern is defined, it will always return true.
+ */
+ private boolean isWellFormedDateTime(String value) {
+ Pattern pattern = getDateTimePattern();
+ return pattern == null || pattern.matcher(value).matches();
+ }
+
+ protected Pair<Boolean, Instant> convertToInstantTime(String input) {
+ // Parse the input timestamp, DateTimeFormatter.ISO_INSTANT is implied here
+ Instant time = null;
+ try {
+ time = Instant.parse(input);
+ } catch (DateTimeParseException ignore) {
+ /* ignore */
+ }
+ return Pair.of(time != null, time);
+ }
+
+ protected Pair<Boolean, LocalTime> convertToLocalTime(String input) {
+ // Parse the input timestamp, DateTimeFormatter.ISO_LOCAL_TIME is implied here
+ LocalTime time = null;
+ try {
+ // Try parsing as an ISO date
+ time = LocalTime.parse(input);
+ } catch (DateTimeParseException ignore) {
+ /* ignore */
+ }
+ return Pair.of(time != null, time);
+ }
+
+ protected Pair<Boolean, LocalDateTime> convertToLocalDateTime(String input) {
+ // Parse the input timestamp, DateTimeFormatter.ISO_LOCAL_DATE_TIME is implied here
+ LocalDateTime time = null;
+ try {
+ // Try parsing as an ISO date
+ time = LocalDateTime.parse(input, getDateTimeFormatter());
+ } catch (DateTimeParseException ignore) {
+ /* ignore */
+ }
+ return Pair.of(time != null, time);
+ }
+ }
+
+ private static class DateLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.IntParser {
+ public DateLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.DATE);
+ }
+
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- return Pair.of(true, value.toString());
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, LocalDate> result = convertToLocalDate(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
+ }
+ LocalDate date = result.getRight();
+ int daysSinceEpoch = (int) ChronoUnit.DAYS.between(LocalDate.ofEpochDay(0), date);
+ return Pair.of(true, daysSinceEpoch);
+ }
+
+ private Pair<Boolean, LocalDate> convertToLocalDate(String input) {
+ // Parse the input timestamp, DateTimeFormatter.ISO_LOCAL_TIME is implied here
+ LocalDate date = null;
+ try {
+ // Try parsing as an ISO date
+ date = LocalDate.parse(input);
+ } catch (DateTimeParseException ignore) {
+ /* ignore */
+ }
+ return Pair.of(date != null, date);
+ }
+ }
+
+ /**
+ * Processor for TimeMilli logical type.
+ */
+ private static class TimeMilliLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.IntParser {
+ public TimeMilliLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.TIME_MILLIS);
}
- };
- }
- private static JsonToAvroFieldProcessor generateBytesTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- // Should return ByteBuffer (see GenericData.isBytes())
- return Pair.of(true, ByteBuffer.wrap(getUTF8Bytes(value.toString())));
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, LocalTime> result = convertToLocalTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
+ }
+ LocalTime time = result.getRight();
+ Integer millisOfDay = time.toSecondOfDay() * 1000 + time.getNano() / 1000000;
+ return Pair.of(true, millisOfDay);
+ }
+ }
+
+ /**
+ * Processor for TimeMicro logical type.
+ */
+ private static class TimeMicroLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.LongParser {
+ public TimeMicroLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.TIME_MICROS);
}
- };
- }
- private static JsonToAvroFieldProcessor generateFixedTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- // The ObjectMapper use List to represent FixedType
- // eg: "decimal_val": [0, 0, 14, -63, -52] will convert to ArrayList<Integer>
- List<Integer> converval = (List<Integer>) value;
- byte[] src = new byte[converval.size()];
- for (int i = 0; i < converval.size(); i++) {
- src[i] = converval.get(i).byteValue();
- }
- byte[] dst = new byte[schema.getFixedSize()];
- System.arraycopy(src, 0, dst, 0, Math.min(schema.getFixedSize(), src.length));
- return Pair.of(true, new GenericData.Fixed(schema, dst));
- }
- };
- }
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, LocalTime> result = convertToLocalTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
+ }
+ LocalTime time = result.getRight();
+ Long microsOfDay = (long) time.toSecondOfDay() * 1000000 + time.getNano() / 1000;
+ return Pair.of(true, microsOfDay);
+ }
+ }
+
+ /**
+ * Processor for TimeMicro logical type.
+ */
+ private static class LocalTimestampMicroLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.LongParser {
+ public LocalTimestampMicroLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MICROS);
+ }
- private static JsonToAvroFieldProcessor generateEnumTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (schema.getEnumSymbols().contains(value.toString())) {
- return Pair.of(true, new GenericData.EnumSymbol(schema, value.toString()));
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, LocalDateTime> result = convertToLocalDateTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
}
- throw new HoodieJsonToAvroConversionException(String.format("Symbol %s not in enum", value.toString()),
- schema.getFullName(), schema, shouldSanitize, invalidCharMask);
+ LocalDateTime time = result.getRight();
+
+ // Calculate the difference in milliseconds
+ long diffInMicros = LOCAL_UNIX_EPOCH.until(time, ChronoField.MICRO_OF_SECOND.getBaseUnit());
+ return Pair.of(true, diffInMicros);
+ }
+ }
+
+ /**
+ * Processor for TimeMicro logical type.
+ */
+ private static class TimestampMicroLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.LongParser {
+ public TimestampMicroLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.TIMESTAMP_MICROS);
}
- };
- }
- private static JsonToAvroFieldProcessor generateRecordTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- GenericRecord result = new GenericData.Record(schema);
- return Pair.of(true, convertJsonToAvro((Map<String, Object>) value, schema, shouldSanitize, invalidCharMask));
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, Instant> result = convertToInstantTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
+ }
+ Instant time = result.getRight();
+
+ // Calculate the difference in milliseconds
+ long diffInMicro = Instant.EPOCH.until(time, ChronoField.MICRO_OF_SECOND.getBaseUnit());
+ return Pair.of(true, diffInMicro);
+ }
+ }
+
+ /**
+ * Processor for TimeMicro logical type.
+ */
+ private static class LocalTimestampMilliLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.LongParser {
+ public LocalTimestampMilliLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MILLIS);
}
- };
- }
- private static JsonToAvroFieldProcessor generateArrayTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Schema elementSchema = schema.getElementType();
- List<Object> listRes = new ArrayList<>();
- for (Object v : (List) value) {
- listRes.add(convertJsonToAvroField(v, name, elementSchema, shouldSanitize, invalidCharMask));
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, LocalDateTime> result = convertToLocalDateTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
}
- return Pair.of(true, new GenericData.Array<>(schema, listRes));
+ LocalDateTime time = result.getRight();
+
+ // Calculate the difference in milliseconds
+ long diffInMillis = LOCAL_UNIX_EPOCH.until(time, ChronoField.MILLI_OF_SECOND.getBaseUnit());
+ return Pair.of(true, diffInMillis);
+ }
+ }
+
+ /**
+ * Processor for TimeMicro logical type.
+ */
+ private static class TimestampMilliLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.LongParser {
+ public TimestampMilliLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.TIMESTAMP_MILLIS);
}
- };
- }
- private static JsonToAvroFieldProcessor generateMapTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Schema valueSchema = schema.getValueType();
- Map<String, Object> mapRes = new HashMap<>();
- for (Map.Entry<String, Object> v : ((Map<String, Object>) value).entrySet()) {
- mapRes.put(v.getKey(), convertJsonToAvroField(v.getValue(), name, valueSchema, shouldSanitize, invalidCharMask));
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, Instant> result = convertToInstantTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
}
- return Pair.of(true, mapRes);
+ Instant time = result.getRight();
+
+ // Calculate the difference in milliseconds
+ long diffInMillis = Instant.EPOCH.until(time, ChronoField.MILLI_OF_SECOND.getBaseUnit());
+ return Pair.of(true, diffInMillis);
}
- };
+ }
+
+ private static JsonToAvroFieldProcessor generateBooleanTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ if (value instanceof Boolean) {
+ return Pair.of(true, value);
+ }
+ return Pair.of(false, null);
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateIntTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ if (value instanceof Number) {
+ return Pair.of(true, ((Number) value).intValue());
+ } else if (value instanceof String) {
+ return Pair.of(true, Integer.valueOf((String) value));
+ }
+ return Pair.of(false, null);
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateDoubleTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ if (value instanceof Number) {
+ return Pair.of(true, ((Number) value).doubleValue());
+ } else if (value instanceof String) {
+ return Pair.of(true, Double.valueOf((String) value));
+ }
+ return Pair.of(false, null);
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateFloatTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ if (value instanceof Number) {
+ return Pair.of(true, ((Number) value).floatValue());
+ } else if (value instanceof String) {
+ return Pair.of(true, Float.valueOf((String) value));
+ }
+ return Pair.of(false, null);
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateLongTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ if (value instanceof Number) {
+ return Pair.of(true, ((Number) value).longValue());
+ } else if (value instanceof String) {
+ return Pair.of(true, Long.valueOf((String) value));
+ }
+ return Pair.of(false, null);
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateStringTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ return Pair.of(true, value.toString());
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateBytesTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ // Should return ByteBuffer (see GenericData.isBytes())
+ return Pair.of(true, ByteBuffer.wrap(value.toString().getBytes()));
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateFixedTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ // The ObjectMapper use List to represent FixedType
+ // eg: "decimal_val": [0, 0, 14, -63, -52] will convert to ArrayList<Integer>
Review Comment:
this is day 1 implementation, it is marked as new code change due to indentation is changed. We should check with the original owner of the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "Davis-Zhang-Onehouse (via GitHub)" <gi...@apache.org>.
Davis-Zhang-Onehouse commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1668961091
##########
hudi-common/src/test/resources/date-type-invalid.avsc:
##########
Review Comment:
Thanks Sagar for the comments, Given the code is in a ready to ship status, can I create a follow up task for such refactor? Mostly for prioritization purpose.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11265:
URL: https://github.com/apache/hudi/pull/11265#issuecomment-2214921106
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24007",
"triggerID" : "c284b2fce8c40e3c5b7448a6cf2f47f3b5ae50c4",
"triggerType" : "PUSH"
}, {
"hash" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24017",
"triggerID" : "134dda251bbfac3cc092ec3f64b3cf3c8b17e2e1",
"triggerType" : "PUSH"
}, {
"hash" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24101",
"triggerID" : "5699a637e30e3fafd056ab25d668640c5c8700c0",
"triggerType" : "PUSH"
}, {
"hash" : "5cdb05c7d909edebbc74ccf75e26c07820272090",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24774",
"triggerID" : "5cdb05c7d909edebbc74ccf75e26c07820272090",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5cdb05c7d909edebbc74ccf75e26c07820272090 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24774)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-7774] Add Avro Logical type support for Merciful Java convertor [hudi]
Posted by "codope (via GitHub)" <gi...@apache.org>.
codope commented on code in PR #11265:
URL: https://github.com/apache/hudi/pull/11265#discussion_r1673285947
##########
hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java:
##########
@@ -187,196 +178,774 @@ private static Object convertJsonToAvroField(Object value, String name, Schema s
throw new HoodieJsonToAvroConversionException(null, name, schema, shouldSanitize, invalidCharMask);
}
- JsonToAvroFieldProcessor processor = FIELD_TYPE_PROCESSORS.get(schema.getType());
- if (null != processor) {
+ return JsonToAvroFieldProcessorUtil.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ private static class JsonToAvroFieldProcessorUtil {
+ /**
+ * Base Class for converting json to avro fields.
+ */
+ private abstract static class JsonToAvroFieldProcessor implements Serializable {
+
+ public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
+ if (!res.getLeft()) {
+ throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+ return res.getRight();
+ }
+
+ protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
+ }
+
+ public static Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ JsonToAvroFieldProcessor processor = getProcessorForSchema(schema);
return processor.convertToAvro(value, name, schema, shouldSanitize, invalidCharMask);
}
- throw new IllegalArgumentException("JsonConverter cannot handle type: " + schema.getType());
- }
- /**
- * Base Class for converting json to avro fields.
- */
- private abstract static class JsonToAvroFieldProcessor implements Serializable {
+ private static JsonToAvroFieldProcessor getProcessorForSchema(Schema schema) {
+ JsonToAvroFieldProcessor processor = null;
+
+ // 3 cases to consider: customized logicalType, logicalType, and type.
+ String customizedLogicalType = schema.getProp("logicalType");
+ LogicalType logicalType = schema.getLogicalType();
+ Type type = schema.getType();
+ if (customizedLogicalType != null && !customizedLogicalType.isEmpty()) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(customizedLogicalType);
+ } else if (logicalType != null) {
+ processor = AVRO_LOGICAL_TYPE_FIELD_PROCESSORS.get(logicalType.getName());
+ } else {
+ processor = AVRO_TYPE_FIELD_TYPE_PROCESSORS.get(type);
+ }
- public Object convertToAvro(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Pair<Boolean, Object> res = convert(value, name, schema, shouldSanitize, invalidCharMask);
- if (!res.getLeft()) {
- throw new HoodieJsonToAvroConversionException(value, name, schema, shouldSanitize, invalidCharMask);
+ if (processor == null) {
+ throw new IllegalArgumentException(String.format("JsonConverter cannot handle type: %s", type));
}
- return res.getRight();
+ return processor;
}
- protected abstract Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask);
- }
+ // Avro primitive and complex type processors.
+ private static final Map<Schema.Type, JsonToAvroFieldProcessor> AVRO_TYPE_FIELD_TYPE_PROCESSORS = getFieldTypeProcessors();
+ // Avro logical type processors.
+ private static final Map<String, JsonToAvroFieldProcessor> AVRO_LOGICAL_TYPE_FIELD_PROCESSORS = getLogicalFieldTypeProcessors();
+
+ /**
+ * Build type processor map for each avro type.
+ */
+ private static Map<Schema.Type, JsonToAvroFieldProcessor> getFieldTypeProcessors() {
+ Map<Schema.Type, JsonToAvroFieldProcessor> fieldTypeProcessors = new EnumMap<>(Schema.Type.class);
+ fieldTypeProcessors.put(Type.STRING, generateStringTypeHandler());
+ fieldTypeProcessors.put(Type.BOOLEAN, generateBooleanTypeHandler());
+ fieldTypeProcessors.put(Type.DOUBLE, generateDoubleTypeHandler());
+ fieldTypeProcessors.put(Type.FLOAT, generateFloatTypeHandler());
+ fieldTypeProcessors.put(Type.INT, generateIntTypeHandler());
+ fieldTypeProcessors.put(Type.LONG, generateLongTypeHandler());
+ fieldTypeProcessors.put(Type.ARRAY, generateArrayTypeHandler());
+ fieldTypeProcessors.put(Type.RECORD, generateRecordTypeHandler());
+ fieldTypeProcessors.put(Type.ENUM, generateEnumTypeHandler());
+ fieldTypeProcessors.put(Type.MAP, generateMapTypeHandler());
+ fieldTypeProcessors.put(Type.BYTES, generateBytesTypeHandler());
+ fieldTypeProcessors.put(Type.FIXED, generateFixedTypeHandler());
+ return Collections.unmodifiableMap(fieldTypeProcessors);
+ }
- private static JsonToAvroFieldProcessor generateBooleanTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ private static Map<String, JsonToAvroFieldProcessor> getLogicalFieldTypeProcessors() {
+ Map<String, JsonToAvroFieldProcessor> logicalFieldTypeProcessors = new HashMap<>();
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.DECIMAL.getValue(), new DecimalLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.TIME_MICROS.getValue(), new TimeMicroLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.TIME_MILLIS.getValue(), new TimeMilliLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.DATE.getValue(), new DateLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MICROS.getValue(), new LocalTimestampMicroLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MILLIS.getValue(), new LocalTimestampMilliLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.TIMESTAMP_MICROS.getValue(), new TimestampMicroLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.TIMESTAMP_MILLIS.getValue(), new TimestampMilliLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.DURATION.getValue(), new DurationLogicalTypeProcessor());
+ logicalFieldTypeProcessors.put(AvroLogicalTypeEnum.UUID.getValue(), generateStringTypeHandler());
+ return Collections.unmodifiableMap(logicalFieldTypeProcessors);
+ }
+
+ private static class DecimalLogicalTypeProcessor extends JsonToAvroFieldProcessor {
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Boolean) {
- return Pair.of(true, value);
+
+ if (!isValidDecimalTypeConfig(schema)) {
+ return Pair.of(false, null);
+ }
+
+ // Case 1: Input is a list. It is expected to be raw Fixed byte array input, and we only support
+ // parsing it to Fixed avro type.
+ if (value instanceof List<?> && schema.getType() == Type.FIXED) {
+ JsonToAvroFieldProcessor processor = generateFixedTypeHandler();
+ return processor.convert(value, name, schema, shouldSanitize, invalidCharMask);
+ }
+
+ // Case 2: Input is a number or String number.
+ LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+ Pair<Boolean, BigDecimal> parseResult = parseObjectToBigDecimal(value);
+ if (Boolean.FALSE.equals(parseResult.getLeft())) {
+ return Pair.of(false, null);
+ }
+ BigDecimal bigDecimal = parseResult.getRight();
+
+ // As we don't do rounding, the validation will enforce the scale part and the integer part are all within the
+ // limit. As a result, if scale is 2 precision is 5, we only allow 3 digits for the integer.
+ // Allowed: 123.45, 123, 0.12
+ // Disallowed: 1234 (4 digit integer while the scale has already reserved 2 digit out of the 5 digit precision)
+ // 123456, 0.12345
+ if (bigDecimal.scale() > decimalType.getScale()
+ || (bigDecimal.precision() - bigDecimal.scale()) > (decimalType.getPrecision() - decimalType.getScale())) {
+ // Correspond to case
+ // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 5 as scale 2 without rounding.
+ // org.apache.avro.AvroTypeException: Cannot encode decimal with scale 3 as scale 2 without rounding
+ return Pair.of(false, null);
+ }
+
+ switch (schema.getType()) {
+ case BYTES:
+ // Convert to primitive Arvo type that logical type Decimal uses.
+ ByteBuffer byteBuffer = new Conversions.DecimalConversion().toBytes(bigDecimal, schema, decimalType);
+ return Pair.of(true, byteBuffer);
+ case FIXED:
+ GenericFixed fixedValue = new Conversions.DecimalConversion().toFixed(bigDecimal, schema, decimalType);
+ return Pair.of(true, fixedValue);
+ default: {
+ return Pair.of(false, null);
+ }
}
- return Pair.of(false, null);
}
- };
- }
- private static JsonToAvroFieldProcessor generateIntTypeHandler() {
- return new JsonToAvroFieldProcessor() {
- @Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).intValue());
- } else if (value instanceof String) {
- return Pair.of(true, Integer.valueOf((String) value));
+ /**
+ * Check if the given schema is a valid decimal type configuration.
+ */
+ private static boolean isValidDecimalTypeConfig(Schema schema) {
+ LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) schema.getLogicalType();
+ // At the time when the schema is found not valid when it is parsed, the Avro Schema.parse will just silently
+ // set the schema to be null instead of throwing exceptions. Correspondingly, we just check if it is null here.
+ if (decimalType == null) {
+ return false;
}
- return Pair.of(false, null);
+ // Even though schema is validated at schema parsing phase, still validate here to be defensive.
+ decimalType.validate(schema);
+ return true;
}
- };
- }
- private static JsonToAvroFieldProcessor generateDoubleTypeHandler() {
- return new JsonToAvroFieldProcessor() {
- @Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).doubleValue());
- } else if (value instanceof String) {
- return Pair.of(true, Double.valueOf((String) value));
+ /**
+ * Parse the object to BigDecimal.
+ *
+ * @param obj Object to be parsed
+ * @return Pair object, with left as boolean indicating if the parsing was successful and right as the
+ * BigDecimal value.
+ */
+ private static Pair<Boolean, BigDecimal> parseObjectToBigDecimal(Object obj) {
+ // Case 1: Object is a number.
+ if (obj instanceof Number) {
+ return Pair.of(true, BigDecimal.valueOf(((Number) obj).doubleValue()));
+ }
+
+ // Case 2: Object is a number in String format.
+ if (obj instanceof String) {
+ BigDecimal bigDecimal = null;
+ try {
+ bigDecimal = new BigDecimal(((String) obj));
+ } catch (java.lang.NumberFormatException ignored) {
+ /* ignore */
+ }
+ return Pair.of(bigDecimal != null, bigDecimal);
}
return Pair.of(false, null);
}
- };
- }
+ }
+
+ private static class DurationLogicalTypeProcessor extends JsonToAvroFieldProcessor {
+ private static final int NUM_ELEMENTS_FOR_DURATION_TYPE = 3;
+
+ /**
+ * We expect the input to be a list of 3 integers representing months, days and milliseconds.
+ */
+ private boolean isValidDurationInput(Object value) {
+ if (!(value instanceof List<?>)) {
+ return false;
+ }
+ List<?> list = (List<?>) value;
+ if (list.size() != NUM_ELEMENTS_FOR_DURATION_TYPE) {
+ return false;
+ }
+ for (Object element : list) {
+ if (!(element instanceof Integer)) {
+ return false;
+ }
+ }
+ return true;
+ }
- private static JsonToAvroFieldProcessor generateFloatTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ /**
+ * Convert the given object to Avro object with schema whose logical type is duration.
+ */
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (value instanceof Number) {
- return Pair.of(true, ((Number) value).floatValue());
- } else if (value instanceof String) {
- return Pair.of(true, Float.valueOf((String) value));
+
+ if (!isValidDurationTypeConfig(schema)) {
+ return Pair.of(false, null);
}
- return Pair.of(false, null);
+ if (!isValidDurationInput(value)) {
+ return Pair.of(false, null);
+ }
+ // After the validation the input can be safely cast to List<Integer> with 3 elements.
+ List<?> list = (List<?>) value;
+ List<Integer> converval = list.stream()
+ .filter(Integer.class::isInstance)
+ .map(Integer.class::cast)
+ .collect(Collectors.toList());
+
+ ByteBuffer buffer = ByteBuffer.allocate(schema.getFixedSize()).order(ByteOrder.LITTLE_ENDIAN);
+ for (Integer element : converval) {
+ buffer.putInt(element); // months
+ }
+ return Pair.of(true, new GenericData.Fixed(schema, buffer.array()));
+ }
+
+ /**
+ * Check if the given schema is a valid decimal type configuration.
+ */
+ private static boolean isValidDurationTypeConfig(Schema schema) {
+ String durationTypeName = AvroLogicalTypeEnum.DURATION.getValue();
+ LogicalType durationType = schema.getLogicalType();
+ String durationTypeProp = schema.getProp("logicalType");
+ // 1. The Avro type should be "Fixed".
+ // 2. Fixed size must be of 12 bytes as it hold 3 integers.
+ // 3. Logical type name should be "duration". The name might be stored in different places based on Avro version
+ // being used here.
+ return schema.getType().equals(Type.FIXED)
+ && schema.getFixedSize() == Integer.BYTES * NUM_ELEMENTS_FOR_DURATION_TYPE
+ && (durationType != null && durationType.getName().equals(durationTypeName)
+ || durationTypeProp != null && durationTypeProp.equals(durationTypeName));
+ }
+ }
+
+ /**
+ * Processor utility handling Number inputs. Consumed by TimeLogicalTypeProcessor.
+ */
+ private interface NumericParser {
+ // Convert the input number to Avro data type according to the class
+ // implementing this interface.
+ Pair<Boolean, Object> handleNumberValue(Number value);
+
+ // Convert the input number to Avro data type according to the class
+ // implementing this interface.
+ // @param value the input number in string format.
+ Pair<Boolean, Object> handleStringNumber(String value);
+
+ interface IntParser extends NumericParser {
+ @Override
+ default Pair<Boolean, Object> handleNumberValue(Number value) {
+ return Pair.of(true, value.intValue());
+ }
+
+ @Override
+ default Pair<Boolean, Object> handleStringNumber(String value) {
+ return Pair.of(true, Integer.parseInt(value));
+ }
+ }
+
+ interface LongParser extends NumericParser {
+ @Override
+ default Pair<Boolean, Object> handleNumberValue(Number value) {
+ return Pair.of(true, value.longValue());
+ }
+
+ @Override
+ default Pair<Boolean, Object> handleStringNumber(String value) {
+ return Pair.of(true, Long.parseLong(value));
+ }
+ }
+ }
+
+ /**
+ * Base Class for converting object to avro logical type TimeMilli/TimeMicro.
+ */
+ private abstract static class TimeLogicalTypeProcessor extends JsonToAvroFieldProcessor implements NumericParser {
+
+ protected static final LocalDateTime LOCAL_UNIX_EPOCH = LocalDateTime.of(1970, 1, 1, 0, 0, 0, 0);
+
+ // Logical type the processor is handling.
+ private final AvroLogicalTypeEnum logicalTypeEnum;
+
+ public TimeLogicalTypeProcessor(AvroLogicalTypeEnum logicalTypeEnum) {
+ this.logicalTypeEnum = logicalTypeEnum;
}
- };
- }
- private static JsonToAvroFieldProcessor generateLongTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ /**
+ * Main function that convert input to Object with java data type specified by schema
+ */
@Override
public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ LogicalType logicalType = schema.getLogicalType();
+ if (logicalType == null) {
+ return Pair.of(false, null);
+ }
+ logicalType.validate(schema);
if (value instanceof Number) {
- return Pair.of(true, ((Number) value).longValue());
- } else if (value instanceof String) {
- return Pair.of(true, Long.valueOf((String) value));
+ return handleNumberValue((Number) value);
+ }
+ if (value instanceof String) {
+ String valStr = (String) value;
+ if (ALL_DIGITS_WITH_OPTIONAL_SIGN.matcher(valStr).matches()) {
+ return handleStringNumber(valStr);
+ } else if (isWellFormedDateTime(valStr)) {
+ return handleStringValue(valStr);
+ }
}
return Pair.of(false, null);
}
- };
- }
- private static JsonToAvroFieldProcessor generateStringTypeHandler() {
- return new JsonToAvroFieldProcessor() {
+ // Handle the case when the input is a string that may be parsed as a time.
+ protected abstract Pair<Boolean, Object> handleStringValue(String value);
+
+ protected DateTimeFormatter getDateTimeFormatter() {
+ DateTimeParseContext ctx = DATE_TIME_PARSE_CONTEXT_MAP.get(logicalTypeEnum);
+ assert ctx != null : String.format("%s should have configured date time context.", logicalTypeEnum.getValue());
+ return ctx.dateTimeFormatter;
+ }
+
+ protected Pattern getDateTimePattern() {
+ DateTimeParseContext ctx = DATE_TIME_PARSE_CONTEXT_MAP.get(logicalTypeEnum);
+ assert ctx != null : String.format("%s should have configured date time context.", logicalTypeEnum.getValue());
+ return ctx.dateTimePattern;
+ }
+
+ // Depending on the logical type the processor handles, they use different parsing context
+ // when they need to parse a timestamp string in handleStringValue.
+ private static class DateTimeParseContext {
+ public DateTimeParseContext(DateTimeFormatter dateTimeFormatter, Pattern dateTimePattern) {
+ this.dateTimeFormatter = dateTimeFormatter;
+ this.dateTimePattern = dateTimePattern;
+ }
+
+ public final Pattern dateTimePattern;
+
+ public final DateTimeFormatter dateTimeFormatter;
+ }
+
+ private static final Map<AvroLogicalTypeEnum, DateTimeParseContext> DATE_TIME_PARSE_CONTEXT_MAP = getParseContext();
+
+ private static Map<AvroLogicalTypeEnum, DateTimeParseContext> getParseContext() {
+ // Formatter for parsing local timestamp. It ignores the time zone info of the timestamp.
+ // No pattern is defined as ISO_INSTANT format internal is not clear.
+ DateTimeParseContext dateTimestampParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_INSTANT,
+ null /* match everything*/);
+ // Formatter for parsing timestamp with time zone. The pattern is derived from ISO_LOCAL_TIME definition.
+ // Pattern asserts the string is
+ // <optional sign><Hour>:<Minute> + optional <second> + optional <fractional second>
+ DateTimeParseContext dateTimeParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_LOCAL_TIME,
+ Pattern.compile("^[+-]?\\d{2}:\\d{2}(?::\\d{2}(?:\\.\\d{1,9})?)?"));
+ // Formatter for parsing local time. The pattern is derived from ISO_LOCAL_TIME definition.
+ // Pattern asserts the string is
+ // <optional sign><Year>-<Month>-<Day>T<Hour>:<Minute> + optional <second> + optional <fractional second>
+ DateTimeParseContext localTimestampParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_LOCAL_DATE_TIME,
+ Pattern.compile("^[+-]?\\d{4,10}-\\d{2}-\\d{2}T\\d{2}:\\d{2}(?::\\d{2}(?:\\.\\d{1,9})?)?")
+ );
+ // Formatter for parsing local date. The pattern is derived from ISO_LOCAL_DATE definition.
+ // Pattern asserts the string is
+ // <optional sign><Year>-<Month>-<Day>
+ DateTimeParseContext localDateParseContext = new DateTimeParseContext(
+ DateTimeFormatter.ISO_LOCAL_DATE,
+ Pattern.compile("^[+-]?\\d{4,10}-\\d{2}-\\d{2}?")
+ );
+
+ EnumMap<AvroLogicalTypeEnum, DateTimeParseContext> ctx = new EnumMap<>(AvroLogicalTypeEnum.class);
+ ctx.put(AvroLogicalTypeEnum.TIME_MICROS, dateTimeParseContext);
+ ctx.put(AvroLogicalTypeEnum.TIME_MILLIS, dateTimeParseContext);
+ ctx.put(AvroLogicalTypeEnum.DATE, localDateParseContext);
+ ctx.put(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MICROS, localTimestampParseContext);
+ ctx.put(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MILLIS, localTimestampParseContext);
+ ctx.put(AvroLogicalTypeEnum.TIMESTAMP_MICROS, dateTimestampParseContext);
+ ctx.put(AvroLogicalTypeEnum.TIMESTAMP_MILLIS, dateTimestampParseContext);
+ return Collections.unmodifiableMap(ctx);
+ }
+
+ // Pattern validating if it is an number in string form.
+ // Only check at most 19 digits as this is the max num of digits for LONG.MAX_VALUE to contain the cost of regex matching.
+ protected static final Pattern ALL_DIGITS_WITH_OPTIONAL_SIGN = Pattern.compile("^[-+]?\\d{1,19}$");
+
+ /**
+ * Check if the given string is a well-formed date time string.
+ * If no pattern is defined, it will always return true.
+ */
+ private boolean isWellFormedDateTime(String value) {
+ Pattern pattern = getDateTimePattern();
+ return pattern == null || pattern.matcher(value).matches();
+ }
+
+ protected Pair<Boolean, Instant> convertToInstantTime(String input) {
+ // Parse the input timestamp, DateTimeFormatter.ISO_INSTANT is implied here
+ Instant time = null;
+ try {
+ time = Instant.parse(input);
+ } catch (DateTimeParseException ignore) {
+ /* ignore */
+ }
+ return Pair.of(time != null, time);
+ }
+
+ protected Pair<Boolean, LocalTime> convertToLocalTime(String input) {
+ // Parse the input timestamp, DateTimeFormatter.ISO_LOCAL_TIME is implied here
+ LocalTime time = null;
+ try {
+ // Try parsing as an ISO date
+ time = LocalTime.parse(input);
+ } catch (DateTimeParseException ignore) {
+ /* ignore */
+ }
+ return Pair.of(time != null, time);
+ }
+
+ protected Pair<Boolean, LocalDateTime> convertToLocalDateTime(String input) {
+ // Parse the input timestamp, DateTimeFormatter.ISO_LOCAL_DATE_TIME is implied here
+ LocalDateTime time = null;
+ try {
+ // Try parsing as an ISO date
+ time = LocalDateTime.parse(input, getDateTimeFormatter());
+ } catch (DateTimeParseException ignore) {
+ /* ignore */
+ }
+ return Pair.of(time != null, time);
+ }
+ }
+
+ private static class DateLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.IntParser {
+ public DateLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.DATE);
+ }
+
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- return Pair.of(true, value.toString());
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, LocalDate> result = convertToLocalDate(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
+ }
+ LocalDate date = result.getRight();
+ int daysSinceEpoch = (int) ChronoUnit.DAYS.between(LocalDate.ofEpochDay(0), date);
+ return Pair.of(true, daysSinceEpoch);
+ }
+
+ private Pair<Boolean, LocalDate> convertToLocalDate(String input) {
+ // Parse the input timestamp, DateTimeFormatter.ISO_LOCAL_TIME is implied here
+ LocalDate date = null;
+ try {
+ // Try parsing as an ISO date
+ date = LocalDate.parse(input);
+ } catch (DateTimeParseException ignore) {
+ /* ignore */
+ }
+ return Pair.of(date != null, date);
+ }
+ }
+
+ /**
+ * Processor for TimeMilli logical type.
+ */
+ private static class TimeMilliLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.IntParser {
+ public TimeMilliLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.TIME_MILLIS);
}
- };
- }
- private static JsonToAvroFieldProcessor generateBytesTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- // Should return ByteBuffer (see GenericData.isBytes())
- return Pair.of(true, ByteBuffer.wrap(getUTF8Bytes(value.toString())));
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, LocalTime> result = convertToLocalTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
+ }
+ LocalTime time = result.getRight();
+ Integer millisOfDay = time.toSecondOfDay() * 1000 + time.getNano() / 1000000;
+ return Pair.of(true, millisOfDay);
+ }
+ }
+
+ /**
+ * Processor for TimeMicro logical type.
+ */
+ private static class TimeMicroLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.LongParser {
+ public TimeMicroLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.TIME_MICROS);
}
- };
- }
- private static JsonToAvroFieldProcessor generateFixedTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- // The ObjectMapper use List to represent FixedType
- // eg: "decimal_val": [0, 0, 14, -63, -52] will convert to ArrayList<Integer>
- List<Integer> converval = (List<Integer>) value;
- byte[] src = new byte[converval.size()];
- for (int i = 0; i < converval.size(); i++) {
- src[i] = converval.get(i).byteValue();
- }
- byte[] dst = new byte[schema.getFixedSize()];
- System.arraycopy(src, 0, dst, 0, Math.min(schema.getFixedSize(), src.length));
- return Pair.of(true, new GenericData.Fixed(schema, dst));
- }
- };
- }
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, LocalTime> result = convertToLocalTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
+ }
+ LocalTime time = result.getRight();
+ Long microsOfDay = (long) time.toSecondOfDay() * 1000000 + time.getNano() / 1000;
+ return Pair.of(true, microsOfDay);
+ }
+ }
+
+ /**
+ * Processor for TimeMicro logical type.
+ */
+ private static class LocalTimestampMicroLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.LongParser {
+ public LocalTimestampMicroLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MICROS);
+ }
- private static JsonToAvroFieldProcessor generateEnumTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- if (schema.getEnumSymbols().contains(value.toString())) {
- return Pair.of(true, new GenericData.EnumSymbol(schema, value.toString()));
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, LocalDateTime> result = convertToLocalDateTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
}
- throw new HoodieJsonToAvroConversionException(String.format("Symbol %s not in enum", value.toString()),
- schema.getFullName(), schema, shouldSanitize, invalidCharMask);
+ LocalDateTime time = result.getRight();
+
+ // Calculate the difference in milliseconds
+ long diffInMicros = LOCAL_UNIX_EPOCH.until(time, ChronoField.MICRO_OF_SECOND.getBaseUnit());
+ return Pair.of(true, diffInMicros);
+ }
+ }
+
+ /**
+ * Processor for TimeMicro logical type.
+ */
+ private static class TimestampMicroLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.LongParser {
+ public TimestampMicroLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.TIMESTAMP_MICROS);
}
- };
- }
- private static JsonToAvroFieldProcessor generateRecordTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- GenericRecord result = new GenericData.Record(schema);
- return Pair.of(true, convertJsonToAvro((Map<String, Object>) value, schema, shouldSanitize, invalidCharMask));
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, Instant> result = convertToInstantTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
+ }
+ Instant time = result.getRight();
+
+ // Calculate the difference in milliseconds
+ long diffInMicro = Instant.EPOCH.until(time, ChronoField.MICRO_OF_SECOND.getBaseUnit());
+ return Pair.of(true, diffInMicro);
+ }
+ }
+
+ /**
+ * Processor for TimeMicro logical type.
+ */
+ private static class LocalTimestampMilliLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.LongParser {
+ public LocalTimestampMilliLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.LOCAL_TIMESTAMP_MILLIS);
}
- };
- }
- private static JsonToAvroFieldProcessor generateArrayTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Schema elementSchema = schema.getElementType();
- List<Object> listRes = new ArrayList<>();
- for (Object v : (List) value) {
- listRes.add(convertJsonToAvroField(v, name, elementSchema, shouldSanitize, invalidCharMask));
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, LocalDateTime> result = convertToLocalDateTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
}
- return Pair.of(true, new GenericData.Array<>(schema, listRes));
+ LocalDateTime time = result.getRight();
+
+ // Calculate the difference in milliseconds
+ long diffInMillis = LOCAL_UNIX_EPOCH.until(time, ChronoField.MILLI_OF_SECOND.getBaseUnit());
+ return Pair.of(true, diffInMillis);
+ }
+ }
+
+ /**
+ * Processor for TimeMicro logical type.
+ */
+ private static class TimestampMilliLogicalTypeProcessor extends TimeLogicalTypeProcessor
+ implements NumericParser.LongParser {
+ public TimestampMilliLogicalTypeProcessor() {
+ super(AvroLogicalTypeEnum.TIMESTAMP_MILLIS);
}
- };
- }
- private static JsonToAvroFieldProcessor generateMapTypeHandler() {
- return new JsonToAvroFieldProcessor() {
@Override
- public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
- Schema valueSchema = schema.getValueType();
- Map<String, Object> mapRes = new HashMap<>();
- for (Map.Entry<String, Object> v : ((Map<String, Object>) value).entrySet()) {
- mapRes.put(v.getKey(), convertJsonToAvroField(v.getValue(), name, valueSchema, shouldSanitize, invalidCharMask));
+ public Pair<Boolean, Object> handleStringValue(String value) {
+ Pair<Boolean, Instant> result = convertToInstantTime(value);
+ if (!result.getLeft()) {
+ return Pair.of(false, null);
}
- return Pair.of(true, mapRes);
+ Instant time = result.getRight();
+
+ // Calculate the difference in milliseconds
+ long diffInMillis = Instant.EPOCH.until(time, ChronoField.MILLI_OF_SECOND.getBaseUnit());
+ return Pair.of(true, diffInMillis);
}
- };
+ }
+
+ private static JsonToAvroFieldProcessor generateBooleanTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ if (value instanceof Boolean) {
+ return Pair.of(true, value);
+ }
+ return Pair.of(false, null);
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateIntTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ if (value instanceof Number) {
+ return Pair.of(true, ((Number) value).intValue());
+ } else if (value instanceof String) {
+ return Pair.of(true, Integer.valueOf((String) value));
+ }
+ return Pair.of(false, null);
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateDoubleTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ if (value instanceof Number) {
+ return Pair.of(true, ((Number) value).doubleValue());
+ } else if (value instanceof String) {
+ return Pair.of(true, Double.valueOf((String) value));
+ }
+ return Pair.of(false, null);
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateFloatTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ if (value instanceof Number) {
+ return Pair.of(true, ((Number) value).floatValue());
+ } else if (value instanceof String) {
+ return Pair.of(true, Float.valueOf((String) value));
+ }
+ return Pair.of(false, null);
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateLongTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ if (value instanceof Number) {
+ return Pair.of(true, ((Number) value).longValue());
+ } else if (value instanceof String) {
+ return Pair.of(true, Long.valueOf((String) value));
+ }
+ return Pair.of(false, null);
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateStringTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ return Pair.of(true, value.toString());
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateBytesTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ // Should return ByteBuffer (see GenericData.isBytes())
+ return Pair.of(true, ByteBuffer.wrap(value.toString().getBytes()));
+ }
+ };
+ }
+
+ private static JsonToAvroFieldProcessor generateFixedTypeHandler() {
+ return new JsonToAvroFieldProcessor() {
+ @Override
+ public Pair<Boolean, Object> convert(Object value, String name, Schema schema, boolean shouldSanitize, String invalidCharMask) {
+ // The ObjectMapper use List to represent FixedType
+ // eg: "decimal_val": [0, 0, 14, -63, -52] will convert to ArrayList<Integer>
Review Comment:
ok if it's not changing behavior
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org