voonhous commented on code in PR #19809:
URL: https://github.com/apache/hudi/pull/19809#discussion_r3916368048
##########
hudi-common/src/main/java/org/apache/hudi/common/avro/HoodieAvroUtils.java:
##########
@@ -883,15 +924,16 @@ private static Object
normalizeAvroLogicalTypeToPrimitive(Object value, Schema s
*
* @param record Hoodie record.
* @param columns Names of the columns to get values.
- * @param schema {@link Schema} instance.
+ * @param schema {@link HoodieSchema} instance.
* @return Column value.
*/
public static Object[] getRecordColumnValues(HoodieRecord record,
String[] columns,
- Schema schema,
+ HoodieSchema schema,
boolean
consistentLogicalTimestampEnabled) {
try {
- GenericRecord genericRecord = (GenericRecord)
(record.toIndexedRecord(HoodieAvroSchemaCache.intern(schema), new
Properties()).get()).getData();
+ // Intern so the identity fast path in BaseAvroPayload#getRecord hits
across callers that parse their own copy of the schema.
+ GenericRecord genericRecord = (GenericRecord)
(record.toIndexedRecord(HoodieSchemaCache.intern(schema), new
Properties()).get()).getData();
Review Comment:
**major:** `HoodieSchemaCache` is value-keyed (`weakValues` + `maximumSize`,
`HoodieSchemaCache.java:36-37`) and `HoodieSchema.equals` only short-circuits
on identity before Avro's deep `RecordSchema.equals`, whereas master's
`HoodieAvroSchemaCache` front is `weakKeys` (identity). Every task whose
deserialized schema copy is not the stored key now pays a deep schema compare
per call: twice per comparison in `RDDBucketIndexPartitioner.java:92,94`, once
per record per column at `HoodieTableMetadataUtil.java:331`. Could this be
`HoodieAvroSchemaCache.intern(schema.toAvroSchema())`, which is byte-for-byte
the master path and lands on the same canonical instance?
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaUtils.java:
##########
@@ -769,47 +788,27 @@ public static String getRecordQualifiedName(String
tableName) {
}
public static boolean hasDecimalField(HoodieSchema schema) {
- return hasDecimalWithCondition(schema, unused -> true);
- }
-
- /**
- * Checks whether the provided schema contains a decimal with a precision
less than or equal to 18,
- * which allows the decimal to be stored as int/long instead of a fixed size
byte array in
- * <a
href="https://github.com/apache/parquet-format/blob/master/LogicalTypes.md";>parquet
logical types</a>
- * @param schema the input schema to search
- * @return true if the schema contains a small precision decimal field and
false otherwise
- */
- public static boolean hasSmallPrecisionDecimalField(HoodieSchema schema) {
- return hasDecimalWithCondition(schema,
HoodieSchemaUtils::isSmallPrecisionDecimalField);
- }
-
- private static boolean hasDecimalWithCondition(HoodieSchema schema,
Function<HoodieSchema.Decimal, Boolean> condition) {
switch (schema.getType()) {
case RECORD:
for (HoodieSchemaField field : schema.getFields()) {
- if (hasDecimalWithCondition(field.schema(), condition)) {
+ if (hasDecimalField(field.schema())) {
return true;
}
}
return false;
case ARRAY:
- return hasDecimalWithCondition(schema.getElementType(), condition);
+ return hasDecimalField(schema.getElementType());
case MAP:
- return hasDecimalWithCondition(schema.getValueType(), condition);
+ return hasDecimalField(schema.getValueType());
case UNION:
- return hasDecimalWithCondition(schema.getNonNullType(), condition);
+ return hasDecimalField(schema.getNonNullType());
Review Comment:
**major:** pre-existing, not introduced by this fold: master has the same
arm (`HoodieSchemaUtils.java:799-800`, since #17600). `getNonNullType()`
returns `this` for a union without a null branch
(`HoodieSchema.java:1397-1399`) and `["null","string","int"]` reduces to
exactly that shape (line 1413), so this recurses to `StackOverflowError`.
`SourceFormatAdapter.java:261` calls it on every JSON source schema;
`HoodieSchemaRepair.hasTimestampMillisField:249` has the identical arm. Out of
scope for a behavior-preserving PR; should the
`getTypes().stream().anyMatch(...)` fix go in a separate PR, with an issue
linked here meanwhile?
##########
hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaUtils.java:
##########
@@ -2198,4 +2083,78 @@ public void testGetNestedFieldComplexNestedMapAndArray()
{
assertEquals("value", result.get().getRight().name());
assertEquals(HoodieSchemaType.LONG,
result.get().getRight().schema().getType());
}
+
+ private static HoodieSchema deleteLogTableSchema() {
+ return HoodieSchema.createRecord(
+ "TestRecord",
+ null,
+ null,
+ Arrays.asList(
+ HoodieSchemaField.of("ts",
HoodieSchema.create(HoodieSchemaType.LONG), "ordering field doc", null),
+ HoodieSchemaField.of("name",
HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaField.of("seq",
HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaField.of("opt_ts", HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.LONG),
HoodieSchema.create(HoodieSchemaType.NULL)))
+ )
+ );
+ }
+
+ @Test
+ public void testCreateDeleteLogSchema() {
+ HoodieSchema deleteLogSchema =
+ HoodieSchemaUtils.createDeleteLogSchema(deleteLogTableSchema(),
Collections.singletonList("ts"));
+
+ assertEquals("hudi_delete_log_record", deleteLogSchema.getName());
+ assertEquals(2, deleteLogSchema.getFields().size());
+
+ // The record key is always present and never nullable.
+ HoodieSchemaField recordKeyField = deleteLogSchema.getFields().get(0);
+ assertEquals(HoodieRecord.RECORD_KEY_METADATA_FIELD,
recordKeyField.name());
+ assertEquals(HoodieSchemaType.STRING, recordKeyField.schema().getType());
+ assertFalse(recordKeyField.isNullable());
+
+ // The ordering field keeps its doc but is made nullable with a null
default, even though
+ // the table schema marks it required.
+ HoodieSchemaField orderingField = deleteLogSchema.getFields().get(1);
+ assertEquals("ts", orderingField.name());
+ assertTrue(orderingField.isNullable());
+ assertEquals(HoodieSchemaType.LONG,
orderingField.getNonNullSchema().getType());
+ assertEquals("ordering field doc", orderingField.doc().get());
+ assertEquals(HoodieSchema.NULL_VALUE, orderingField.defaultVal().get());
+ }
+
+ @Test
+ public void testCreateDeleteLogSchemaOrderingFieldVariants() {
+ HoodieSchema tableSchema = deleteLogTableSchema();
+
+ // Multiple ordering fields keep the caller's order and their own types.
+ HoodieSchema multiOrderingSchema =
HoodieSchemaUtils.createDeleteLogSchema(tableSchema, Arrays.asList("ts",
"seq"));
+ assertEquals(Arrays.asList(HoodieRecord.RECORD_KEY_METADATA_FIELD, "ts",
"seq"),
+
multiOrderingSchema.getFields().stream().map(HoodieSchemaField::name).collect(Collectors.toList()));
+ HoodieSchemaField seqField = multiOrderingSchema.getFields().get(2);
+ assertEquals("seq", seqField.name());
+ assertTrue(seqField.isNullable());
+ assertEquals(HoodieSchemaType.STRING,
seqField.getNonNullSchema().getType());
+ assertEquals(HoodieSchema.NULL_VALUE, seqField.defaultVal().get());
+
+ // No ordering fields leaves the record key alone.
+ assertEquals(1, HoodieSchemaUtils.createDeleteLogSchema(tableSchema,
Collections.emptyList()).getFields().size());
+
+ // An already-nullable ordering field is left as-is rather than
double-wrapped.
Review Comment:
**nit:** feel free to ignore. "left as-is rather than double-wrapped" is
only pinned indirectly: `isNullable()` and `getNonNullSchema()` hold for a
re-wrapped `[null, long]` too, and only the dropped default proves the `[long,
null]` order survived. Could we assert the union shape directly
(`Arrays.asList(LONG, NULL)` against `optTsField.schema().getTypes()` mapped to
types), and drop line 2135, which cannot fail after the field-name list
assertion at 2132?
##########
hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaUtils.java:
##########
@@ -2198,4 +2083,78 @@ public void testGetNestedFieldComplexNestedMapAndArray()
{
assertEquals("value", result.get().getRight().name());
assertEquals(HoodieSchemaType.LONG,
result.get().getRight().schema().getType());
}
+
+ private static HoodieSchema deleteLogTableSchema() {
+ return HoodieSchema.createRecord(
+ "TestRecord",
+ null,
+ null,
+ Arrays.asList(
+ HoodieSchemaField.of("ts",
HoodieSchema.create(HoodieSchemaType.LONG), "ordering field doc", null),
+ HoodieSchemaField.of("name",
HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaField.of("seq",
HoodieSchema.create(HoodieSchemaType.STRING)),
+ HoodieSchemaField.of("opt_ts", HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.LONG),
HoodieSchema.create(HoodieSchemaType.NULL)))
+ )
+ );
+ }
+
+ @Test
+ public void testCreateDeleteLogSchema() {
+ HoodieSchema deleteLogSchema =
+ HoodieSchemaUtils.createDeleteLogSchema(deleteLogTableSchema(),
Collections.singletonList("ts"));
+
+ assertEquals("hudi_delete_log_record", deleteLogSchema.getName());
+ assertEquals(2, deleteLogSchema.getFields().size());
+
+ // The record key is always present and never nullable.
+ HoodieSchemaField recordKeyField = deleteLogSchema.getFields().get(0);
+ assertEquals(HoodieRecord.RECORD_KEY_METADATA_FIELD,
recordKeyField.name());
+ assertEquals(HoodieSchemaType.STRING, recordKeyField.schema().getType());
+ assertFalse(recordKeyField.isNullable());
+
+ // The ordering field keeps its doc but is made nullable with a null
default, even though
+ // the table schema marks it required.
+ HoodieSchemaField orderingField = deleteLogSchema.getFields().get(1);
+ assertEquals("ts", orderingField.name());
+ assertTrue(orderingField.isNullable());
+ assertEquals(HoodieSchemaType.LONG,
orderingField.getNonNullSchema().getType());
+ assertEquals("ordering field doc", orderingField.doc().get());
+ assertEquals(HoodieSchema.NULL_VALUE, orderingField.defaultVal().get());
+ }
+
+ @Test
+ public void testCreateDeleteLogSchemaOrderingFieldVariants() {
Review Comment:
**minor:** not blocking. Order, empty list and a null-last union are covered
now, but no test anywhere builds a delete-log schema from a logical-typed
ordering field, which is where the delete path's history is (#13998
timestamp-millis ordering values, #13163, #12006);
`TestLsmFileIterators.java:163` and `HoodieFileSliceTestUtils.java:104` both
use a plain long. The code handles it today. Could `deleteLogTableSchema()`
gain a timestamp-millis and a decimal field, with one assertion each that the
logical type and precision/scale survive?
##########
hudi-common/src/test/java/org/apache/hudi/common/avro/TestHoodieAvroUtils.java:
##########
@@ -955,6 +1085,25 @@ void
testGetSortColumnValuesWithPartitionPathAndRecordKey(boolean suffixRecordKe
}
}
+ @Test
+ void testGetRecordColumnValues() {
Review Comment:
**minor:** not blocking. This test passes with or without the intern at
`HoodieAvroUtils.java:936`: `RewriteAvroPayload.getInsertValue` ignores its
schema argument (`RewriteAvroPayload.java:51-53`), so the
`BaseAvroPayload#getRecord` identity path never runs. Could one case use
`OverwriteWithLatestAvroPayload`, build the record on the interned Avro schema,
and call with a freshly parsed equal `HoodieSchema`? With the intern the string
comes back as `String`; without it the payload round-trips through
`fromAvroBytes` and yields `Utf8`. `student.firstnameNested` is already pinned
on this fixture at lines 476-479, so it could go.
##########
hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaUtils.java:
##########
@@ -1047,6 +1043,11 @@ public void testGenerateProjectionSchema() {
assertTrue(fieldNames1.contains("_row_key"));
assertTrue(fieldNames1.contains("timestamp"));
+ // Field names are matched case-insensitively; HiveHoodieReaderContext
lowercases names before calling this.
+ HoodieSchema schema2 =
HoodieSchemaUtils.generateProjectionSchema(originalSchema,
Arrays.asList("_ROW_KEY"));
Review Comment:
**minor:** not blocking, pre-existing. This pins case-insensitive matching,
but `generateProjectionSchema` lowercases with the default locale
(`HoodieSchemaUtils.java:475,478`) while its only production caller uses
`Locale.ROOT` (`HiveHoodieReaderContext.java:236`); under `tr_TR` an `ID`
column lowercases to dotless-i on one side and `id` on the other, and the
projection throws "Field id not found in log schema". `_ROW_KEY` has no `I`, so
this case cannot catch it. Should the `Locale.ROOT` fix and an `I`-bearing name
(`PII_COL`) go in a separate PR, with an issue linked here?
##########
hudi-common/src/main/java/org/apache/hudi/common/avro/HoodieAvroUtils.java:
##########
@@ -883,15 +924,16 @@ private static Object
normalizeAvroLogicalTypeToPrimitive(Object value, Schema s
*
* @param record Hoodie record.
* @param columns Names of the columns to get values.
- * @param schema {@link Schema} instance.
+ * @param schema {@link HoodieSchema} instance.
* @return Column value.
*/
public static Object[] getRecordColumnValues(HoodieRecord record,
String[] columns,
- Schema schema,
+ HoodieSchema schema,
boolean
consistentLogicalTimestampEnabled) {
try {
- GenericRecord genericRecord = (GenericRecord)
(record.toIndexedRecord(HoodieAvroSchemaCache.intern(schema), new
Properties()).get()).getData();
+ // Intern so the identity fast path in BaseAvroPayload#getRecord hits
across callers that parse their own copy of the schema.
Review Comment:
**minor:** not blocking. This comment describes callers that parse their own
schema copy, but the sibling `getSortColumnValuesWithPartitionPathAndRecordKey`
(line 961) reaches the same sort path through `SortUtils.java:225,259` and
never interned, on master either. Is the asymmetry intended, kept because this
PR is behavior-preserving? If so could the comment say so, and if not should
the sibling intern too?
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaUtils.java:
##########
@@ -731,11 +720,41 @@ public static HoodieSchema
getRecordKeyPartitionPathSchema() {
return HoodieSchema.createRecord("HoodieRecordKey", "", "", false,
toBeAddedFields);
}
+ /**
+ * Schema of a native delete log record: the record key plus the ordering
fields, which are
+ * always nullable (see the comment in the body).
+ */
+ public static HoodieSchema createDeleteLogSchema(HoodieSchema tableSchema,
List<String> orderingFieldNames) {
+ // Native delete logs store only the record key plus optional ordering
values, so ordering fields in
+ // the delete-log schema must always be nullable even when the table
schema marks them required.
+ // A delete record such as HoodieEmptyRecord may carry
OrderingValues.getDefault() as an in-memory
+ // sentinel rather than a real field value. Persist NULL for that missing
value so readers can map it
+ // back to the default ordering without confusing it with a real business
value such as 0.
+ List<HoodieSchemaField> fields = Stream.concat(
+ Stream.of(createNewSchemaField(
+ HoodieRecord.RECORD_KEY_METADATA_FIELD,
HoodieSchema.create(HoodieSchemaType.STRING), null, null)),
+ orderingFieldNames.stream().map(orderingFieldName ->
tableSchema.getField(orderingFieldName)
+ .map(field -> createNewSchemaField(
+ field.name(), HoodieSchema.createNullable(field.schema()),
field.doc().orElse(null), HoodieSchema.NULL_VALUE))
+ .orElseThrow(() ->
+ new IllegalArgumentException("Ordering field " +
orderingFieldName + " not found in table schema"))))
+ .collect(Collectors.toList());
+ return HoodieSchema.createRecord("hudi_delete_log_record", null, null,
fields);
+ }
+
/**
* Fetches projected schema given list of fields to project. The field can
be nested in format `a.b.c` where a is
- * the top level field, b is at second level and so on.
+ * the top level field, b is at second level and so on. Field names are
matched case-sensitively.
* This is equivalent to {@link HoodieAvroUtils#projectSchema(Schema, List)}
but operates on HoodieSchema.
*
+ * <p>The two sibling projection helpers differ:</p>
+ * <ul>
+ * <li>{@link #generateProjectionSchema(HoodieSchema, List)} - top-level
fields only, matched
Review Comment:
**minor:** not blocking. This is the only place that says
`generateProjectionSchema` matches case-insensitively; the method's own javadoc
(lines 461-469) does not, and the reason survives only on the hadoop-mr twin
`HoodieRealtimeRecordReaderUtils.java:126-133` (Hive lowercases column
projections, `37838cea6094`). The lowercase `toMap` also throws
`IllegalStateException: Duplicate key` for two fields differing only in case.
Could the two-line rationale and the collision caveat be ported onto the
method's javadoc?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSource.java:
##########
@@ -296,7 +296,7 @@ private static void verifyDecimalValue(List<GenericRecord>
records, HoodieSchema
double maxVal = Math.pow(10, decSchema.getPrecision() -
decSchema.getScale());
double minVal = maxVal * 0.1;
for (GenericRecord record : records) {
- BigDecimal dec =
org.apache.hudi.common.schema.HoodieSchemaUtils.convertBytesToBigDecimal(((ByteBuffer)
record.get(fieldname)).array(), decSchema);
+ BigDecimal dec =
org.apache.hudi.common.avro.HoodieAvroUtils.convertBytesToBigDecimal(((ByteBuffer)
record.get(fieldname)).array(), decSchema);
Review Comment:
**nit:** feel free to ignore. The fully-qualified name was only needed
because `org.apache.hudi.HoodieSchemaUtils` (line 21) clashed with the old
`common.schema.HoodieSchemaUtils` target; `HoodieAvroUtils` has no clash in
this file. Could we import it and use the short name?
##########
hudi-common/src/main/java/org/apache/hudi/common/avro/HoodieAvroUtils.java:
##########
@@ -883,15 +924,16 @@ private static Object
normalizeAvroLogicalTypeToPrimitive(Object value, Schema s
*
* @param record Hoodie record.
* @param columns Names of the columns to get values.
- * @param schema {@link Schema} instance.
+ * @param schema {@link HoodieSchema} instance.
Review Comment:
**nit:** feel free to ignore. This line moved to `{@link HoodieSchema}`, but
the sibling `getSortColumnValuesWithPartitionPathAndRecordKey` javadoc at line
952 still says `@param schema {@link Schema} instance` for a `HoodieSchema`
parameter (stale on master; `d74b3cd4f3b6` missed it). Could it get the same
edit?
##########
hudi-common/src/test/java/org/apache/hudi/common/avro/TestHoodieAvroUtils.java:
##########
@@ -566,6 +567,135 @@ public void
testConvertValueForAvroLogicalTypesCrossAvroVersion() {
assertEquals(FIXTURE_EPOCH_MICROS,
HoodieAvroUtils.convertValueForAvroLogicalTypes(LOCAL_TS_MICROS_SCHEMA,
FIXTURE_LOCAL_DT_MICROS, false));
}
+ @Test
+ public void testConvertValueForSpecificDataTypes_NullSchema() {
+ // Test with null schema - should return value unchanged
+ String testValue = "test_value";
+ Object result = HoodieAvroUtils.convertValueForSpecificDataTypes(null,
testValue, false);
+ assertEquals(testValue, result);
+ }
+
+ @Test
+ public void testConvertValueForSpecificDataTypes_NullValue_NullableSchema() {
+ // Test with null value and nullable schema - should return null
+ Schema nullableIntSchema =
HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.INT)).toAvroSchema();
+ Object result =
HoodieAvroUtils.convertValueForSpecificDataTypes(nullableIntSchema, null,
false);
+ assertNull(result);
+ }
+
+ @Test
+ public void
testConvertValueForSpecificDataTypes_NullValue_NonNullableSchema() {
+ // Test with null value and non-nullable schema - should throw exception
+ Schema nonNullableSchema = Schema.create(Schema.Type.STRING);
+ assertThrows(IllegalStateException.class, () ->
+ HoodieAvroUtils.convertValueForSpecificDataTypes(nonNullableSchema,
null, false));
+ }
+
+ @Test
+ public void testConvertValueForSpecificDataTypes_DateLogicalType() {
+ // Create date schema
+ Schema dateSchema = HoodieSchema.createDate().toAvroSchema();
Review Comment:
**nit:** feel free to ignore. The moved `convertValueForSpecificDataTypes_*`
tests build the date / timestamp-millis / timestamp-micros schemas by hand
(lines 597, 610, 623, 632) while this class already declares them as
`DATE_SCHEMA`, `TS_MILLIS_SCHEMA` and `TS_MICROS_SCHEMA` (lines 527-529). Could
they reuse the constants now that they live here?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
