This is an automated email from the ASF dual-hosted git repository.
prashantwason 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 c167f7b39244 fix(hive-sync): skip null type fields when syncing the
table schema (#20354)
c167f7b39244 is described below
commit c167f7b392441cfb86b1dd64bb6c2708878e80d6
Author: Prashant Wason <[email protected]>
AuthorDate: Thu Oct 15 19:20:02 2026 -0700
fix(hive-sync): skip null type fields when syncing the table schema (#20354)
* fix(hive-sync): skip null type fields when syncing the table schema
A field declared with the Avro primitive type "null" (not a nullable
union) is valid Avro but always holds null. The parquet writer does not
write such fields, and Hive has no equivalent type. Since meta sync moved
from the parquet MessageType to HoodieSchema, HiveSchemaUtil and
SparkSchemaUtils throw "Cannot convert type NULL" for these fields, so
Hive sync fails for any table whose schema contains one, at the top level
or nested inside a record.
Skip "null" type fields in both converters so the synced schema matches
the columns in the data files, as it did before the HoodieSchema migration.
* Address review comments: inline the null type check and wrap the stream
chain
* Add an end-to-end Hive sync test for "null" type fields
* Use lowercase field names in the null type field sync test
---
.../org/apache/hudi/hive/util/HiveSchemaUtil.java | 8 ++++
.../org/apache/hudi/hive/TestHiveSyncTool.java | 30 +++++++++++++
.../org/apache/hudi/hive/TestSparkSchemaUtils.java | 21 +++++++++
.../apache/hudi/hive/util/TestHiveSchemaUtil.java | 51 ++++++++++++++++++++++
.../hudi/sync/common/util/SparkSchemaUtils.java | 14 +++---
5 files changed, 119 insertions(+), 5 deletions(-)
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveSchemaUtil.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveSchemaUtil.java
index 065d3515614a..bbc9ef94546f 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveSchemaUtil.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/HiveSchemaUtil.java
@@ -174,6 +174,11 @@ public class HiveSchemaUtil {
LinkedHashMap<String, String> hiveSchema = new LinkedHashMap<>();
List<HoodieSchemaField> fields = schema.getFields();
for (HoodieSchemaField field : fields) {
+ // Fields of the "null" type always hold null and have no Hive
equivalent. They are skipped, the same
+ // way the parquet writer skips them, so the Hive schema matches the
columns present in the data files.
+ if (field.schema().isSchemaNull()) {
+ continue;
+ }
hiveSchema.put(field.name(), convertField(field.schema(),
supportTimestamp, doFormat));
}
return hiveSchema;
@@ -269,6 +274,9 @@ public class HiveSchemaUtil {
StringBuilder struct = new StringBuilder();
struct.append(doFormat ? "STRUCT< " : "STRUCT<");
for (HoodieSchemaField field : fields) {
+ if (field.schema().isSchemaNull()) {
+ continue;
+ }
// TODO: struct field name is only translated to support special char($)
// We will need to extend it to other collection type
struct.append(hiveCompatibleFieldName(field.name(), true,
doFormat)).append(doFormat ? " : " : ":");
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
index 7f758b79dbfa..86fba5a36969 100644
---
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
@@ -1141,6 +1141,36 @@ public class TestHiveSyncTool {
"The last commit that was synced should be updated in the
TBLPROPERTIES");
}
+ @ParameterizedTest
+ @MethodSource("syncMode")
+ public void testSyncWithNullTypeFields(String syncMode) throws Exception {
+ hiveSyncProps.setProperty(HIVE_SYNC_MODE.key(), syncMode);
+ String commitTime = "100";
+ HiveTestUtil.createCOWTableWithSchema(commitTime,
HoodieSchema.parse("{\"type\": \"record\", \"name\": \"User\", "
+ + "\"namespace\": \"example.avro\", \"fields\": ["
+ + "{\"name\": \"name\", \"type\": \"string\"},"
+ + "{\"name\": \"clientpromotion\", \"type\": \"null\", \"default\":
null},"
+ + "{\"name\": \"address\", \"type\": {\"type\": \"record\", \"name\":
\"Address\", \"fields\": ["
+ + "{\"name\": \"city\", \"type\": \"string\"},"
+ + "{\"name\": \"nestedpromotion\", \"type\": \"null\", \"default\":
null}]}}]}"));
+
+ reInitHiveSyncClient();
+ reSyncHiveTable();
+ assertEquals(commitTime,
hiveClient.getLastCommitTimeSynced(HiveTestUtil.TABLE_NAME).get(),
+ "The last commit that was synced should be updated in the
TBLPROPERTIES");
+ Map<String, String> hiveSchema =
hiveClient.getMetastoreSchema(HiveTestUtil.TABLE_NAME);
+ assertEquals("string", hiveSchema.get("name").toLowerCase(Locale.ROOT));
+ assertFalse(hiveSchema.containsKey("clientpromotion"), "null type field
should not be synced");
+ String addressType = hiveSchema.get("address").toLowerCase(Locale.ROOT);
+ assertTrue(addressType.contains("city"), "nested field should be synced");
+ assertFalse(addressType.contains("nestedpromotion"), "nested null type
field should not be synced");
+ String sparkSchema = IMetaStoreClientUtil.getMSC(getHiveConf())
+ .getTable(HiveTestUtil.DB_NAME,
HiveTestUtil.TABLE_NAME).getParameters().get("spark.sql.sources.schema.part.0");
+ assertTrue(sparkSchema.contains("\"city\""), "spark schema should contain
the nested field");
+ assertFalse(sparkSchema.contains("clientpromotion"), "spark schema should
not contain the null type field");
+ assertFalse(sparkSchema.contains("nestedpromotion"), "spark schema should
not contain the nested null type field");
+ }
+
@ParameterizedTest
@MethodSource("syncModeAndEnablePushDown")
public void testSyncIncremental(String syncMode, String enablePushDown)
throws Exception {
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestSparkSchemaUtils.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestSparkSchemaUtils.java
index 9ce19238f093..d5c1b4fa4813 100644
---
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestSparkSchemaUtils.java
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestSparkSchemaUtils.java
@@ -258,4 +258,25 @@ public class TestSparkSchemaUtils {
StructType variantStruct = (StructType) sparkSchema.fields()[1].dataType();
assertEquals(2, variantStruct.fields().length);
}
+
+ @Test
+ public void testConvertSchemaSkipsNullTypeFields() {
+ HoodieSchema nested = HoodieSchema.createRecord("nested", null, null,
false, Arrays.asList(
+ HoodieSchemaField.of("name",
HoodieSchema.createNullable(HoodieSchemaType.STRING), null, null),
+ HoodieSchemaField.of("nested_null",
HoodieSchema.create(HoodieSchemaType.NULL), null, null)
+ ));
+ HoodieSchema schema = HoodieSchema.createRecord("root", null, null, false,
Arrays.asList(
+ HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.INT),
null, null),
+ HoodieSchemaField.of("top_level_null",
HoodieSchema.create(HoodieSchemaType.NULL), null, null),
+ HoodieSchemaField.of("nested", HoodieSchema.createNullable(nested),
null, null)
+ ));
+
+ StructType sparkSchema = (StructType)
StructType.fromJson(SparkSchemaUtils.convertToSparkSchemaJson(schema));
+ assertEquals(2, sparkSchema.fields().length);
+ assertEquals("id", sparkSchema.fields()[0].name());
+ assertEquals("nested", sparkSchema.fields()[1].name());
+ StructType nestedStruct = (StructType) sparkSchema.fields()[1].dataType();
+ assertEquals(1, nestedStruct.fields().length);
+ assertEquals("name", nestedStruct.fields()[0].name());
+ }
}
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveSchemaUtil.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveSchemaUtil.java
index 4c1c7507f3cb..f15b3c12345b 100644
---
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveSchemaUtil.java
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/util/TestHiveSchemaUtil.java
@@ -198,4 +198,55 @@ public class TestHiveSchemaUtil {
"STRUCT< `title` : string, `variant_data` : STRUCT< `metadata` :
binary, `value` : binary>>");
assertEquals(expected, actual);
}
+
+ /**
+ * Fields of the Avro primitive type "null" (not a nullable union) are valid
Avro but carry no data.
+ * They are not written to parquet and have no Hive equivalent, so Hive sync
must skip them,
+ * at the top level as well as inside nested records.
+ */
+ @Test
+ void testNullTypeFieldsAreSkipped() throws IOException {
+ Map<String, String> actual =
HiveSchemaUtil.convertSchemaToHiveSchema(schemaWithNullTypeFields(), false);
+ Map<String, String> expected = new HashMap<>();
+ expected.put("`id`", "string");
+ expected.put("`msg`",
+ "STRUCT< `fare` : STRUCT< `breakdown` : STRUCT< `profile` : string,
`currency` : string, `total` : string>>>");
+ assertEquals(expected, actual);
+ }
+
+ /**
+ * A table whose Hive schema was synced before (without the "null" type
fields) must not
+ * fail or report a schema difference.
+ */
+ @Test
+ void testSchemaDiffIgnoresNullTypeFields() {
+ Map<String, String> tableSchema = new HashMap<>();
+ tableSchema.put("id", "string");
+ tableSchema.put("msg",
"struct<fare:struct<breakdown:struct<profile:string,currency:string,total:string>>>");
+ SchemaDifference schemaDifference =
HiveSchemaUtil.getSchemaDifference(schemaWithNullTypeFields(),
+ tableSchema, Collections.emptyList(), false);
+ assertTrue(schemaDifference.isEmpty(), "Unexpected schema difference: " +
schemaDifference);
+ }
+
+ /**
+ * Mirrors a Kafka topic schema where some fields are declared with the
primitive type "null",
+ * nested three records deep.
+ */
+ private static HoodieSchema schemaWithNullTypeFields() {
+ return
HoodieSchema.parse("{\"type\":\"record\",\"name\":\"h\",\"fields\":["
+ + "{\"name\":\"id\",\"type\":[\"null\",\"string\"],\"default\":null},"
+ + "{\"name\":\"top_level_null\",\"type\":\"null\",\"default\":null},"
+ +
"{\"name\":\"msg\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"msg\",\"fields\":["
+ +
"{\"name\":\"fare\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"fare\",\"fields\":["
+ +
"{\"name\":\"breakdown\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"breakdown\",\"fields\":["
+ +
"{\"name\":\"profile\",\"type\":[\"null\",\"string\"],\"default\":null},"
+ + "{\"name\":\"clientPromotion\",\"type\":\"null\",\"default\":null},"
+ +
"{\"name\":\"currency\",\"type\":[\"null\",\"string\"],\"default\":null},"
+ +
"{\"name\":\"personalizedDiscount\",\"type\":\"null\",\"default\":null},"
+ +
"{\"name\":\"total\",\"type\":[\"null\",\"string\"],\"default\":null}"
+ + "]}],\"default\":null}"
+ + "]}],\"default\":null}"
+ + "]}],\"default\":null}"
+ + "]}");
+ }
}
diff --git
a/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/util/SparkSchemaUtils.java
b/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/util/SparkSchemaUtils.java
index 044681123ad1..7aa6590a1615 100644
---
a/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/util/SparkSchemaUtils.java
+++
b/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/util/SparkSchemaUtils.java
@@ -41,11 +41,15 @@ public class SparkSchemaUtils {
* as {@code comment}, which spark displays as the
column comment
*/
public static String convertToSparkSchemaJson(HoodieSchema schema, boolean
includeFieldDocs) {
- String fieldsJsonString = schema.getFields().stream().map(field -> {
- String metadata = convertFieldMetadata(field, includeFieldDocs);
- return "{\"name\":\"" + field.name() + "\",\"type\":" +
convertFieldType(field.getNonNullSchema(), includeFieldDocs)
- + ",\"nullable\":" + field.isNullable() + ",\"metadata\":" +
metadata + "}";
- }).reduce((a, b) -> a + "," + b).orElse("");
+ // Fields of the "null" type always hold null and are not written to the
data files, so they are skipped
+ String fieldsJsonString = schema.getFields().stream()
+ .filter(field -> !field.schema().isSchemaNull())
+ .map(field -> {
+ String metadata = convertFieldMetadata(field, includeFieldDocs);
+ return "{\"name\":\"" + field.name() + "\",\"type\":" +
convertFieldType(field.getNonNullSchema(), includeFieldDocs)
+ + ",\"nullable\":" + field.isNullable() + ",\"metadata\":" +
metadata + "}";
+ })
+ .reduce((a, b) -> a + "," + b).orElse("");
return "{\"type\":\"struct\",\"fields\":[" + fieldsJsonString + "]}";
}
