This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 9b701ca1e27e649ada053352713c5eff6934446f Author: Krishen <[email protected]> AuthorDate: Tue Jun 16 19:29:07 2026 -0700 feat(flink): Support writing out-of-line BLOB columns (#18958) * feat(flink): Support writing out-of-line BLOB columns --------- Co-authored-by: Krishen Bhan <“[email protected]”> Co-authored-by: Cursor <[email protected]> (cherry picked from commit 3ada03d744fd3f0f8d2c5b88c2589ebcdac0b3ce) --- .../apache/hudi/util/HoodieSchemaConverter.java | 44 ++-- .../apache/hudi/util/RowDataToAvroConverters.java | 8 + .../hudi/util/TestHoodieSchemaConverter.java | 17 ++ .../org/apache/hudi/table/ITTestBlobWrite.java | 226 +++++++++++++++++++++ .../hudi/utils/TestRowDataToAvroConverters.java | 120 +++++++++++ 5 files changed, 394 insertions(+), 21 deletions(-) diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/HoodieSchemaConverter.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/HoodieSchemaConverter.java index c413101dde46..c859675488ad 100644 --- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/HoodieSchemaConverter.java +++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/HoodieSchemaConverter.java @@ -272,6 +272,16 @@ public class HoodieSchemaConverter { /** * Detects if a Flink RowType represents a BLOB structure by validating it matches the schema defined in {@link HoodieSchema.Blob}. + * + * <p>Detection intentionally keys off the stable structural signals (field names and base type + * roots) and does <b>not</b> assert nested-field nullability. Flink SQL {@code CREATE TABLE} does + * not reliably preserve {@code NOT NULL} constraints on nested {@code ROW} fields, so requiring an + * exact nullability match would silently demote a user's BLOB column to a generic record when the + * column is declared through DDL. The canonical nullability is restored by {@link HoodieSchema#createBlob()}. + * + * <p>TODO: This heuristic is a workaround for the lack of a native Flink/Parquet BLOB logical + * type. See <a href="https://github.com/apache/hudi/issues/18711";>apache/hudi#18711</a> for + * the tracked work to remove this structural inference. */ private static boolean isBlobStructure(RowType rowType) { // Validate: 3 fields with exact names @@ -286,24 +296,20 @@ public class HoodieSchemaConverter { return false; } - // Validate 'type' field: non-null STRING - LogicalType typeField = rowType.getTypeAt(0); - if (!isFamily(typeField, LogicalTypeFamily.CHARACTER_STRING) || typeField.isNullable()) { + // Validate 'type' field: STRING + if (!isFamily(rowType.getTypeAt(0), LogicalTypeFamily.CHARACTER_STRING)) { return false; } - // Validate 'data' field: nullable BYTES/VARBINARY + // Validate 'data' field: BYTES/VARBINARY LogicalType dataField = rowType.getTypeAt(1); if (dataField.getTypeRoot() != LogicalTypeRoot.BINARY && dataField.getTypeRoot() != LogicalTypeRoot.VARBINARY) { return false; } - if (!dataField.isNullable()) { - return false; - } - // Validate 'reference' field: nullable ROW + // Validate 'reference' field: ROW LogicalType referenceField = rowType.getTypeAt(2); - if (!referenceField.isNullable() || referenceField.getTypeRoot() != LogicalTypeRoot.ROW) { + if (referenceField.getTypeRoot() != LogicalTypeRoot.ROW) { return false; } @@ -322,24 +328,20 @@ public class HoodieSchemaConverter { } // Validate reference sub-field types - // external_path: non-null STRING - if (!isFamily(referenceRow.getTypeAt(0), LogicalTypeFamily.CHARACTER_STRING) - || referenceRow.getTypeAt(0).isNullable()) { + // external_path: STRING + if (!isFamily(referenceRow.getTypeAt(0), LogicalTypeFamily.CHARACTER_STRING)) { return false; } - // offset: nullable BIGINT - if (referenceRow.getTypeAt(1).getTypeRoot() != LogicalTypeRoot.BIGINT - || !referenceRow.getTypeAt(1).isNullable()) { + // offset: BIGINT + if (referenceRow.getTypeAt(1).getTypeRoot() != LogicalTypeRoot.BIGINT) { return false; } - // length: nullable BIGINT - if (referenceRow.getTypeAt(2).getTypeRoot() != LogicalTypeRoot.BIGINT - || !referenceRow.getTypeAt(2).isNullable()) { + // length: BIGINT + if (referenceRow.getTypeAt(2).getTypeRoot() != LogicalTypeRoot.BIGINT) { return false; } - // managed: non-null BOOLEAN - if (referenceRow.getTypeAt(3).getTypeRoot() != LogicalTypeRoot.BOOLEAN - || referenceRow.getTypeAt(3).isNullable()) { + // managed: BOOLEAN + if (referenceRow.getTypeAt(3).getTypeRoot() != LogicalTypeRoot.BOOLEAN) { return false; } diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/RowDataToAvroConverters.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/RowDataToAvroConverters.java index 6bf94b527074..050ad483de10 100644 --- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/RowDataToAvroConverters.java +++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/RowDataToAvroConverters.java @@ -148,6 +148,14 @@ public class RowDataToAvroConverters { @Override public Object convert(HoodieSchema schema, Object object) { + // The BLOB `type` discriminator is a STRING in Flink but an ENUM in Avro. + // Detect that at call time from the HoodieSchema so the converter stays + // reusable across any row shape — not hard-wired by Flink row structure alone. + HoodieSchema nonNullSchema = schema.getNonNullType(); + if (nonNullSchema.getType() == HoodieSchemaType.ENUM) { + return new GenericData.EnumSymbol( + nonNullSchema.toAvroSchema(), object.toString()); + } return new Utf8(((BinaryStringData) object).toBytes()); } }; diff --git a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestHoodieSchemaConverter.java b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestHoodieSchemaConverter.java index 9f6ad8f53bf7..b14aad4877de 100644 --- a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestHoodieSchemaConverter.java +++ b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestHoodieSchemaConverter.java @@ -771,6 +771,23 @@ public class TestHoodieSchemaConverter { HoodieSchema convertedSchema = HoodieSchemaConverter.convertToSchema(blobLikeRowType); assertEquals(HoodieSchemaType.BLOB, convertedSchema.getType()); + // Positive case: same structure but every nested field nullable. Flink SQL CREATE TABLE does not + // preserve NOT NULL on nested ROW fields, so detection must not depend on nested nullability. + DataType allNullableBlobRow = DataTypes.ROW( + DataTypes.FIELD(HoodieSchema.Blob.TYPE, DataTypes.STRING().nullable()), + DataTypes.FIELD(HoodieSchema.Blob.INLINE_DATA_FIELD, DataTypes.BYTES().nullable()), + DataTypes.FIELD(HoodieSchema.Blob.EXTERNAL_REFERENCE, DataTypes.ROW( + DataTypes.FIELD(HoodieSchema.Blob.EXTERNAL_REFERENCE_PATH, DataTypes.STRING().nullable()), + DataTypes.FIELD(HoodieSchema.Blob.EXTERNAL_REFERENCE_OFFSET, DataTypes.BIGINT().nullable()), + DataTypes.FIELD(HoodieSchema.Blob.EXTERNAL_REFERENCE_LENGTH, DataTypes.BIGINT().nullable()), + DataTypes.FIELD(HoodieSchema.Blob.EXTERNAL_REFERENCE_IS_MANAGED, DataTypes.BOOLEAN().nullable()) + ).nullable()) + ).notNull(); + + RowType allNullableBlobRowType = (RowType) allNullableBlobRow.getLogicalType(); + HoodieSchema allNullableConverted = HoodieSchemaConverter.convertToSchema(allNullableBlobRowType); + assertEquals(HoodieSchemaType.BLOB, allNullableConverted.getType()); + // Negative case 1: Different field names DataType differentNames = DataTypes.ROW( DataTypes.FIELD("wrong_name", DataTypes.STRING().notNull()), diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestBlobWrite.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestBlobWrite.java new file mode 100644 index 000000000000..a66c9f034765 --- /dev/null +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestBlobWrite.java @@ -0,0 +1,226 @@ +/* + * 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.table; + +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.schema.HoodieSchemaField; +import org.apache.hudi.common.schema.HoodieSchemaUtils; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.TableSchemaResolver; +import org.apache.hudi.util.HoodieSchemaConverter; +import org.apache.hudi.util.StreamerUtil; +import org.apache.hudi.utils.FlinkMiniCluster; + +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.TableResult; +import org.apache.flink.table.api.config.ExecutionConfigOptions; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.Row; +import org.apache.flink.util.CollectionUtil; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +import java.io.File; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * IT case for writing out-of-line (OOL) BLOB columns through the Flink writer and reading them back. + * + * <p>Verifies that the Hudi {@link HoodieSchema.Blob} structure round-trips through the Flink + * write/read pipeline (both COW and MOR), that updates land through the MOR Avro log path, and that + * the stored table schema keeps the BLOB logical type instead of degrading the column to a generic + * Flink/Avro record. + */ +@ExtendWith(FlinkMiniCluster.class) +public class ITTestBlobWrite { + + @TempDir + File tempFile; + + private static final String BLOB_COLUMN = "blob_col"; + + private static final String BLOB_COLUMN_DDL = + " " + BLOB_COLUMN + " ROW<\n" + + " `type` STRING NOT NULL,\n" + + " `data` BYTES,\n" + + " `reference` ROW<\n" + + " external_path STRING NOT NULL,\n" + + " `offset` BIGINT,\n" + + " `length` BIGINT,\n" + + " managed BOOLEAN NOT NULL\n" + + " >\n" + + " >,\n"; + + /** + * Regression for {@code HoodieSchemaConverter#isBlobStructure}: Flink SQL {@code CREATE TABLE} + * may declare nested {@code ROW} fields as {@code NOT NULL}, but + * {@link ResolvedSchema#toPhysicalRowDataType()} does not always preserve those constraints in + * the {@link RowType}. Schema inference must still recognize the BLOB shape (same path as + * {@code HoodieTableFactory#inferAvroSchema}), otherwise {@code blob_col} would degrade to a + * generic RECORD in the committed Hoodie schema. + */ + @Test + public void testFlinkSqlDdlPhysicalRowTypeStillMapsToHoodieBlob() { + TableEnvironment tableEnv = batchEnv(); + final String probeTable = "flink_blob_ddl_probe"; + String createProbeDdl = + "CREATE TABLE " + + probeTable + + " (\n" + + " id BIGINT,\n" + + " name STRING,\n" + + BLOB_COLUMN_DDL + + " ts BIGINT\n" + + ") WITH ('connector'='blackhole')"; + tableEnv.executeSql(createProbeDdl); + + ResolvedSchema resolved = tableEnv.from(probeTable).getResolvedSchema(); + RowType physical = (RowType) resolved.toPhysicalRowDataType().getLogicalType(); + // DDL uses NOT NULL on nested BLOB ROW fields where the canonical Hoodie BLOB shape is + // stricter. Flink's physical RowType from ResolvedSchema#toPhysicalRowDataType() may or may + // not preserve those flags across versions (see Flink table config / release notes linked in + // the PR). We do not assert which fields widen — only that Hudi still recognizes the column as + // a BLOB (regression for HoodieSchemaConverter#isBlobStructure). + + HoodieSchema recordSchema = + HoodieSchemaConverter.convertToSchema( + physical, HoodieSchemaUtils.getRecordQualifiedName(probeTable)); + HoodieSchemaField blobField = + recordSchema + .getField(BLOB_COLUMN) + .orElseThrow(() -> new AssertionError("blob_col missing from converted HoodieSchema")); + assertTrue( + blobField.schema().isBlobField(), + "Physical RowType from Flink SQL DDL must still map to Hoodie BLOB, got: " + + blobField.schema()); + + tableEnv.executeSql("DROP TABLE " + probeTable); + } + + private void createTable(TableEnvironment tableEnv, String tablePath, HoodieTableType tableType) { + String createTableDdl = String.format( + "CREATE TABLE blob_table (\n" + + " id BIGINT,\n" + + " name STRING,\n" + + BLOB_COLUMN_DDL + + " ts BIGINT,\n" + + " PRIMARY KEY (id) NOT ENFORCED\n" + + ") WITH (\n" + + " 'connector' = 'hudi',\n" + + " 'path' = '%s',\n" + + " 'table.type' = '%s',\n" + + " 'ordering.fields' = 'ts'\n" + + ");", + tablePath, tableType.name()); + tableEnv.executeSql(createTableDdl); + } + + @ParameterizedTest + @EnumSource(value = HoodieTableType.class) + public void testWriteAndReadOutOfLineBlob(HoodieTableType tableType) throws Exception { + TableEnvironment tableEnv = batchEnv(); + String tablePath = new File(tempFile, "blob_table").getAbsolutePath(); + createTable(tableEnv, tablePath, tableType); + + // First batch: insert two OOL blob references. + execInsert(tableEnv, + "INSERT INTO blob_table VALUES\n" + + "(1, 'doc-1', ROW('OUT_OF_LINE', CAST(NULL AS BYTES), " + + "ROW('file1.bin', CAST(0 AS BIGINT), CAST(100 AS BIGINT), false)), 1000),\n" + + "(2, 'doc-2', ROW('OUT_OF_LINE', CAST(NULL AS BYTES), " + + "ROW('file1.bin', CAST(100 AS BIGINT), CAST(200 AS BIGINT), false)), 2000)"); + + List<Row> rows = readOrdered(tableEnv); + assertEquals(2, rows.size()); + assertOutOfLineRow(rows.get(0), 1L, "doc-1", "file1.bin", 0L, 100L, 1000L); + assertOutOfLineRow(rows.get(1), 2L, "doc-2", "file1.bin", 100L, 200L, 2000L); + + // Second batch: upsert the same keys with new references. For MOR this exercises the Avro + // log write path (RowData -> Avro), including the BLOB enum `type` field. + execInsert(tableEnv, + "INSERT INTO blob_table VALUES\n" + + "(1, 'doc-1', ROW('OUT_OF_LINE', CAST(NULL AS BYTES), " + + "ROW('file2.bin', CAST(500 AS BIGINT), CAST(300 AS BIGINT), false)), 3000),\n" + + "(2, 'doc-2', ROW('OUT_OF_LINE', CAST(NULL AS BYTES), " + + "ROW('file2.bin', CAST(800 AS BIGINT), CAST(400 AS BIGINT), false)), 4000)"); + + List<Row> updated = readOrdered(tableEnv); + assertEquals(2, updated.size()); + assertOutOfLineRow(updated.get(0), 1L, "doc-1", "file2.bin", 500L, 300L, 3000L); + assertOutOfLineRow(updated.get(1), 2L, "doc-2", "file2.bin", 800L, 400L, 4000L); + + // The stored table schema must keep the BLOB logical type, not a generic record. + assertBlobTypePreserved(tablePath); + } + + private static List<Row> readOrdered(TableEnvironment tableEnv) { + return CollectionUtil.iteratorToList( + tableEnv.executeSql("select id, name, blob_col, ts from blob_table order by id").collect()); + } + + private static void assertOutOfLineRow(Row row, long id, String name, String path, + long offset, long length, long ts) { + assertEquals(id, row.getField(0)); + assertEquals(name, row.getField(1)); + Row blob = (Row) row.getField(2); + assertNotNull(blob, "blob struct must be populated"); + assertEquals("OUT_OF_LINE", blob.getField(0)); + assertNull(blob.getField(1), "inline data must be null for OUT_OF_LINE blob"); + Row reference = (Row) blob.getField(2); + assertNotNull(reference, "reference must be populated for OUT_OF_LINE blob"); + assertEquals(path, reference.getField(0)); + assertEquals(offset, reference.getField(1)); + assertEquals(length, reference.getField(2)); + assertEquals(false, reference.getField(3)); + assertEquals(ts, row.getField(3)); + } + + private static void assertBlobTypePreserved(String tablePath) throws Exception { + HoodieTableMetaClient metaClient = + StreamerUtil.createMetaClient(tablePath, new org.apache.hadoop.conf.Configuration()); + HoodieSchema tableSchema = new TableSchemaResolver(metaClient).getTableSchema(); + HoodieSchemaField blobField = tableSchema.getField(BLOB_COLUMN) + .orElseThrow(() -> new AssertionError("blob_col field missing from table schema")); + assertTrue(blobField.schema().isBlobField(), + "blob_col must keep the BLOB logical type, found: " + blobField.schema()); + } + + private static TableEnvironment batchEnv() { + TableEnvironment tableEnv = org.apache.hudi.utils.TestTableEnvs.getBatchTableEnv(); + tableEnv.getConfig().getConfiguration() + .set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 1); + return tableEnv; + } + + private static void execInsert(TableEnvironment tableEnv, String insertSql) throws Exception { + TableResult result = tableEnv.executeSql(insertSql); + result.getJobClient().get().getJobExecutionResult().get(120, TimeUnit.SECONDS); + } +} diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestRowDataToAvroConverters.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestRowDataToAvroConverters.java index 185c5830616c..d7c68b7e318d 100644 --- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestRowDataToAvroConverters.java +++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestRowDataToAvroConverters.java @@ -18,15 +18,22 @@ package org.apache.hudi.utils; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.schema.HoodieSchemaField; +import org.apache.hudi.common.schema.HoodieSchemaType; import org.apache.hudi.util.HoodieSchemaConverter; import org.apache.hudi.util.RowDataToAvroConverters; +import org.apache.avro.generic.GenericData; import org.apache.avro.generic.GenericRecord; +import org.apache.avro.util.Utf8; import org.apache.flink.formats.common.TimestampFormat; import org.apache.flink.formats.json.JsonToRowDataConverters; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.StringData; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.RowType; import org.junit.jupiter.api.Assertions; @@ -36,6 +43,7 @@ import java.time.Instant; import java.time.LocalDateTime; import java.time.ZoneId; import java.time.format.DateTimeFormatter; +import java.util.Arrays; import static org.apache.flink.table.api.DataTypes.FIELD; import static org.apache.flink.table.api.DataTypes.ROW; @@ -81,4 +89,116 @@ class TestRowDataToAvroConverters { Assertions.assertEquals("2021-03-30 08:44:29", formatter.format(LocalDateTime.ofInstant(Instant.ofEpochMilli((Long) avroRecord.get(0)), ZoneId.of("UTC+1")))); Assertions.assertEquals("2021-03-30 15:44:29", formatter.format(LocalDateTime.ofInstant(Instant.ofEpochMilli((Long) avroRecord.get(0)), ZoneId.of("Asia/Shanghai")))); } + + @Test + void testRowDataToAvroBlobTypeFieldWritesEnumSymbol() { + // Flink models the BLOB `type` discriminator as STRING, but its Avro encoding is an ENUM + // (blob_storage_type). The converter must emit a GenericData.EnumSymbol, not a plain Utf8, + // otherwise Avro log-block writes (MOR) fail with "value OUT_OF_LINE (a Utf8) is not a + // blob_storage_type". + DataType blobRow = DataTypes.ROW( + DataTypes.FIELD(HoodieSchema.Blob.TYPE, DataTypes.STRING().notNull()), + DataTypes.FIELD(HoodieSchema.Blob.INLINE_DATA_FIELD, DataTypes.BYTES().nullable()), + DataTypes.FIELD(HoodieSchema.Blob.EXTERNAL_REFERENCE, DataTypes.ROW( + DataTypes.FIELD(HoodieSchema.Blob.EXTERNAL_REFERENCE_PATH, DataTypes.STRING().notNull()), + DataTypes.FIELD(HoodieSchema.Blob.EXTERNAL_REFERENCE_OFFSET, DataTypes.BIGINT().nullable()), + DataTypes.FIELD(HoodieSchema.Blob.EXTERNAL_REFERENCE_LENGTH, DataTypes.BIGINT().nullable()), + DataTypes.FIELD(HoodieSchema.Blob.EXTERNAL_REFERENCE_IS_MANAGED, DataTypes.BOOLEAN().notNull()) + ).nullable())); + RowType rowType = (RowType) DataTypes.ROW(DataTypes.FIELD("blob_col", blobRow)).getLogicalType(); + + GenericRowData reference = new GenericRowData(4); + reference.setField(0, StringData.fromString("file1.bin")); + reference.setField(1, 0L); + reference.setField(2, 100L); + reference.setField(3, false); + + GenericRowData blob = new GenericRowData(3); + blob.setField(0, StringData.fromString(HoodieSchema.Blob.OUT_OF_LINE)); + blob.setField(1, null); + blob.setField(2, reference); + + GenericRowData top = new GenericRowData(1); + top.setField(0, blob); + + RowDataToAvroConverters.RowDataToAvroConverter converter = + RowDataToAvroConverters.createConverter(rowType); + GenericRecord avroRecord = + (GenericRecord) converter.convert(HoodieSchemaConverter.convertToSchema(rowType), top); + + GenericRecord blobRecord = (GenericRecord) avroRecord.get(0); + Object typeValue = blobRecord.get(HoodieSchema.Blob.TYPE); + Assertions.assertInstanceOf(GenericData.EnumSymbol.class, typeValue, + "BLOB `type` must be written as an Avro EnumSymbol, found: " + + (typeValue == null ? "null" : typeValue.getClass().getName())); + Assertions.assertEquals(HoodieSchema.Blob.OUT_OF_LINE, typeValue.toString()); + } + + /** + * A ROW whose field names match the BLOB structure but whose {@link HoodieSchema} carries a + * plain {@code STRING} (not {@code ENUM}) for the {@code type} field must write a plain + * {@link Utf8}, not a {@link GenericData.EnumSymbol}. + */ + @Test + void testBlobShapedRowWithPlainStringSchemaWritesUtf8() { + DataType blobShapedRow = DataTypes.ROW( + DataTypes.FIELD(HoodieSchema.Blob.TYPE, DataTypes.STRING().notNull()), + DataTypes.FIELD(HoodieSchema.Blob.INLINE_DATA_FIELD, DataTypes.BYTES().nullable()), + DataTypes.FIELD(HoodieSchema.Blob.EXTERNAL_REFERENCE, DataTypes.ROW( + DataTypes.FIELD(HoodieSchema.Blob.EXTERNAL_REFERENCE_PATH, DataTypes.STRING().notNull()), + DataTypes.FIELD(HoodieSchema.Blob.EXTERNAL_REFERENCE_OFFSET, DataTypes.BIGINT().nullable()), + DataTypes.FIELD(HoodieSchema.Blob.EXTERNAL_REFERENCE_LENGTH, DataTypes.BIGINT().nullable()), + DataTypes.FIELD(HoodieSchema.Blob.EXTERNAL_REFERENCE_IS_MANAGED, DataTypes.BOOLEAN().notNull()) + ).nullable())); + RowType outerRowType = (RowType) DataTypes.ROW( + DataTypes.FIELD("blob_col", blobShapedRow)).getLogicalType(); + + // Plain RECORD schema: field[0] is STRING (not ENUM) — mimics a non-BLOB record whose + // shape happens to match the BLOB structure. + HoodieSchema refSchema = HoodieSchema.createRecord("reference", null, null, Arrays.asList( + HoodieSchemaField.of(HoodieSchema.Blob.EXTERNAL_REFERENCE_PATH, + HoodieSchema.create(HoodieSchemaType.STRING)), + HoodieSchemaField.of(HoodieSchema.Blob.EXTERNAL_REFERENCE_OFFSET, + HoodieSchema.createNullable(HoodieSchemaType.LONG)), + HoodieSchemaField.of(HoodieSchema.Blob.EXTERNAL_REFERENCE_LENGTH, + HoodieSchema.createNullable(HoodieSchemaType.LONG)), + HoodieSchemaField.of(HoodieSchema.Blob.EXTERNAL_REFERENCE_IS_MANAGED, + HoodieSchema.create(HoodieSchemaType.BOOLEAN)) + )); + HoodieSchema plainBlobShapedSchema = HoodieSchema.createRecord("blob_col", null, null, Arrays.asList( + HoodieSchemaField.of(HoodieSchema.Blob.TYPE, HoodieSchema.create(HoodieSchemaType.STRING)), + HoodieSchemaField.of(HoodieSchema.Blob.INLINE_DATA_FIELD, + HoodieSchema.createNullable(HoodieSchemaType.BYTES)), + HoodieSchemaField.of(HoodieSchema.Blob.EXTERNAL_REFERENCE, + HoodieSchema.createNullable(refSchema)) + )); + HoodieSchema outerSchema = HoodieSchema.createRecord("outer", null, null, Arrays.asList( + HoodieSchemaField.of("blob_col", plainBlobShapedSchema) + )); + + GenericRowData reference = new GenericRowData(4); + reference.setField(0, StringData.fromString("file1.bin")); + reference.setField(1, 0L); + reference.setField(2, 100L); + reference.setField(3, false); + + GenericRowData blobRow = new GenericRowData(3); + blobRow.setField(0, StringData.fromString("OUT_OF_LINE")); + blobRow.setField(1, null); + blobRow.setField(2, reference); + + GenericRowData top = new GenericRowData(1); + top.setField(0, blobRow); + + RowDataToAvroConverters.RowDataToAvroConverter converter = + RowDataToAvroConverters.createConverter(outerRowType); + GenericRecord avroRecord = (GenericRecord) converter.convert(outerSchema, top); + + GenericRecord blobRecord = (GenericRecord) avroRecord.get(0); + Object typeValue = blobRecord.get(HoodieSchema.Blob.TYPE); + Assertions.assertInstanceOf(Utf8.class, typeValue, + "STRING field must write as Utf8 (not EnumSymbol) when HoodieSchema is not ENUM; found: " + + (typeValue == null ? "null" : typeValue.getClass().getName())); + Assertions.assertEquals("OUT_OF_LINE", typeValue.toString()); + } } \ No newline at end of file
