This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 9b5742cc961e refactor(schema): dissolve AvroSchemaUtils, dedupe compat
(#19810)
9b5742cc961e is described below
commit 9b5742cc961e8756261e82254cbb0dd5fc0605b4
Author: voonhous <[email protected]>
AuthorDate: Fri Sep 4 19:37:49 2026 +0800
refactor(schema): dissolve AvroSchemaUtils, dedupe compat (#19810)
Part 2 of #16639. Part 1 (#19809, 93f1f711e065) wrote down where
schema helpers belong; this removes the class that has no place in
that rule and the logic the compatibility cluster had implemented
twice. Behavior-preserving, internal API only.
AvroSchemaUtils is gone. Its three remaining uses were all inside
HoodieAvroUtils, so the helpers move there verbatim:
getNonNullTypeFromUnion stays public, isNullable becomes private and
createNewSchemaFromFieldsWithReference package-private.
HoodieSchema.Blob builds its fields with a local nullable() helper and
derives REFERENCE_FIELD_COUNT from a REFERENCE_SCHEMA constant instead
of unwrapping the union around it. The caller-less
HoodieSchemaUtils.createNullableSchema shim goes with it.
HoodieSchemaUtils.asNullable runs the InternalSchema nullability
update on the HoodieSchema directly. The old path was HoodieSchema ->
Avro -> HoodieSchema -> InternalSchema -> HoodieSchema -> Avro ->
HoodieSchema, and the outputs are identical. Two details are kept:
Avro's Schema.isNullable is true for a bare NULL type where
HoodieSchema.isNullable is not, so NULL-typed fields are excluded
explicitly, and the all-nullable case returns the input instance. A
non-RECORD argument is now rejected with IllegalArgumentException
rather than an Avro "Not a record" error.
Compatibility cluster:
- lookupWriterField was implemented twice; the facade keeps its
stricter RECORD precondition and delegates to the checker.
- The checker's LONG/FLOAT/DOUBLE/BYTES/STRING cases collapse into one
HoodieSchemaTypePromotion.canPromote call, against the same table
the projection checker uses.
- TIMESTAMP over LONG and UUID over STRING stay checker-only.
isCompatibleProjectionOf(source, target) tests canPromote(target,
source), so adding them would make a timestamp a compatible
projection of a bare long, and writer-schema deduction would then
keep the table's long as the writer schema and silently drop the
logical type.
- Decimal widening is documented, not unified: the projection checker
requires fixed-size parity, the compatibility checker does not.
- The 3-arg isSchemaCompatible named its parameters reader/writer
while routing them prev/new, so they are renamed, not reordered.
Tests pin what the refactor could have moved: the persisted Blob
schema JSON as a literal, the promotion table from both sides
including the five reverse numeric narrowings and canPromote(LONG,
DATE/TIME), lookupWriterField (direct, alias, ambiguous, absent,
non-record), the argument order of areSchemasCompatible, and the three
pre-existing losses of the asNullable round trip (non-null default,
ENUM, null-last union) with a VECTOR column surviving it.
TestAvroSchemaUtils folds into TestHoodieAvroUtils.
TestHoodieTableSchemaEvolution.testFieldWithAlias gains a type change
so the alias match, not the writerField != null guard, decides its
outcome.
Two javadocs said the surviving toAvroSchema/fromAvroSchema
delegations were being retired under #16639. Where a helper belongs is
that issue; whether its body round-trips is #14263, which allows
conversions only at the memory-to-disk, disk-to-memory and engine
boundaries. They now point at #14263.
Six review items raised on part 1 but not applied before it merged
come along: a bloom-filter fixture spelled out three times in
TestHoodieMetadataPayload, a note on the assertSame there that is
implied by the instance check above it, the missing explanation for why
an equal-but-distinct class schema throws
ArrayIndexOutOfBoundsException, a note on why
collectColumnRangeFieldValueV1's AVRO branch calls the static helper
rather than dispatching (HoodieAvroIndexedRecord reports type AVRO and
overrides getColumnValues), two "@return Column value." on methods
returning an Object[], and javadoc for createSchemaErrorString.
Found on the way and filed rather than fixed here: asNullable on a
table with a BLOB column returns the wrong field, because
InternalSchemaConverter gives the blob's nested reference fields ids
0..3 into a flat id map (#19833).
Closes #16639
---
.../hudi/table/TestHoodieTableSchemaEvolution.java | 11 ++
.../apache/hudi/common/avro/AvroSchemaUtils.java | 146 -----------------
.../apache/hudi/common/avro/HoodieAvroUtils.java | 88 +++++++++--
.../apache/hudi/common/schema/HoodieSchema.java | 28 +++-
.../common/schema/HoodieSchemaCompatibility.java | 48 +++---
.../schema/HoodieSchemaCompatibilityChecker.java | 22 +--
.../common/schema/HoodieSchemaTypePromotion.java | 13 +-
.../hudi/common/schema/HoodieSchemaUtils.java | 91 +++++++----
.../hudi/metadata/HoodieTableMetadataUtil.java | 3 +
.../hudi/common/avro/TestAvroSchemaUtils.java | 70 ---------
.../hudi/common/avro/TestHoodieAvroUtils.java | 74 ++++++++-
.../hudi/common/schema/TestHoodieSchema.java | 21 ++-
.../schema/TestHoodieSchemaCompatibility.java | 153 ++++++++++++++++++
.../schema/TestHoodieSchemaTypePromotion.java | 9 ++
.../hudi/common/schema/TestHoodieSchemaUtils.java | 174 +++++++++++++++++++++
.../avro/TestHoodieAvroWriteSupportShredding.java | 2 +-
.../hudi/metadata/TestHoodieMetadataPayload.java | 23 +--
17 files changed, 658 insertions(+), 318 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/TestHoodieTableSchemaEvolution.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/TestHoodieTableSchemaEvolution.java
index 793a21d741ec..4dd44817b27f 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/TestHoodieTableSchemaEvolution.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/TestHoodieTableSchemaEvolution.java
@@ -189,6 +189,17 @@ public class TestHoodieTableSchemaEvolution {
// Should pass because the field is found via alias and type hasn't changed
assertDoesNotThrow(() ->
HoodieTable.validateSecondaryIndexSchemaEvolution(tableSchema,
writerSchema, indexMetadata));
+
+ // The assertion above passes whether or not the alias resolves: an
unresolved alias returns a null writer
+ // field, which the writerField != null guard then skips. Changing the
type behind the same alias is what
+ // makes the lookup decide the outcome.
+ String retypedWriterSchemaStr = writerSchemaStr.replace(
+ "{\"name\": \"new_name\", \"type\": \"string\"}",
+ "{\"name\": \"new_name\", \"type\": \"int\"}");
+ HoodieSchema retypedWriterSchema =
HoodieSchema.parse(retypedWriterSchemaStr);
+
+ assertThrows(SchemaCompatibilityException.class, () ->
+ HoodieTable.validateSecondaryIndexSchemaEvolution(tableSchema,
retypedWriterSchema, indexMetadata));
}
@Test
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/avro/AvroSchemaUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/avro/AvroSchemaUtils.java
deleted file mode 100644
index 1b12c3840456..000000000000
--- a/hudi-common/src/main/java/org/apache/hudi/common/avro/AvroSchemaUtils.java
+++ /dev/null
@@ -1,146 +0,0 @@
-/*
- * 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.common.avro;
-
-import org.apache.hudi.common.schema.HoodieSchema;
-import org.apache.hudi.common.schema.internal.InternalSchema;
-import org.apache.hudi.common.schema.internal.action.TableChanges;
-import org.apache.hudi.common.schema.internal.utils.SchemaChangeUtils;
-import org.apache.hudi.exception.HoodieAvroSchemaException;
-
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.avro.Schema;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.stream.Collectors;
-
-import static
org.apache.hudi.common.schema.internal.convert.InternalSchemaConverter.convert;
-import static org.apache.hudi.common.util.CollectionUtils.reduce;
-import static org.apache.hudi.common.util.ValidationUtils.checkState;
-
-/**
- * Avro-typed schema helpers, retained only as the delegate target of the call
sites that have not moved to
- * HoodieSchema yet: {@link
org.apache.hudi.common.schema.HoodieSchemaUtils#asNullable(HoodieSchema)} and
- * {@code HoodieSchemaUtils#createNullableSchema}, the field construction
inside {@link HoodieSchema.Blob},
- * and a handful of internal uses in {@link HoodieAvroUtils}.
- *
- * <p>This class is being retired under #16639. Do not add methods here: every
method on this class except
- * {@link #getNonNullTypeFromUnion(Schema)} already has a HoodieSchema twin,
so use
- * {@link org.apache.hudi.common.schema.HoodieSchema} or {@link
org.apache.hudi.common.schema.HoodieSchemaUtils} instead.</p>
- */
-@NoArgsConstructor(access = AccessLevel.PRIVATE)
-@Slf4j
-public class AvroSchemaUtils {
-
- /**
- * Create a new schema but maintain all meta info from the old schema
- *
- * @param schema schema to get the meta info from
- * @param fields list of fields in order that will be in the new schema
- *
- * @return schema with fields from fields, and metadata from schema
- */
- public static Schema createNewSchemaFromFieldsWithReference(Schema schema,
List<Schema.Field> fields) {
- if (schema == null) {
- throw new IllegalArgumentException("Schema must not be null");
- }
- Schema newSchema = Schema.createRecord(schema.getName(), schema.getDoc(),
schema.getNamespace(), schema.isError());
- Map<String, Object> schemaProps = Collections.emptyMap();
- try {
- schemaProps = schema.getObjectProps();
- } catch (Exception e) {
- log.warn("Error while getting object properties from schema: {}",
schema, e);
- }
- for (Map.Entry<String, Object> prop : schemaProps.entrySet()) {
- newSchema.addProp(prop.getKey(), prop.getValue());
- }
- newSchema.setFields(fields);
- return newSchema;
- }
-
- /**
- * Returns true in case provided {@link Schema} is nullable (ie accepting
null values),
- * returns false otherwise
- */
- public static boolean isNullable(Schema schema) {
- if (schema.getType() != Schema.Type.UNION) {
- return false;
- }
-
- List<Schema> innerTypes = schema.getTypes();
- return innerTypes.size() > 1 && innerTypes.stream().anyMatch(it ->
it.getType() == Schema.Type.NULL);
- }
-
- /**
- * Resolves typical Avro's nullable schema definition: {@code
Union(Schema.Type.NULL, <NonNullType>)},
- * decomposing union and returning the target non-null type
- * <p>
- * This is the strict variant: it throws unless the union has exactly one
null branch and one non-null
- * branch. See the union-unwrapping note on {@link HoodieAvroUtils} for the
lenient alternatives.
- * </p>
- */
- public static Schema getNonNullTypeFromUnion(Schema schema) {
- if (schema.getType() != Schema.Type.UNION) {
- return schema;
- }
-
- List<Schema> innerTypes = schema.getTypes();
-
- if (innerTypes.size() != 2) {
- throw new HoodieAvroSchemaException(
- String.format("Unsupported Avro UNION type %s: Only UNION of a null
type and a non-null type is supported", schema));
- }
- Schema firstInnerType = innerTypes.get(0);
- Schema secondInnerType = innerTypes.get(1);
- if ((firstInnerType.getType() != Schema.Type.NULL &&
secondInnerType.getType() != Schema.Type.NULL)
- || (firstInnerType.getType() == Schema.Type.NULL &&
secondInnerType.getType() == Schema.Type.NULL)) {
- throw new HoodieAvroSchemaException(
- String.format("Unsupported Avro UNION type %s: Only UNION of a null
type and a non-null type is supported", schema));
- }
- return firstInnerType.getType() == Schema.Type.NULL ? secondInnerType :
firstInnerType;
- }
-
- public static Schema createNullableSchema(Schema schema) {
- checkState(schema.getType() != Schema.Type.NULL);
- return Schema.createUnion(Schema.create(Schema.Type.NULL), schema);
- }
-
- /**
- * Create a new schema by force changing all the fields as nullable.
- *
- * @param schema original schema
- * @return a new schema with all the fields updated as nullable.
- */
- public static Schema asNullable(Schema schema) {
- List<String> filterCols = schema.getFields().stream()
- .filter(f ->
!f.schema().isNullable()).map(Schema.Field::name).collect(Collectors.toList());
- if (filterCols.isEmpty()) {
- return schema;
- }
- InternalSchema internalSchema =
convert(HoodieSchema.fromAvroSchema(schema));
- TableChanges.ColumnUpdateChange schemaChange =
TableChanges.ColumnUpdateChange.get(internalSchema);
- schemaChange = reduce(filterCols, schemaChange,
- (change, field) -> change.updateColumnNullability(field, true));
- return convert(SchemaChangeUtils.applyTableChanges2Schema(internalSchema,
schemaChange), schema.getFullName()).toAvroSchema();
- }
-}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/avro/HoodieAvroUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/avro/HoodieAvroUtils.java
index 1a83ba67fa63..cc98489cce52 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/avro/HoodieAvroUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/avro/HoodieAvroUtils.java
@@ -145,7 +145,7 @@ import static
org.apache.hudi.common.util.ValidationUtils.checkState;
* <ul>
* <li>{@code unwrapNullable} (this class) - lenient: the first non-null
branch of any union</li>
* <li>{@code getActualSchemaFromUnion} (this class, private) - resolves
complex unions against the datum</li>
- * <li>{@code AvroSchemaUtils#getNonNullTypeFromUnion} - strict: throws
unless the union is exactly one null
+ * <li>{@code getNonNullTypeFromUnion} (this class) - strict: throws unless
the union is exactly one null
* branch and one non-null branch</li>
* <li>{@link HoodieSchema#getNonNullType()} - strips null branches and
never throws</li>
* <li>{@code HoodieSchemaUtils#resolveUnionSchema} - selects a branch by
full name</li>
@@ -387,6 +387,76 @@ public class HoodieAvroUtils {
return new Schema.Field(name, schema, doc,
convertDefaultValueForAvroCompatibility(defaultValue), order);
}
+ /**
+ * Resolves typical Avro's nullable schema definition: {@code
Union(Schema.Type.NULL, <NonNullType>)},
+ * decomposing union and returning the target non-null type
+ * <p>
+ * This is the strict variant: a non-union schema is returned as is, while a
union must have exactly two
+ * branches of which exactly one is {@link Schema.Type#NULL}, otherwise a
{@link HoodieAvroSchemaException}
+ * is thrown. For the lenient alternatives see {@code unwrapNullable} on
this class (first non-null branch
+ * of any union) and {@link HoodieSchema#getNonNullType()} (strips null
branches and never throws).
+ * </p>
+ */
+ public static Schema getNonNullTypeFromUnion(Schema schema) {
+ if (schema.getType() != Schema.Type.UNION) {
+ return schema;
+ }
+
+ List<Schema> innerTypes = schema.getTypes();
+
+ if (innerTypes.size() != 2) {
+ throw new HoodieAvroSchemaException(
+ String.format("Unsupported Avro UNION type %s: Only UNION of a null
type and a non-null type is supported", schema));
+ }
+ Schema firstInnerType = innerTypes.get(0);
+ Schema secondInnerType = innerTypes.get(1);
+ if ((firstInnerType.getType() != Schema.Type.NULL &&
secondInnerType.getType() != Schema.Type.NULL)
+ || (firstInnerType.getType() == Schema.Type.NULL &&
secondInnerType.getType() == Schema.Type.NULL)) {
+ throw new HoodieAvroSchemaException(
+ String.format("Unsupported Avro UNION type %s: Only UNION of a null
type and a non-null type is supported", schema));
+ }
+ return firstInnerType.getType() == Schema.Type.NULL ? secondInnerType :
firstInnerType;
+ }
+
+ /**
+ * Returns true in case provided {@link Schema} is nullable (ie accepting
null values),
+ * returns false otherwise
+ */
+ private static boolean isNullable(Schema schema) {
+ if (schema.getType() != Schema.Type.UNION) {
+ return false;
+ }
+
+ List<Schema> innerTypes = schema.getTypes();
+ return innerTypes.size() > 1 && innerTypes.stream().anyMatch(it ->
it.getType() == Schema.Type.NULL);
+ }
+
+ /**
+ * Create a new schema but maintain all meta info from the old schema
+ *
+ * @param schema schema to get the meta info from
+ * @param fields list of fields in order that will be in the new schema
+ *
+ * @return schema with fields from fields, and metadata from schema
+ */
+ static Schema createNewSchemaFromFieldsWithReference(Schema schema,
List<Schema.Field> fields) {
+ if (schema == null) {
+ throw new IllegalArgumentException("Schema must not be null");
+ }
+ Schema newSchema = Schema.createRecord(schema.getName(), schema.getDoc(),
schema.getNamespace(), schema.isError());
+ Map<String, Object> schemaProps = Collections.emptyMap();
+ try {
+ schemaProps = schema.getObjectProps();
+ } catch (Exception e) {
+ log.warn("Error while getting object properties from schema: {}",
schema, e);
+ }
+ for (Map.Entry<String, Object> prop : schemaProps.entrySet()) {
+ newSchema.addProp(prop.getKey(), prop.getValue());
+ }
+ newSchema.setFields(fields);
+ return newSchema;
+ }
+
private static Schema removeFields(Schema schema, Set<String>
fieldsToRemove) {
List<Schema.Field> filteredFields = schema.getFields()
.stream()
@@ -394,7 +464,7 @@ public class HoodieAvroUtils {
.map(HoodieAvroUtils::createNewSchemaField)
.collect(Collectors.toList());
- return AvroSchemaUtils.createNewSchemaFromFieldsWithReference(schema,
filteredFields);
+ return createNewSchemaFromFieldsWithReference(schema, filteredFields);
}
@VisibleForTesting
@@ -404,13 +474,13 @@ public class HoodieAvroUtils {
.stream()
.map(field -> {
if (Objects.equals(field.name(), fieldName)) {
- return createNewSchemaField(field.name(),
AvroSchemaUtils.getNonNullTypeFromUnion(field.schema()), field.doc(),
fieldDefaultValue);
+ return createNewSchemaField(field.name(),
getNonNullTypeFromUnion(field.schema()), field.doc(), fieldDefaultValue);
} else {
return createNewSchemaField(field);
}
})
.collect(Collectors.toList());
- return AvroSchemaUtils.createNewSchemaFromFieldsWithReference(schema,
filteredFields);
+ return createNewSchemaFromFieldsWithReference(schema, filteredFields);
}
/**
@@ -757,11 +827,11 @@ public class HoodieAvroUtils {
if (fieldSchema == null) {
return fieldValue;
} else if (fieldValue == null) {
- checkState(AvroSchemaUtils.isNullable(fieldSchema));
+ checkState(isNullable(fieldSchema));
return null;
}
- return
convertValueForAvroLogicalTypes(AvroSchemaUtils.getNonNullTypeFromUnion(fieldSchema),
fieldValue, consistentLogicalTimestampEnabled);
+ return
convertValueForAvroLogicalTypes(getNonNullTypeFromUnion(fieldSchema),
fieldValue, consistentLogicalTimestampEnabled);
}
/**
@@ -924,7 +994,7 @@ public class HoodieAvroUtils {
* @param record Hoodie record.
* @param columns Names of the columns to get values.
* @param schema {@link HoodieSchema} instance.
- * @return Column value.
+ * @return the column values, in the order of {@code columns}.
*/
public static Object[] getRecordColumnValues(HoodieRecord record,
String[] columns,
@@ -950,7 +1020,7 @@ public class HoodieAvroUtils {
* @param record Hoodie record.
* @param columns Names of the columns to get values.
* @param schema {@link HoodieSchema} instance.
- * @return Column value.
+ * @return the column values, in the order of {@code columns}.
*/
public static Object[]
getSortColumnValuesWithPartitionPathAndRecordKey(HoodieRecord record,
String[] columns,
@@ -1070,7 +1140,7 @@ public class HoodieAvroUtils {
newRecord.put(i,
rewriteRecordWithNewSchema(indexedRecord.get(oldField.pos()),
oldField.schema(), newField.schema(), renameCols, fieldNames, false));
} else if (newField.defaultVal() instanceof JsonProperties.Null) {
newRecord.put(i, null);
- } else if (!AvroSchemaUtils.isNullable(newField.schema()) &&
newField.defaultVal() == null) {
+ } else if (!isNullable(newField.schema()) && newField.defaultVal()
== null) {
throw new SchemaCompatibilityException("Field " +
createFullName(fieldNames) + " has no default value and is non-nullable");
} else {
newRecord.put(i, newField.defaultVal());
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java
b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java
index 5f6cec1bc483..53fe10e362e9 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java
@@ -18,7 +18,6 @@
package org.apache.hudi.common.schema;
-import org.apache.hudi.common.avro.AvroSchemaUtils;
import org.apache.hudi.common.schema.internal.HoodieSchemaException;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
@@ -2887,8 +2886,10 @@ public class HoodieSchema implements Serializable {
public static class Blob extends HoodieSchema {
public static final String TYPE_DESCRIPTOR = "BLOB";
private static final String DEFAULT_NAME = "blob";
+ // declared before BLOB_FIELDS: createBlobFields() reads it while the
class is being initialized
+ private static final Schema REFERENCE_SCHEMA = createReferenceSchema();
private static final List<Schema.Field> BLOB_FIELDS = createBlobFields();
- private static final int REFERENCE_FIELD_COUNT =
AvroSchemaUtils.getNonNullTypeFromUnion(BLOB_FIELDS.get(2).schema()).getFields().size();
+ private static final int REFERENCE_FIELD_COUNT =
REFERENCE_SCHEMA.getFields().size();
public static final String INLINE = "INLINE";
public static final String OUT_OF_LINE = "OUT_OF_LINE";
@@ -2953,23 +2954,34 @@ public class HoodieSchema implements Serializable {
return blobSchema;
}
- private static List<Schema.Field> createBlobFields() {
- Schema bytesField = Schema.create(Schema.Type.BYTES);
+ private static Schema createReferenceSchema() {
Schema referenceField = Schema.createRecord(EXTERNAL_REFERENCE, null,
null, false);
List<Schema.Field> referenceFields = Arrays.asList(
new Schema.Field(EXTERNAL_REFERENCE_PATH,
Schema.create(Schema.Type.STRING), null, null),
- new Schema.Field(EXTERNAL_REFERENCE_OFFSET,
AvroSchemaUtils.createNullableSchema(Schema.create(Schema.Type.LONG)), null,
null),
- new Schema.Field(EXTERNAL_REFERENCE_LENGTH,
AvroSchemaUtils.createNullableSchema(Schema.create(Schema.Type.LONG)), null,
null),
+ new Schema.Field(EXTERNAL_REFERENCE_OFFSET,
nullable(Schema.create(Schema.Type.LONG)), null, null),
+ new Schema.Field(EXTERNAL_REFERENCE_LENGTH,
nullable(Schema.create(Schema.Type.LONG)), null, null),
new Schema.Field(EXTERNAL_REFERENCE_IS_MANAGED,
Schema.create(Schema.Type.BOOLEAN), null, null)
);
referenceField.setFields(referenceFields);
+ return referenceField;
+ }
+ private static List<Schema.Field> createBlobFields() {
+ Schema bytesField = Schema.create(Schema.Type.BYTES);
return Arrays.asList(
new Schema.Field(TYPE, Schema.createEnum("blob_storage_type", null,
null, Arrays.asList(INLINE, OUT_OF_LINE)), null, null),
- new Schema.Field(INLINE_DATA_FIELD,
AvroSchemaUtils.createNullableSchema(bytesField), null,
Schema.Field.NULL_DEFAULT_VALUE),
- new Schema.Field(EXTERNAL_REFERENCE,
AvroSchemaUtils.createNullableSchema(referenceField), null,
Schema.Field.NULL_DEFAULT_VALUE)
+ new Schema.Field(INLINE_DATA_FIELD, nullable(bytesField), null,
Schema.Field.NULL_DEFAULT_VALUE),
+ new Schema.Field(EXTERNAL_REFERENCE, nullable(REFERENCE_SCHEMA),
null, Schema.Field.NULL_DEFAULT_VALUE)
);
}
+
+ /**
+ * Wraps the given schema into the canonical Avro nullable union {@code
[null, schema]}. None of the blob
+ * field types is NULL, so no further validation is needed here.
+ */
+ private static Schema nullable(Schema schema) {
+ return Schema.createUnion(Schema.create(Schema.Type.NULL), schema);
+ }
}
private void writeObject(ObjectOutputStream oos) throws IOException {
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaCompatibility.java
b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaCompatibility.java
index 3c98b3b45ee1..e9a0aea0d9d0 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaCompatibility.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaCompatibility.java
@@ -51,6 +51,20 @@ import java.util.stream.Collectors;
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public final class HoodieSchemaCompatibility {
+ /**
+ * Establishes whether data written with {@code writerSchema} can be read
with {@code tableSchema}.
+ *
+ * <p>NOTE: the reader is the FIRST argument here, the opposite order from
+ * {@link #isSchemaCompatible(HoodieSchema, HoodieSchema, boolean,
boolean)}, where the reader is the
+ * second argument. Schema fully-qualified names are NOT checked. Unlike
{@code isSchemaCompatible}
+ * with {@code allowProjection=false}, no missing-field (projection)
pre-check is applied: a reader
+ * that simply drops writer fields is reported as compatible.</p>
+ *
+ * @param tableSchema the schema used to read the data
+ * @param writerSchema the schema the data was written with
+ * @return true if tableSchema can read data written with writerSchema
+ * @see #isSchemaCompatible(HoodieSchema, HoodieSchema, boolean, boolean)
+ */
public static boolean areSchemasCompatible(HoodieSchema tableSchema,
HoodieSchema writerSchema) {
return
HoodieSchemaCompatibilityChecker.checkReaderWriterCompatibility(tableSchema,
writerSchema, false).getType() ==
HoodieSchemaCompatibilityChecker.SchemaCompatibilityType.COMPATIBLE;
}
@@ -205,18 +219,21 @@ public final class HoodieSchemaCompatibility {
}
/**
- * Checks if two schemas are compatible with projection support.
- * This allows the reader schema to have fewer fields than the writer schema.
+ * Establishes whether {@code newSchema} is compatible w/ {@code
prevSchema}, checking schemas
+ * fully-qualified names.
+ * From avro's compatibility standpoint, prevSchema is the writer schema and
newSchema is the reader schema.
+ * {@code newSchema} is considered compatible to {@code prevSchema}, iff
data written using {@code prevSchema}
+ * could be read by {@code newSchema}
*
- * @param readerSchema the schema used to read the data
- * @param writerSchema the schema used to write the data
+ * @param prevSchema previous instance of the schema
+ * @param newSchema new instance of the schema
* @param allowProjection whether to allow fewer fields in reader schema
* @return true if reader schema can read data written with writer schema
* @throws IllegalArgumentException if schemas are null
*/
- public static boolean isSchemaCompatible(HoodieSchema readerSchema,
HoodieSchema writerSchema,
+ public static boolean isSchemaCompatible(HoodieSchema prevSchema,
HoodieSchema newSchema,
boolean allowProjection) {
- return isSchemaCompatible(readerSchema, writerSchema, true,
allowProjection);
+ return isSchemaCompatible(prevSchema, newSchema, true, allowProjection);
}
/**
@@ -281,25 +298,12 @@ public final class HoodieSchemaCompatibility {
* @param writerSchema Schema of the record where to look for the writer
field.
* @param readerField Reader field to identify the corresponding writer
field
* of.
- * @return the writer field, if any does correspond, or None.
+ * @return the writer field, if any does correspond, or null.
+ * @throws IllegalArgumentException if {@code writerSchema} is not a record
*/
public static HoodieSchemaField lookupWriterField(final HoodieSchema
writerSchema, final HoodieSchemaField readerField) {
ValidationUtils.checkArgument(writerSchema.getType() ==
HoodieSchemaType.RECORD, writerSchema + " is not a record");
- Option<HoodieSchemaField> directOpt =
writerSchema.getField(readerField.name());
- // Check aliases
- for (final String readerFieldAliasName :
readerField.getAvroField().aliases()) {
- final Option<HoodieSchemaField> writerFieldOpt =
writerSchema.getField(readerFieldAliasName);
- if (writerFieldOpt.isPresent()) {
- if (directOpt.isPresent()) {
- // Multiple matches found, fail fast
- throw new HoodieSchemaException(String.format(
- "Reader record field %s matches multiple fields in writer record
schema %s", readerField, writerSchema));
- }
- directOpt = writerFieldOpt;
- }
- }
-
- return directOpt.orElse(null);
+ return HoodieSchemaCompatibilityChecker.lookupWriterField(writerSchema,
readerField);
}
/**
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaCompatibilityChecker.java
b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaCompatibilityChecker.java
index f92c043c8b45..28c3dc4eafda 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaCompatibilityChecker.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaCompatibilityChecker.java
@@ -112,7 +112,7 @@ public class HoodieSchemaCompatibilityChecker {
* @param writerSchema Schema of the record where to look for the writer
field.
* @param readerField Reader field to identify the corresponding writer
field
* of.
- * @return the writer field, if any does correspond, or None.
+ * @return the writer field, if any does correspond, or null.
*/
public static HoodieSchemaField lookupWriterField(final HoodieSchema
writerSchema, final HoodieSchemaField readerField) {
assert (writerSchema.hasFields());
@@ -357,24 +357,26 @@ public class HoodieSchemaCompatibilityChecker {
case DATE:
case DECIMAL:
return result.mergedWith(typeMismatch(reader, writer, locations));
+ // TIMESTAMP over LONG and UUID over STRING are reader/writer
compatibility rules only. They are deliberately
+ // absent from HoodieSchemaTypePromotion:
isCompatibleProjectionOf(source, target) tests
+ // canPromote(target, source), so the entry would make a timestamp a
compatible projection of a bare long,
+ // and writer-schema deduction (HoodieSchemaUtils.scala) would then
keep the table's long as the writer
+ // schema and silently drop the logical type. The deduction that
produces a TIMESTAMP-reader / LONG-writer
+ // pair is gated per field by
hoodie.write.timestamp.logical.type.overrides (#19384); this checker accepts
+ // the pair once produced.
case TIMESTAMP:
return (writer.getType() == HoodieSchemaType.LONG) ? result :
result.mergedWith(typeMismatch(reader, writer, locations));
case UUID:
return (writer.getType() == HoodieSchemaType.STRING) ? result :
result.mergedWith(typeMismatch(reader, writer, locations));
+ // The primitive widening table (LONG <- INT, FLOAT <- INT/LONG,
DOUBLE <- INT/LONG/FLOAT, BYTES <- STRING,
+ // STRING <- BYTES or any numeric) lives in
HoodieSchemaTypePromotion, shared with the projection checker.
case LONG:
- return (writer.getType() == HoodieSchemaType.INT) ? result :
result.mergedWith(typeMismatch(reader, writer, locations));
case FLOAT:
- return ((writer.getType() == HoodieSchemaType.INT) ||
(writer.getType() == HoodieSchemaType.LONG)) ? result
- : result.mergedWith(typeMismatch(reader, writer, locations));
case DOUBLE:
- return ((writer.getType() == HoodieSchemaType.INT) ||
(writer.getType() == HoodieSchemaType.LONG) || (writer.getType() ==
HoodieSchemaType.FLOAT))
- ? result
- : result.mergedWith(typeMismatch(reader, writer, locations));
case BYTES:
- return (writer.getType() == HoodieSchemaType.STRING) ? result :
result.mergedWith(typeMismatch(reader, writer, locations));
case STRING:
- return (writer.getType().isNumeric() || (writer.getType() ==
HoodieSchemaType.BYTES)
- ? result : result.mergedWith(typeMismatch(reader, writer,
locations)));
+ return HoodieSchemaTypePromotion.canPromote(reader.getType(),
writer.getType())
+ ? result : result.mergedWith(typeMismatch(reader, writer,
locations));
case ARRAY:
return result.mergedWith(typeMismatch(reader, writer, locations));
case MAP:
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaTypePromotion.java
b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaTypePromotion.java
index 13ae75f194a6..b70e2c4903ae 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaTypePromotion.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaTypePromotion.java
@@ -19,7 +19,8 @@
package org.apache.hudi.common.schema;
/**
- * The single table of primitive widening promotions, used by {@link
HoodieSchemaProjectionChecker}.
+ * The single table of primitive widening promotions, used by {@link
HoodieSchemaProjectionChecker} and, for the
+ * primitive cases, by {@link HoodieSchemaCompatibilityChecker}.
*
* <p>A promotion lets a reader schema with a wider type read data written
with a narrower one:</p>
* <ul>
@@ -33,16 +34,16 @@ package org.apache.hudi.common.schema;
*
* <p>Logical-type-over-primitive promotions are deliberately NOT in this
table. A TIMESTAMP reader over a
* LONG writer, or a UUID reader over a STRING writer, is accepted by
- * {@link HoodieSchemaCompatibilityChecker} for reader/writer compatibility,
but it must not make a bare
- * long a "compatible projection" of a timestamp: writer-schema deduction
would then silently drop the
- * logical type. Compatibility and projection are different questions, so they
use different tables.</p>
+ * {@link HoodieSchemaCompatibilityChecker} for reader/writer compatibility,
but it must not make a
+ * timestamp a "compatible projection" of a bare long -- {@code
isCompatibleProjectionOf(source, target)}
+ * tests {@code canPromote(target, source)}, so the entry would let
writer-schema deduction keep the
+ * table's long as the writer schema and silently drop the logical type.
Compatibility and projection are
+ * different questions, so they use different tables.</p>
*
* <p>One more difference is documented rather than resolved:
* {@link #isDecimalWidening(HoodieSchema, HoodieSchema)} additionally
requires the same backing (fixed
* versus bytes) and, for fixed, an equal fixed size, whereas the decimal
check in
* {@code HoodieSchemaCompatibilityChecker} compares only precision and
scale.</p>
- *
- * <p>This class is package-private and used only by {@link
HoodieSchemaProjectionChecker}.</p>
*/
class HoodieSchemaTypePromotion {
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaUtils.java
index fea4bc3ae8fa..ac2de5fe79fc 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaUtils.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaUtils.java
@@ -18,10 +18,14 @@
package org.apache.hudi.common.schema;
-import org.apache.hudi.common.avro.AvroSchemaUtils;
import org.apache.hudi.common.avro.HoodieAvroUtils;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.schema.internal.HoodieSchemaException;
+import org.apache.hudi.common.schema.internal.InternalSchema;
+import org.apache.hudi.common.schema.internal.action.TableChanges;
+import org.apache.hudi.common.schema.internal.convert.InternalSchemaConverter;
+import org.apache.hudi.common.schema.internal.utils.SchemaChangeUtils;
+import org.apache.hudi.common.util.CollectionUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.VisibleForTesting;
@@ -64,6 +68,7 @@ import java.util.stream.Stream;
* {@link #toJavaDefaultValue(HoodieSchemaField)}</li>
* <li>nullability: {@link #asNullable(HoodieSchema)}</li>
* <li>naming: {@link #sanitizeName(String)}, {@link
#getRecordQualifiedName(String)}</li>
+ * <li>error text: {@link #createSchemaErrorString(String, HoodieSchema,
HoodieSchema)}</li>
* <li>lookups and predicates that need more than {@link HoodieSchema}
offers on its own:
* {@link #findNestedField(HoodieSchema, String)}, {@link
#findMissingFields(HoodieSchema, HoodieSchema)},
* {@link #resolveUnionSchema(HoodieSchema, String)}, {@link
#hasDecimalField(HoodieSchema)}</li>
@@ -80,11 +85,12 @@ import java.util.stream.Stream;
* <li>the field-id InternalSchema (schema-on-read) domain: {@code
org.apache.hudi.common.schema.internal}</li>
* </ul>
*
- * <p>A few methods here still delegate to Avro-typed implementations
- * ({@link #asNullable(HoodieSchema)}, {@link
#createNullableSchema(HoodieSchema)},
- * {@link #projectSchema(HoodieSchema, List)} and the 5-arg
- * {@link #createNewSchemaField(String, HoodieSchema, String, Object,
HoodieFieldOrder)}). Those delegations
- * are being retired under #16639; new methods must be implemented on
HoodieSchema directly.</p>
+ * <p>A couple of methods here still delegate to Avro-typed implementations
+ * ({@link #projectSchema(HoodieSchema, List)} and the 5-arg
+ * {@link #createNewSchemaField(String, HoodieSchema, String, Object,
HoodieFieldOrder)}). An internal
+ * toAvroSchema/fromAvroSchema hop is not one of the conversion boundaries
RFC-99 allows (memory to disk,
+ * disk to memory, engine boundary), so those delegations are being retired
under #14263; new methods must
+ * be implemented on HoodieSchema directly. Where a helper belongs, by
contrast, is #16639.</p>
*
* @since 1.2.0
*/
@@ -248,36 +254,46 @@ public final class HoodieSchemaUtils {
}
/**
- * Creates a nullable version of the given schema (union of null and the
schema).
- *
- * <p>{@link HoodieSchema#createNullable(HoodieSchema)} is the idempotent
native equivalent and is
- * preferred; this overload round-trips through Avro and is retained only
for existing call sites.</p>
- *
- * @param schema the input schema
- * @return new HoodieSchema that allows null values
- * @throws IllegalArgumentException if schema is null
- */
- public static HoodieSchema createNullableSchema(HoodieSchema schema) {
- ValidationUtils.checkArgument(schema != null, "Schema cannot be null");
-
- // Delegate to AvroSchemaUtils
- Schema nullableAvro =
AvroSchemaUtils.createNullableSchema(schema.toAvroSchema());
- return HoodieSchema.fromAvroSchema(nullableAvro);
- }
-
- /**
- * Create a new schema by force changing all the fields as nullable.
- *
- * @return a new schema with all the fields updated as nullable
- * @throws IllegalArgumentException if schema is null
- * @see AvroSchemaUtils#asNullable(Schema)
+ * Create a new schema by force changing all the top-level fields as
nullable.
+ *
+ * <p>The rewrite runs through the field-id {@link InternalSchema}: the
record is converted, every
+ * still-required top-level field is marked nullable with a {@link
TableChanges.ColumnUpdateChange},
+ * and the updated InternalSchema is converted back under the original full
name. Only the top level
+ * changes - the inner fields of a nested record keep the nullability they
had. Because the record is
+ * rebuilt from the InternalSchema, its full name, field order and per-field
docs survive, while the
+ * record-level doc and any custom record properties do not. Three more
effects of that round trip are
+ * pre-existing and pinned by tests: a non-null field default becomes {@code
null}, an ENUM field comes
+ * back as STRING, and an already-nullable null-last union is reordered
null-first.</p>
+ *
+ * <p>When every top-level field is already nullable the input instance
itself is returned and no
+ * conversion runs.</p>
+ *
+ * @param schema original schema
+ * @return a schema with all the top-level fields updated as nullable, or
{@code schema} itself when
+ * there is nothing to change
+ * @throws IllegalArgumentException if schema is null or not a RECORD
*/
public static HoodieSchema asNullable(HoodieSchema schema) {
ValidationUtils.checkArgument(schema != null, "Schema cannot be null");
+ ValidationUtils.checkArgument(schema.getType() == HoodieSchemaType.RECORD,
+ "asNullable expects a RECORD schema, got: " + schema.getType());
+
+ // NOTE: HoodieSchema#isNullable is false for a bare NULL type, unlike
Avro's Schema#isNullable, so a
+ // NULL-typed field is excluded explicitly to keep it out of the
update list as it always was.
+ List<String> requiredCols = schema.getFields().stream()
+ .filter(f -> !(f.schema().isNullable() || f.schema().getType() ==
HoodieSchemaType.NULL))
+ .map(HoodieSchemaField::name)
+ .collect(Collectors.toList());
+ if (requiredCols.isEmpty()) {
+ return schema;
+ }
- // Delegate to AvroSchemaUtils
- Schema nullableAvro = AvroSchemaUtils.asNullable(schema.toAvroSchema());
- return HoodieSchema.fromAvroSchema(nullableAvro);
+ InternalSchema internalSchema = InternalSchemaConverter.convert(schema);
+ TableChanges.ColumnUpdateChange schemaChange =
TableChanges.ColumnUpdateChange.get(internalSchema);
+ schemaChange = CollectionUtils.reduce(requiredCols, schemaChange,
+ (change, field) -> change.updateColumnNullability(field, true));
+ return InternalSchemaConverter.convert(
+ SchemaChangeUtils.applyTableChanges2Schema(internalSchema,
schemaChange), schema.getFullName());
}
/**
@@ -398,7 +414,7 @@ public final class HoodieSchemaUtils {
* Alias of {@link HoodieSchemaField#of(String, HoodieSchema, String,
Object, HoodieFieldOrder)} with
* argument validation. Prefer {@code HoodieSchemaField.of} directly in new
code; this overload still
* round-trips through the Avro-typed {@code
HoodieAvroUtils#createNewSchemaField} and is being retired
- * under #16639.
+ * under #14263.
*
* @param name field name
* @param schema field schema
@@ -948,6 +964,15 @@ public final class HoodieSchemaUtils {
return
INVALID_AVRO_CHARS_IN_NAMES_PATTERN.matcher(name).replaceAll(invalidCharMask);
}
+ /**
+ * Formats a schema error with the writer and table schemas appended on
their own lines, so callers
+ * throwing {@code SchemaCompatibilityException} report both sides in a
consistent shape.
+ *
+ * @param errorMessage the message to lead with
+ * @param writerSchema the incoming writer schema
+ * @param tableSchema the current table schema
+ * @return the message followed by both schemas, one per line
+ */
public static String createSchemaErrorString(String errorMessage,
HoodieSchema writerSchema, HoodieSchema tableSchema) {
return String.format("%s\nwriterSchema: %s\ntableSchema: %s",
errorMessage, writerSchema, tableSchema);
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
index 6c87d584765a..655fa6e58a9a 100644
---
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
+++
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
@@ -328,6 +328,9 @@ public class HoodieTableMetadataUtil {
Object fieldValue;
HoodieSchemaType fieldSchemaType = fieldSchema.getType();
if (record.getRecordType() == HoodieRecordType.AVRO) {
+ // Deliberately the static helper rather than record.getColumnValues,
which the SPARK branch below uses:
+ // HoodieAvroIndexedRecord also reports type AVRO but overrides
getColumnValues with a decode plus
+ // AvroRecordContext.getFieldValueFromIndexedRecord, so dispatching
would read nested fields differently.
fieldValue = HoodieAvroUtils.getRecordColumnValues(record, new
String[]{fieldName}, recordSchema, false)[0];
if (fieldValue != null && fieldSchemaType.equals(HoodieSchemaType.DATE))
{
fieldValue = java.sql.Date.valueOf(fieldValue.toString());
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/avro/TestAvroSchemaUtils.java
b/hudi-common/src/test/java/org/apache/hudi/common/avro/TestAvroSchemaUtils.java
deleted file mode 100644
index 9ff2d9fb768e..000000000000
---
a/hudi-common/src/test/java/org/apache/hudi/common/avro/TestAvroSchemaUtils.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * 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.common.avro;
-
-import org.apache.avro.Schema;
-import org.junit.jupiter.api.Test;
-
-import java.util.Collections;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-
-public class TestAvroSchemaUtils {
-
- @Test
- public void testCreateNewSchemaFromFieldsWithReference_NullSchema() {
- // This test should throw an IllegalArgumentException
- assertThrows(IllegalArgumentException.class, () ->
AvroSchemaUtils.createNewSchemaFromFieldsWithReference(null,
Collections.emptyList()));
- }
-
- @Test
- public void testCreateNewSchemaFromFieldsWithReference_NullObjectProps() {
- // Create a schema without any object properties
- String schemaStr = "{ \"type\": \"record\", \"name\": \"TestRecord\",
\"fields\": [] }";
- Schema schema = new Schema.Parser().parse(schemaStr);
-
- // Ensure getObjectProps returns null by mocking or creating a schema
without props
- Schema newSchema =
AvroSchemaUtils.createNewSchemaFromFieldsWithReference(schema,
Collections.emptyList());
-
- // Validate the new schema
- assertEquals("TestRecord", newSchema.getName());
- assertEquals(0, newSchema.getFields().size());
- }
-
- @Test
- public void testCreateNewSchemaFromFieldsWithReference_WithObjectProps() {
- // Create a schema with object properties
- String schemaStr = "{ \"type\": \"record\", \"name\": \"TestRecord\",
\"fields\": [], \"prop1\": \"value1\" }";
- Schema schema = new Schema.Parser().parse(schemaStr);
-
- // Add an object property to the schema
- schema.addProp("prop1", "value1");
-
- // Create new fields to add
- Schema.Field newField = new Schema.Field("newField",
Schema.create(Schema.Type.STRING), null, (Object) null);
- Schema newSchema =
AvroSchemaUtils.createNewSchemaFromFieldsWithReference(schema,
Collections.singletonList(newField));
-
- // Validate the new schema
- assertEquals("TestRecord", newSchema.getName());
- assertEquals(1, newSchema.getFields().size());
- assertEquals("value1", newSchema.getProp("prop1"));
- assertEquals("newField", newSchema.getFields().get(0).name());
- }
-}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/avro/TestHoodieAvroUtils.java
b/hudi-common/src/test/java/org/apache/hudi/common/avro/TestHoodieAvroUtils.java
index d59171dd653a..f899e585b46c 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/avro/TestHoodieAvroUtils.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/avro/TestHoodieAvroUtils.java
@@ -68,6 +68,7 @@ import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaType;
import org.apache.hudi.common.schema.HoodieSchemaUtils;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieAvroSchemaException;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.SchemaCompatibilityException;
@@ -119,7 +120,7 @@ import java.util.Random;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import static
org.apache.hudi.common.avro.AvroSchemaUtils.getNonNullTypeFromUnion;
+import static
org.apache.hudi.common.avro.HoodieAvroUtils.getNonNullTypeFromUnion;
import static
org.apache.hudi.common.avro.HoodieAvroWrapperUtils.unwrapAvroValueWrapper;
import static
org.apache.hudi.common.avro.HoodieAvroWrapperUtils.wrapValueIntoAvro;
import static org.apache.hudi.common.schema.HoodieSchemaUtils.sanitizeName;
@@ -129,6 +130,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -1366,4 +1368,74 @@ public class TestHoodieAvroUtils {
// the pom.properties lookup must agree with the jar that actually defines
Schema
assertEquals(Schema.class.getPackage().getImplementationVersion(),
HoodieAvroUtils.AVRO_VERSION);
}
+
+ @Test
+ public void testCreateNewSchemaFromFieldsWithReference_NullSchema() {
+ // This test should throw an IllegalArgumentException
+ assertThrows(IllegalArgumentException.class, () ->
HoodieAvroUtils.createNewSchemaFromFieldsWithReference(null,
Collections.emptyList()));
+ }
+
+ @Test
+ public void testCreateNewSchemaFromFieldsWithReference_NullObjectProps() {
+ // Create a schema without any object properties
+ String schemaStr = "{ \"type\": \"record\", \"name\": \"TestRecord\",
\"fields\": [] }";
+ Schema schema = new Schema.Parser().parse(schemaStr);
+
+ // Ensure getObjectProps returns null by mocking or creating a schema
without props
+ Schema newSchema =
HoodieAvroUtils.createNewSchemaFromFieldsWithReference(schema,
Collections.emptyList());
+
+ // Validate the new schema
+ assertEquals("TestRecord", newSchema.getName());
+ assertEquals(0, newSchema.getFields().size());
+ }
+
+ @Test
+ public void testCreateNewSchemaFromFieldsWithReference_WithObjectProps() {
+ // Create a schema with object properties
+ String schemaStr = "{ \"type\": \"record\", \"name\": \"TestRecord\",
\"fields\": [], \"prop1\": \"value1\" }";
+ Schema schema = new Schema.Parser().parse(schemaStr);
+
+ // Add an object property to the schema
+ schema.addProp("prop1", "value1");
+
+ // Create new fields to add
+ Schema.Field newField = new Schema.Field("newField",
Schema.create(Schema.Type.STRING), null, (Object) null);
+ Schema newSchema =
HoodieAvroUtils.createNewSchemaFromFieldsWithReference(schema,
Collections.singletonList(newField));
+
+ // Validate the new schema
+ assertEquals("TestRecord", newSchema.getName());
+ assertEquals(1, newSchema.getFields().size());
+ assertEquals("value1", newSchema.getProp("prop1"));
+ assertEquals("newField", newSchema.getFields().get(0).name());
+ }
+
+ @Test
+ public void testGetNonNullTypeFromUnionReturnsNonUnionAsIs() {
+ Schema intSchema = Schema.create(Schema.Type.INT);
+ assertSame(intSchema, getNonNullTypeFromUnion(intSchema));
+ }
+
+ @Test
+ public void testGetNonNullTypeFromUnionUnwrapsBothBranchOrders() {
+ Schema intSchema = Schema.create(Schema.Type.INT);
+ Schema nullSchema = Schema.create(Schema.Type.NULL);
+ assertSame(intSchema,
getNonNullTypeFromUnion(Schema.createUnion(nullSchema, intSchema)));
+ assertSame(intSchema,
getNonNullTypeFromUnion(Schema.createUnion(intSchema, nullSchema)));
+ }
+
+ @Test
+ public void testGetNonNullTypeFromUnionRejectsUnsupportedUnions() {
+ List<Schema> unsupported = Arrays.asList(
+ // more than two branches, even with a null one
+ Schema.createUnion(Schema.create(Schema.Type.NULL),
Schema.create(Schema.Type.INT), Schema.create(Schema.Type.STRING)),
+ // exactly two branches, but neither of them is null
+ Schema.createUnion(Schema.create(Schema.Type.INT),
Schema.create(Schema.Type.STRING)),
+ // a single branch is not a nullable union either
+ Schema.createUnion(Schema.create(Schema.Type.INT)));
+ for (Schema schema : unsupported) {
+ HoodieAvroSchemaException e =
+ assertThrows(HoodieAvroSchemaException.class, () ->
getNonNullTypeFromUnion(schema));
+ assertTrue(e.getMessage().contains("Only UNION of a null type and a
non-null type"), e.getMessage());
+ }
+ }
}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchema.java
b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchema.java
index fded6184026a..c5e9b8b090b2 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchema.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchema.java
@@ -75,6 +75,17 @@ public class TestHoodieSchema {
+ " ]"
+ "}";
+ // The Blob schema is persisted in the table schema of every BLOB table, so
its serialized shape is pinned literally.
+ private static final String EXPECTED_BLOB_SCHEMA_JSON =
"{\"type\":\"record\",\"name\":\"blob\",\"fields\":["
+ +
"{\"name\":\"type\",\"type\":{\"type\":\"enum\",\"name\":\"blob_storage_type\",\"symbols\":[\"INLINE\",\"OUT_OF_LINE\"]}},"
+ + "{\"name\":\"data\",\"type\":[\"null\",\"bytes\"],\"default\":null},"
+ +
"{\"name\":\"reference\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"reference\",\"fields\":["
+ + "{\"name\":\"external_path\",\"type\":\"string\"},"
+ + "{\"name\":\"offset\",\"type\":[\"null\",\"long\"]},"
+ + "{\"name\":\"length\",\"type\":[\"null\",\"long\"]},"
+ + "{\"name\":\"managed\",\"type\":\"boolean\"}]}],\"default\":null}],"
+ + "\"logicalType\":\"blob\"}";
+
/**
* Checks if the given Avro schema is a Variant schema. This checks for the
Variant logical type.
*
@@ -2485,8 +2496,10 @@ public class TestHoodieSchema {
@Test
public void testBlobFieldCountMethods() {
- assertTrue(HoodieSchema.Blob.getFieldCount() > 0);
- assertTrue(HoodieSchema.Blob.getReferenceFieldCount() > 0);
+ // type, data, reference
+ assertEquals(3, HoodieSchema.Blob.getFieldCount());
+ // external_path, offset, length, managed
+ assertEquals(4, HoodieSchema.Blob.getReferenceFieldCount());
}
@Test
@@ -2550,6 +2563,10 @@ public class TestHoodieSchema {
assertTrue(managedOpt.isPresent());
assertEquals(HoodieSchemaType.BOOLEAN,
managedOpt.get().schema().getType());
assertFalse(managedOpt.get().schema().isNullable());
+
+ // Enum symbols, null-first unions, the null defaults on data/reference
(and none on offset/length),
+ // the reference record name and the blob logical type, pinned as the
persisted string.
+ assertEquals(EXPECTED_BLOB_SCHEMA_JSON, blob.toAvroSchema().toString());
}
@Test
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaCompatibility.java
b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaCompatibility.java
index 5be6716c2e7a..67ba3059228d 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaCompatibility.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaCompatibility.java
@@ -19,6 +19,7 @@
package org.apache.hudi.common.schema;
import org.apache.hudi.common.avro.VariantSchemaUtils;
+import org.apache.hudi.common.schema.internal.HoodieSchemaException;
import org.apache.hudi.exception.SchemaBackwardsCompatibilityException;
import org.apache.hudi.exception.SchemaCompatibilityException;
@@ -42,6 +43,7 @@ import static
org.apache.hudi.common.schema.TestHoodieSchemaUtils.SIMPLE_SCHEMA;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -701,6 +703,157 @@ public class TestHoodieSchemaCompatibility {
assertFalse(HoodieSchemaCompatibility.isSchemaCompatible(longS, intS,
true, true));
}
+ /**
+ * Sibling of {@link #testIsSchemaCompatibleWithTypePromotion()} covering
the rest of the reader/writer type
+ * table: the primitive widening cases shared with {@link
HoodieSchemaTypePromotion}, plus the two
+ * logical-type-over-primitive rules (TIMESTAMP over LONG, UUID over STRING)
that are compatibility-only.
+ *
+ * <p>All pairs are asserted through the 4-arg {@code
isSchemaCompatible(prev = writer, new = reader, true, true)}.</p>
+ */
+ @Test
+ public void testIsSchemaCompatibleWithLogicalTypesAndWidening() {
+ // Logical type over its backing primitive: accepted for reader/writer
compatibility.
+ assertCompatible(HoodieSchema.createTimestampMillis(),
HoodieSchema.create(HoodieSchemaType.LONG));
+ assertCompatible(HoodieSchema.createUUID(),
HoodieSchema.create(HoodieSchemaType.STRING));
+
+ // ... but only in that direction, and only over the matching primitive.
+ assertIncompatible(HoodieSchema.create(HoodieSchemaType.LONG),
HoodieSchema.createTimestampMillis());
+ assertIncompatible(HoodieSchema.createTimestampMillis(),
HoodieSchema.create(HoodieSchemaType.INT));
+ // DATE has no such rule at all, even though it is backed by INT.
+ assertIncompatible(HoodieSchema.createDate(),
HoodieSchema.create(HoodieSchemaType.INT));
+
+ // Primitive widening, delegated to HoodieSchemaTypePromotion.
+ assertCompatible(HoodieSchema.create(HoodieSchemaType.DOUBLE),
HoodieSchema.create(HoodieSchemaType.FLOAT));
+ assertIncompatible(HoodieSchema.create(HoodieSchemaType.FLOAT),
HoodieSchema.create(HoodieSchemaType.DOUBLE));
+ // The narrowing direction is rejected for every numeric pair. The
hudi-spark guard for the reversed
+ // argument bug class (TestTableSchemaEvolution, HUDI-1493) never runs, so
the pairs are pinned here.
+ assertIncompatible(HoodieSchema.create(HoodieSchemaType.INT),
HoodieSchema.create(HoodieSchemaType.LONG));
+ assertIncompatible(HoodieSchema.create(HoodieSchemaType.INT),
HoodieSchema.create(HoodieSchemaType.FLOAT));
+ assertIncompatible(HoodieSchema.create(HoodieSchemaType.INT),
HoodieSchema.create(HoodieSchemaType.DOUBLE));
+ assertIncompatible(HoodieSchema.create(HoodieSchemaType.LONG),
HoodieSchema.create(HoodieSchemaType.FLOAT));
+ assertIncompatible(HoodieSchema.create(HoodieSchemaType.LONG),
HoodieSchema.create(HoodieSchemaType.DOUBLE));
+ assertCompatible(HoodieSchema.create(HoodieSchemaType.STRING),
HoodieSchema.create(HoodieSchemaType.BYTES));
+ assertCompatible(HoodieSchema.create(HoodieSchemaType.BYTES),
HoodieSchema.create(HoodieSchemaType.STRING));
+ assertCompatible(HoodieSchema.create(HoodieSchemaType.STRING),
HoodieSchema.create(HoodieSchemaType.INT));
+ }
+
+ @Test
+ public void testAreSchemasCompatibleReaderIsFirstArgument() {
+ HoodieSchema longRecord = HoodieSchemaTestUtils.createRecord("R",
HoodieSchemaField.of("f", HoodieSchema.create(HoodieSchemaType.LONG), null,
null));
+ HoodieSchema intRecord = HoodieSchemaTestUtils.createRecord("R",
HoodieSchemaField.of("f", HoodieSchema.create(HoodieSchemaType.INT), null,
null));
+
+ // A long reader can read int data ...
+ assertTrue(HoodieSchemaCompatibility.areSchemasCompatible(longRecord,
intRecord));
+ // ... but not the other way round, which pins the reader as the FIRST
argument.
+ assertFalse(HoodieSchemaCompatibility.areSchemasCompatible(intRecord,
longRecord));
+ }
+
+ @Test
+ public void testLookupWriterFieldDirectMatch() {
+ HoodieSchemaField readerField = readerFieldWithAlias();
+ HoodieSchema writerSchema =
HoodieSchema.parse("{\"type\":\"record\",\"name\":\"W\",\"fields\":["
+ + "{\"name\":\"a\",\"type\":\"int\"}]}");
+
+ HoodieSchemaField writerField =
HoodieSchemaCompatibility.lookupWriterField(writerSchema, readerField);
+ assertEquals("a", writerField.name());
+ }
+
+ /**
+ * The alias path is also driven end to end through the sole production
caller,
+ * {@code HoodieTable#validateSecondaryIndexSchemaEvolution} (see
+ * {@code TestHoodieTableSchemaEvolution#testFieldWithAlias}, whose
type-change half is what makes the alias
+ * match decide the outcome: an unresolved alias returns null and is skipped
by that caller's
+ * {@code writerField != null} guard); this case pins the facade on its own.
+ */
+ @Test
+ public void testLookupWriterFieldAliasMatch() {
+ HoodieSchemaField readerField = readerFieldWithAlias();
+ HoodieSchema writerSchema =
HoodieSchema.parse("{\"type\":\"record\",\"name\":\"W\",\"fields\":["
+ + "{\"name\":\"old_a\",\"type\":\"int\"}]}");
+
+ HoodieSchemaField writerField =
HoodieSchemaCompatibility.lookupWriterField(writerSchema, readerField);
+ assertEquals("old_a", writerField.name());
+ }
+
+ @Test
+ public void testLookupWriterFieldAmbiguousMatchThrows() {
+ HoodieSchemaField readerField = readerFieldWithAlias();
+ HoodieSchema writerSchema =
HoodieSchema.parse("{\"type\":\"record\",\"name\":\"W\",\"fields\":["
+ + "{\"name\":\"a\",\"type\":\"int\"},"
+ + "{\"name\":\"old_a\",\"type\":\"int\"}]}");
+
+ assertThrows(HoodieSchemaException.class,
+ () -> HoodieSchemaCompatibility.lookupWriterField(writerSchema,
readerField));
+ }
+
+ @Test
+ public void testLookupWriterFieldNoMatchReturnsNull() {
+ HoodieSchemaField readerField = readerFieldWithAlias();
+ HoodieSchema writerSchema =
HoodieSchema.parse("{\"type\":\"record\",\"name\":\"W\",\"fields\":["
+ + "{\"name\":\"unrelated\",\"type\":\"int\"}]}");
+
+ assertNull(HoodieSchemaCompatibility.lookupWriterField(writerSchema,
readerField));
+ }
+
+ @Test
+ public void testLookupWriterFieldRejectsNonRecordWriterSchema() {
+ HoodieSchemaField readerField = readerFieldWithAlias();
+ HoodieSchema notARecord = HoodieSchema.create(HoodieSchemaType.STRING);
+
+ assertThrows(IllegalArgumentException.class,
+ () -> HoodieSchemaCompatibility.lookupWriterField(notARecord,
readerField));
+ }
+
+ /**
+ * Reader field {@code a}, aliased {@code old_a}. Aliases have no builder on
HoodieSchemaField, so the
+ * reader record is parsed from JSON.
+ */
+ private static HoodieSchemaField readerFieldWithAlias() {
+ HoodieSchema readerSchema =
HoodieSchema.parse("{\"type\":\"record\",\"name\":\"R\",\"fields\":["
+ + "{\"name\":\"a\",\"type\":\"int\",\"aliases\":[\"old_a\"]}]}");
+ return readerSchema.getField("a").get();
+ }
+
+ /**
+ * Asserts that a field written as {@code writerFieldSchema} can be read
back as {@code readerFieldSchema},
+ * that is, the {@code writer -> reader} conversion is allowed.
+ *
+ * <p>Arguments are in reader-first ("to", "from") order, matching
+ * {@link HoodieSchemaCompatibility#areSchemasCompatible(HoodieSchema,
HoodieSchema)}. That is the reverse
+ * of the (prev = writer, new = reader) order taken by
+ * {@link HoodieSchemaCompatibility#isSchemaCompatible(HoodieSchema,
HoodieSchema, boolean, boolean)};
+ * the helper flips the pair before calling it.</p>
+ *
+ * <p>Both schemas are wrapped in a single-field record {@code R { f }}, so
only the per-field type rules
+ * are exercised here. Naming checks and projection are both enabled; field
addition, removal and renaming
+ * are covered by the other tests in this class.</p>
+ *
+ * @param readerFieldSchema type the field is read as (the "to" side of the
conversion)
+ * @param writerFieldSchema type the field was written with (the "from" side)
+ */
+ private static void assertCompatible(HoodieSchema readerFieldSchema,
HoodieSchema writerFieldSchema) {
+ assertTrue(HoodieSchemaCompatibility.isSchemaCompatible(
+ HoodieSchemaTestUtils.createRecord("R", HoodieSchemaField.of("f",
writerFieldSchema, null, null)),
+ HoodieSchemaTestUtils.createRecord("R", HoodieSchemaField.of("f",
readerFieldSchema, null, null)), true, true),
+ "reader " + readerFieldSchema + " should read writer " +
writerFieldSchema);
+ }
+
+ /**
+ * Negative counterpart of {@link #assertCompatible(HoodieSchema,
HoodieSchema)}: asserts that a field
+ * written as {@code writerFieldSchema} cannot be read back as {@code
readerFieldSchema}, that is, the
+ * {@code writer -> reader} conversion is rejected. Same reader-first
argument order and same single-field
+ * record wrapping.
+ *
+ * @param readerFieldSchema type the field would be read as (the "to" side
of the conversion)
+ * @param writerFieldSchema type the field was written with (the "from" side)
+ */
+ private static void assertIncompatible(HoodieSchema readerFieldSchema,
HoodieSchema writerFieldSchema) {
+ assertFalse(HoodieSchemaCompatibility.isSchemaCompatible(
+ HoodieSchemaTestUtils.createRecord("R", HoodieSchemaField.of("f",
writerFieldSchema, null, null)),
+ HoodieSchemaTestUtils.createRecord("R", HoodieSchemaField.of("f",
readerFieldSchema, null, null)), true, true),
+ "reader " + readerFieldSchema + " should not read writer " +
writerFieldSchema);
+ }
+
@Test
public void testIsSchemaCompatibleWithNestedSchemas() {
// Test with nested record schemas
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaTypePromotion.java
b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaTypePromotion.java
index e6215b9a19aa..9286f1c03b3f 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaTypePromotion.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaTypePromotion.java
@@ -86,6 +86,15 @@ public class TestHoodieSchemaTypePromotion {
assertFalse(HoodieSchemaTypePromotion.canPromote(HoodieSchemaType.BOOLEAN,
HoodieSchemaType.INT));
assertFalse(HoodieSchemaTypePromotion.canPromote(HoodieSchemaType.INT,
HoodieSchemaType.BOOLEAN));
assertFalse(HoodieSchemaTypePromotion.canPromote(HoodieSchemaType.LONG,
HoodieSchemaType.STRING));
+ // Logical-type-over-primitive pairs are reader/writer compatibility rules
only (see
+ // HoodieSchemaCompatibilityChecker); they must never be reported as
compatible projections.
+
assertFalse(HoodieSchemaTypePromotion.canPromote(HoodieSchemaType.TIMESTAMP,
HoodieSchemaType.LONG));
+ assertFalse(HoodieSchemaTypePromotion.canPromote(HoodieSchemaType.UUID,
HoodieSchemaType.STRING));
+ assertFalse(HoodieSchemaTypePromotion.canPromote(HoodieSchemaType.DATE,
HoodieSchemaType.INT));
+ // The same holds on the writer side: an int-backed logical type is never
a plain INT to the widening
+ // table, which is what keeps the checker's LONG/FLOAT/DOUBLE cases from
accepting it.
+ assertFalse(HoodieSchemaTypePromotion.canPromote(HoodieSchemaType.LONG,
HoodieSchemaType.DATE));
+ assertFalse(HoodieSchemaTypePromotion.canPromote(HoodieSchemaType.LONG,
HoodieSchemaType.TIME));
}
@Test
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaUtils.java
b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaUtils.java
index 1862f2c80a33..99d1df06055f 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaUtils.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaUtils.java
@@ -24,6 +24,7 @@ import
org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieNullSchemaTypeException;
import org.apache.avro.generic.GenericRecord;
import org.junit.jupiter.api.Test;
@@ -50,6 +51,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -2084,6 +2086,178 @@ public class TestHoodieSchemaUtils {
assertEquals(HoodieSchemaType.LONG,
result.get().getRight().schema().getType());
}
+ /**
+ * Record with a namespace, a record-level doc, a custom record prop,
per-field docs and a nested
+ * record - all of its top-level fields required.
+ */
+ private static HoodieSchema allRequiredPersonSchema() {
+ HoodieSchema address = HoodieSchema.createRecord(
+ "Address",
+ "ns.test",
+ "the address record",
+ Arrays.asList(
+ HoodieSchemaField.of("city",
HoodieSchema.create(HoodieSchemaType.STRING), "city doc", null),
+ HoodieSchemaField.of("zip",
HoodieSchema.create(HoodieSchemaType.INT), null, null)));
+ HoodieSchema schema = HoodieSchema.createRecord(
+ "Person",
+ "ns.test",
+ "the person record",
+ Arrays.asList(
+ HoodieSchemaField.of("id",
HoodieSchema.create(HoodieSchemaType.INT), "id doc", null),
+ HoodieSchemaField.of("name",
HoodieSchema.create(HoodieSchemaType.STRING), null, null),
+ HoodieSchemaField.of("address", address, "address doc", null)));
+ schema.addProp("hoodie.custom.prop", "custom-value");
+ return schema;
+ }
+
+ @Test
+ public void testAsNullableMakesEveryTopLevelFieldNullable() {
+ HoodieSchema schema = allRequiredPersonSchema();
+
+ HoodieSchema nullable = HoodieSchemaUtils.asNullable(schema);
+
+ assertNotSame(schema, nullable);
+ assertEquals("Person", nullable.getName());
+ assertEquals("ns.test", nullable.getNamespace().get());
+ assertEquals("ns.test.Person", nullable.getFullName());
+ assertEquals(Arrays.asList("id", "name", "address"),
+
nullable.getFields().stream().map(HoodieSchemaField::name).collect(Collectors.toList()));
+
+ for (HoodieSchemaField field : nullable.getFields()) {
+ assertTrue(field.isNullable(), "Field " + field.name() + " should be
nullable");
+ assertEquals(HoodieSchema.NULL_VALUE, field.defaultVal().get());
+ }
+ assertEquals(HoodieSchemaType.INT,
nullable.getField("id").get().getNonNullSchema().getType());
+ assertEquals(HoodieSchemaType.STRING,
nullable.getField("name").get().getNonNullSchema().getType());
+
+ // Per-field docs survive the InternalSchema round trip.
+ assertEquals("id doc", nullable.getField("id").get().doc().get());
+ assertFalse(nullable.getField("name").get().doc().isPresent());
+ assertEquals("address doc",
nullable.getField("address").get().doc().get());
+
+ // Only the top level changes: the nested record keeps its own required
fields.
+ HoodieSchema nestedAddress =
nullable.getField("address").get().getNonNullSchema();
+ assertEquals(HoodieSchemaType.RECORD, nestedAddress.getType());
+ assertEquals("ns.test.Address", nestedAddress.getFullName());
+ assertFalse(nestedAddress.getField("city").get().isNullable());
+ assertFalse(nestedAddress.getField("zip").get().isNullable());
+ assertEquals("city doc", nestedAddress.getField("city").get().doc().get());
+
+ // The InternalSchema carries neither a record doc nor record props, so
both are dropped. This is
+ // the behaviour the Avro-typed implementation had as well, since it ran
the same conversion.
+ assertFalse(nullable.getDoc().isPresent());
+ assertTrue(nullable.getObjectProps().isEmpty());
+
+ // The input schema is left untouched.
+ assertEquals("the person record", schema.getDoc().get());
+ assertEquals("custom-value",
schema.getObjectProps().get("hoodie.custom.prop"));
+ assertFalse(schema.getField("id").get().isNullable());
+ }
+
+ @Test
+ public void testAsNullableReturnsSameInstanceWhenAllFieldsAlreadyNullable() {
+ HoodieSchema schema = HoodieSchema.createRecord(
+ "AllNullable",
+ "ns.test",
+ null,
+ Arrays.asList(
+ HoodieSchemaField.of("id",
HoodieSchema.createNullable(HoodieSchemaType.INT), null,
HoodieSchema.NULL_VALUE),
+ HoodieSchemaField.of("name",
HoodieSchema.createNullable(HoodieSchemaType.STRING), "name doc",
HoodieSchema.NULL_VALUE)));
+
+ assertSame(schema, HoodieSchemaUtils.asNullable(schema));
+ }
+
+ @Test
+ public void testAsNullableLeavesAlreadyNullableFieldsUntouched() {
+ HoodieSchema nullableName =
HoodieSchema.createNullable(HoodieSchemaType.STRING);
+ HoodieSchema schema = HoodieSchema.createRecord(
+ "Mixed",
+ "ns.test",
+ null,
+ Arrays.asList(
+ HoodieSchemaField.of("optional_name", nullableName, "name doc",
HoodieSchema.NULL_VALUE),
+ HoodieSchemaField.of("required_id",
HoodieSchema.create(HoodieSchemaType.LONG), null, null)));
+
+ HoodieSchema nullable = HoodieSchemaUtils.asNullable(schema);
+
+ assertEquals(nullableName,
nullable.getField("optional_name").get().schema());
+ assertEquals("name doc",
nullable.getField("optional_name").get().doc().get());
+
+ HoodieSchemaField requiredId = nullable.getField("required_id").get();
+ assertTrue(requiredId.isNullable());
+ assertEquals(HoodieSchemaType.LONG,
requiredId.getNonNullSchema().getType());
+ }
+
+ @Test
+ public void testAsNullableTreatsBareNullFieldAsAlreadyNullable() {
+ // Avro's Schema#isNullable is true for a bare NULL type while
HoodieSchema#isNullable is not, so a
+ // record made only of NULL-typed fields must still short-circuit rather
than attempt a conversion.
+ HoodieSchema schema = HoodieSchema.createRecord(
+ "OnlyNull",
+ "ns.test",
+ null,
+ Collections.singletonList(
+ HoodieSchemaField.of("nothing",
HoodieSchema.create(HoodieSchemaType.NULL), null, null)));
+
+ assertSame(schema, HoodieSchemaUtils.asNullable(schema));
+ }
+
+ @Test
+ public void testAsNullableRejectsBareNullFieldAlongsideARequiredField() {
+ // A NULL-typed field is never added to the update list, but as soon as
some other field does need
+ // updating the InternalSchema conversion runs and rejects the NULL type
outright. Pinned because it
+ // is what the previous Avro-typed implementation did too.
+ HoodieSchema schema = HoodieSchema.createRecord(
+ "NullAndRequired",
+ "ns.test",
+ null,
+ Arrays.asList(
+ HoodieSchemaField.of("nothing",
HoodieSchema.create(HoodieSchemaType.NULL), null, null),
+ HoodieSchemaField.of("id",
HoodieSchema.create(HoodieSchemaType.INT), null, null)));
+
+ HoodieNullSchemaTypeException exception =
assertThrows(HoodieNullSchemaTypeException.class,
+ () -> HoodieSchemaUtils.asNullable(schema));
+ assertTrue(exception.getMessage().contains("nothing"),
exception.getMessage());
+ }
+
+ @Test
+ public void testAsNullablePinsTheInternalSchemaRoundTripLosses() {
+ // All three losses are what the previous Avro-typed implementation
produced as well: the
+ // InternalSchema has no field defaults, no ENUM type and no union branch
order of its own.
+ HoodieSchema schema = HoodieSchema.createRecord(
+ "Lossy",
+ "ns.test",
+ null,
+ Arrays.asList(
+ HoodieSchemaField.of("count",
HoodieSchema.create(HoodieSchemaType.INT), null, 0),
+ HoodieSchemaField.of("kind", HoodieSchema.createEnum("Kind",
"ns.test", null, Arrays.asList("A", "B")), null, null),
+ HoodieSchemaField.of("null_last",
+
HoodieSchema.createUnion(HoodieSchema.create(HoodieSchemaType.STRING),
HoodieSchema.create(HoodieSchemaType.NULL)), null, null),
+ HoodieSchemaField.of("embedding", HoodieSchema.createVector(3),
null, null)));
+
+ HoodieSchema nullable = HoodieSchemaUtils.asNullable(schema);
+
+ // A non-null default is replaced by the null default of the new union.
+ assertEquals(HoodieSchema.NULL_VALUE,
nullable.getField("count").get().defaultVal().get());
+ // ENUM is lowered to STRING.
+ assertEquals(HoodieSchemaType.STRING,
nullable.getField("kind").get().getNonNullSchema().getType());
+ // An already-nullable null-last union comes back null-first.
+ assertEquals(Arrays.asList(HoodieSchemaType.NULL, HoodieSchemaType.STRING),
+
nullable.getField("null_last").get().schema().getTypes().stream().map(HoodieSchema::getType).collect(Collectors.toList()));
+ // A VECTOR column, the Flink clustering case, survives with its logical
type and dimension.
+ HoodieSchema embedding =
nullable.getField("embedding").get().getNonNullSchema();
+ assertEquals(HoodieSchemaType.VECTOR, embedding.getType());
+ assertEquals(3, ((HoodieSchema.Vector) embedding).getDimension());
+ }
+
+ @Test
+ public void testAsNullableRejectsNonRecordSchema() {
+ assertThrows(IllegalArgumentException.class,
+ () ->
HoodieSchemaUtils.asNullable(HoodieSchema.create(HoodieSchemaType.STRING)));
+ assertThrows(IllegalArgumentException.class,
+ () ->
HoodieSchemaUtils.asNullable(HoodieSchema.createNullable(allRequiredPersonSchema())));
+ }
+
private static HoodieSchema deleteLogTableSchema() {
return HoodieSchema.createRecord(
"TestRecord",
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroWriteSupportShredding.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroWriteSupportShredding.java
index e4ae84422fbc..4c35a5300d6d 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroWriteSupportShredding.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroWriteSupportShredding.java
@@ -65,7 +65,7 @@ import java.util.Map;
import java.util.Properties;
import java.util.stream.Collectors;
-import static
org.apache.hudi.common.avro.AvroSchemaUtils.getNonNullTypeFromUnion;
+import static
org.apache.hudi.common.avro.HoodieAvroUtils.getNonNullTypeFromUnion;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataPayload.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataPayload.java
index e569d9b7f753..103a36d3fd9e 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataPayload.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataPayload.java
@@ -357,11 +357,15 @@ public class TestHoodieMetadataPayload extends
HoodieCommonTestHarness {
"record-key", PARTITION_NAME, "not-a-uuid", "20240101000000000",
0));
}
- @Test
- public void testProjectedInsertValueIncludesBloomFilter() throws IOException
{
- HoodieMetadataPayload bloomFilterPayload =
HoodieMetadataPayload.createBloomFilterMetadataRecord(
+ private static HoodieMetadataPayload newBloomFilterPayload() {
+ return HoodieMetadataPayload.createBloomFilterMetadataRecord(
PARTITION_NAME, "file-id_1-0-1_20240101000000000.parquet",
"20240101000000000", "SIMPLE",
ByteBuffer.wrap("bloom-data".getBytes()), false).getData();
+ }
+
+ @Test
+ public void testProjectedInsertValueIncludesBloomFilter() throws IOException
{
+ HoodieMetadataPayload bloomFilterPayload = newBloomFilterPayload();
Schema projectedSchema = HoodieSchemaUtils.addMetadataFields(
HoodieSchema.fromAvroSchema(HoodieMetadataRecord.getClassSchema())).toAvroSchema();
@@ -373,13 +377,12 @@ public class TestHoodieMetadataPayload extends
HoodieCommonTestHarness {
@Test
public void testInsertValueFastPathOnlyForClassSchema() throws IOException {
- HoodieMetadataPayload payload =
HoodieMetadataPayload.createBloomFilterMetadataRecord(
- PARTITION_NAME, "file-id_1-0-1_20240101000000000.parquet",
"20240101000000000", "SIMPLE",
- ByteBuffer.wrap("bloom-data".getBytes()), false).getData();
+ HoodieMetadataPayload payload = newBloomFilterPayload();
// The class schema singleton (and no schema) takes the reference-equality
fast path and returns the generated record.
IndexedRecord fastPath =
payload.getInsertValue(HoodieMetadataRecord.getClassSchema()).get();
assertInstanceOf(HoodieMetadataRecord.class, fastPath);
+ // Implied by the check above (the generated class returns SCHEMA$ from
both accessors); pins the invariant.
assertSame(HoodieMetadataRecord.getClassSchema(), fastPath.getSchema());
assertInstanceOf(HoodieMetadataRecord.class,
payload.getInsertValue(null).get());
@@ -388,7 +391,9 @@ public class TestHoodieMetadataPayload extends
HoodieCommonTestHarness {
HoodieSchema.fromAvroSchema(HoodieMetadataRecord.getClassSchema())).toAvroSchema();
assertInstanceOf(GenericData.Record.class,
payload.getInsertValue(withMetaFields).get());
- // So an equal but distinct copy of the bare class schema is not usable:
the fast path has to hit by identity.
+ // So an equal but distinct copy of the bare class schema is not usable.
It misses the identity check, and the
+ // slow path then writes at the offsets the metadata fields would occupy,
which a bare copy does not have:
+ // hence ArrayIndexOutOfBoundsException rather than a wrong-but-returned
record.
Schema equalCopy = new
Schema.Parser().parse(HoodieMetadataRecord.getClassSchema().toString());
assertEquals(HoodieMetadataRecord.getClassSchema(), equalCopy);
assertThrows(ArrayIndexOutOfBoundsException.class, () ->
payload.getInsertValue(equalCopy));
@@ -401,9 +406,7 @@ public class TestHoodieMetadataPayload extends
HoodieCommonTestHarness {
assertTrue(filesPayload.toString().contains("creations=[file.parquet]"));
assertTrue(filesPayload.toString().contains("deletions=[old.parquet]"));
- HoodieMetadataPayload bloomFilterPayload =
HoodieMetadataPayload.createBloomFilterMetadataRecord(
- PARTITION_NAME, "file-id_1-0-1_20240101000000000.parquet",
"20240101000000000", "SIMPLE",
- ByteBuffer.wrap("bloom-data".getBytes()), false).getData();
+ HoodieMetadataPayload bloomFilterPayload = newBloomFilterPayload();
assertTrue(bloomFilterPayload.toString().contains("BloomFilter"));
HoodieColumnRangeMetadata<Comparable> columnRange =
HoodieColumnRangeMetadata.<Comparable>create(
