wombatu-kun commented on code in PR #19687:
URL: https://github.com/apache/hudi/pull/19687#discussion_r3863129263
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringOperator.java:
##########
@@ -175,6 +177,13 @@ public void open() throws Exception {
// scan successfully without schema validating exception.
this.readerSchema = HoodieSchemaUtils.asNullable(schema);
+ if (this.sortClusteringEnabled) {
Review Comment:
`Pipelines.cluster` attaches this operator to the ingestion stream and
`processElement` runs `doClustering` inside `NonThrownExecutor`, so today an
unorderable sort column fails only the clustering task, while throwing from
`open` fails the task on every restart and takes ingestion down with it. Was
moving the failure out of `doClustering` intended, or should the check run
there?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala:
##########
@@ -203,4 +212,143 @@ object ParquetSchemaEvolutionUtils {
internalSchemaOpt
}
}
+
+ /**
+ * Fails fast when schema-on-read meets a shredded variant file. The
internal schema models a
+ * variant as a two-field {metadata, value} record (with sentinel negative
field ids, see
+ * InternalSchemaConverter), so the merged request clips the file's
typed_value away and the
+ * typed rows would read back with a null value residual - silent data loss.
Reconstruction
+ * under schema-on-read is tracked by #18285; until then the read must fail
loudly. The check
+ * anchors on the sentinel ids, which no real user field can carry, so plain
user structs of
+ * the same shape are left alone. The walk recurses through structs, arrays
and maps because
+ * the row writer shreds nested variants too (see VariantSchemaUtils).
+ *
+ * Footer columns are resolved by the query-schema name. A column renamed
under schema-on-read
+ * still carries its old name in the file and is not matched here; such
reads are left to
+ * #18285 with reconstruction itself.
+ *
+ * A request in the full-variant projection shape fails fast regardless of
the file's layout:
+ * the merged internal-schema request materializes the variant as {metadata,
value} while the
+ * consumer expects the ordinal-named extraction struct, so the read cannot
be served either
+ * way (pruning treats the rewritten struct as the variant column itself, see
+ * SparkInternalSchemaConverter.isVariantRewriteStruct). Two producers ask
for that shape: a
+ * query rewritten by Spark's PushVariantIntoScan (4.x), and Hudi's own
base-file reads on
+ * 4.1+ (SparkFileFormatInternalRowReaderContext, via
SparkAdapter.buildFullVariantReadSchema)
+ * whenever their reader context carries the table's internal schema -
SparkReaderContextFactory
+ * puts the table path and valid commits on the conf once one is committed,
so inline compaction
+ * and clustering under a schema-on-read write, and CDC reads, land here on
an unshredded
Review Comment:
CDC does not get its conf from `SparkReaderContextFactory`:
`HoodieFileGroupReaderBasedFileFormat.setSchemaEvolutionConfigs` stamps it
query-side and `CDCFileGroupIterator` builds the reader context from that
broadcast conf. Splitting the sentence - write-side services via the factory,
CDC via the file format - would keep the javadoc usable for tracing.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunClusteringProcedure.scala:
##########
@@ -149,6 +149,20 @@ class RunClusteringProcedure extends BaseProcedure
logInfo("No options")
}
+ // Normalise once so the plan stores the same trimmed list the strategies
and partitioners
+ // work from, and validate it up front, before any plan is scheduled -
whichever of `order`
+ // or `options` set it. A blank value is no sort at all, as the strategies
read it.
+ confs.get(HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS.key()).foreach
{ sortColumns =>
+ val normalized =
sortColumns.split(",").map(_.trim).filter(_.nonEmpty).mkString(",")
+ if (normalized.isEmpty) {
+ confs = confs - HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS.key()
Review Comment:
Removing the key is not the blank value the comment above describes: `confs`
is the top override in `HoodieCLIUtils.createHoodieWriteClient`, so an empty
`order` or `options` now falls through to a sort column set in the session conf
or table config, unvalidated. Storing `""` keeps the override and still reads
as no sort in `SparkSizeBasedClusteringPlanStrategy.getStrategyParams`.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunClusteringProcedure.scala:
##########
@@ -149,6 +149,20 @@ class RunClusteringProcedure extends BaseProcedure
logInfo("No options")
}
+ // Normalise once so the plan stores the same trimmed list the strategies
and partitioners
+ // work from, and validate it up front, before any plan is scheduled -
whichever of `order`
+ // or `options` set it. A blank value is no sort at all, as the strategies
read it.
+ confs.get(HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS.key()).foreach
{ sortColumns =>
+ val normalized =
sortColumns.split(",").map(_.trim).filter(_.nonEmpty).mkString(",")
+ if (normalized.isEmpty) {
+ confs = confs - HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS.key()
+ } else {
+ validateOrderColumns(normalized, metaClient)
Review Comment:
`validateOrderColumns` resolves against `getTableSchema(false)` by top-level
name, so routing `options` through it now rejects `nested.field` and
`_hoodie_*` sort columns that this route accepted before and that the
partitioners do resolve (`getNestedFieldVal`, and `new Column(name)` on the row
path). Resolving dotted paths and passing `getTableSchema(true)` there would
keep both routes working.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringOperator.java:
##########
@@ -175,6 +177,13 @@ public void open() throws Exception {
// scan successfully without schema validating exception.
this.readerSchema = HoodieSchemaUtils.asNullable(schema);
+ if (this.sortClusteringEnabled) {
+ // Reject a MAP or VARIANT sort column here, once and by name, as the
Spark and Java
+ // clients do: left alone it reaches SortOperatorGen, whose generated
comparator throws
+ // "Unsupported sort field value type" per record inside the sorter.
+ SortUtils.validateSortableColumns(sortColumns(), schema);
Review Comment:
`SortUtils.validateSortableColumns` follows Spark's
`RowOrdering.isOrderable`, so a Flink ROW, ARRAY, BLOB or VECTOR sort column
passes here, while `SortOperatorGen`'s generated `compareValues` accepts only
`byte[]` and `Comparable` - and `BinaryRowData`/`BinaryArrayData` are neither,
so those still throw "Unsupported sort field value type" per record. Rejecting
every column whose Flink `LogicalType` falls to `compareExpression`'s default
branch would match Flink's own set, or the comment could say the shared check
catches MAP and VARIANT only.
##########
hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HiveHoodieReaderContext.java:
##########
@@ -148,6 +152,35 @@ private ClosableIterator<ArrayWritable>
getFileRecordIterator(StoragePath filePa
fileSchema = dataSchema;
}
+ // Fail fast on shredded variant columns: this reader hands the file to a
plain
+ // parquet-avro read at the requested {metadata, value} projection, so a
file whose variant
+ // group carries typed_value would come back with silent nulls (the typed
rows keep their
+ // payload in typed_value, which the projection drops). Detection is
shape-based on the
+ // footer schema and anchored on the requested column being a variant, so
plain user structs
+ // of the same shape are left alone. toShreddedReadSchema recurses through
structs, array
+ // elements and map values, matching the row writer, which shreds nested
variants too.
+ // Columns not requested (e.g. count(*)) stay readable, and so does a read
whose nested column
+ // paths (hive.io.file.readNestedColumn.paths) all miss the shredded
group: Hive's parquet
+ // reader materializes only the paths it is given, and the mask rewrite
below already handles
+ // the compacted projection such a read comes back in.
+ if (isParquetOrOrc && requiredSchema.getType() == HoodieSchemaType.RECORD)
{
+ HoodieSchema shreddedReadSchema =
VariantSchemaUtils.toShreddedReadSchema(requiredSchema, fileSchema);
+ if (shreddedReadSchema != requiredSchema) {
+ List<String> shreddedPaths = new ArrayList<>();
+ collectShreddedVariantPaths(requiredSchema, shreddedReadSchema, "",
shreddedPaths);
+ List<String> offendingColumns =
HoodieColumnProjectionUtils.columnsReadingShreddedPaths(
Review Comment:
`FileGroupReaderSchemaHandler.getRequiredSchema` returns the whole table
schema when the merge mode is CUSTOM, since no merger overrides
`isProjectionCompatible`, and `columnsReadingShreddedPaths` filters only by
nested column paths - so `select id` on such a table now throws on a variant
column it never requested. Intersecting `offendingColumns` with
`HoodieColumnProjectionUtils.getReadColumnNames` on the same conf would hold
the "columns not requested stay readable" this comment promises.
##########
hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java:
##########
@@ -233,4 +268,266 @@ private RecordReader<NullWritable, ArrayWritable>
createBootstrappingRecordReade
true);
}
}
-}
\ No newline at end of file
+
+ /**
+ * The file-group-reader path fails fast on shredded variant reads inside
+ * HiveHoodieReaderContext, but a split can bypass it three ways (see
+ * HoodieInputFormatUtils.shouldUseFilegroupReader): the file group reader
disabled,
+ * schema-on-read enabled, and bootstrap splits. Those land on Hive's plain
parquet reader at
+ * the synced {metadata, value} projection, which silently nulls typed_value
- so repeat the
+ * fail-fast for them. Only reads that request a column holding a shredded
variant fail;
+ * count(*) and projections that skip the variant keep working. The footer
read is gated on a
+ * requested column whose synced Hive type embeds the variant {metadata,
value} shape, so
+ * non-variant tables never pay it: the raw columns.types string is screened
for the shape's
+ * marker before it is parsed, keeping the type parse itself off every other
table's splits.
+ *
+ * <p>The footer read here is in addition to the one Hive's
ParquetRecordReaderBase.getSplit
+ * performs right after, so a variant table pays one extra footer read per
legacy-path split that
+ * requests the variant column; non-variant tables are screened out before
it.
+ *
+ * <p>The footer's MessageType is inspected directly, without converting it
to Avro:
+ * AvroSchemaConverterWithTimestampNTZ.convertINT96 throws unless
parquet.avro.readInt96AsFixed
+ * is set (nothing in Hudi sets it), and Spark writes timestamps as INT96 by
default, so the
+ * conversion would fail the very reads this guard is careful to leave
working. The footer
+ * carries no variant logical type either way, so its groups are matched by
shape at any depth,
+ * and only at a path where the requested column's parsed Hive type declares
a variant node.
+ *
+ * <p>Hive's read column names are top-level only, so a requested struct
column does not imply
+ * its whole interior: nested column pruning is carried separately as dotted
paths
+ * (hive.io.file.readNestedColumn.paths), which
+ * {@link HoodieColumnProjectionUtils#columnsReadingShreddedPaths} applies
for both this guard
+ * and its file-group-reader sibling in HiveHoodieReaderContext.
+ *
+ * <p>The guard is best-effort throughout: a malformed columns/columns.types
pairing, an
+ * unparseable type string, or a projection that names no column all fall
through to the plain
+ * parquet reader rather than failing a read it would have served.
+ */
+ @VisibleForTesting
+ static void validateNoShreddedVariantRead(InputSplit split, JobConf job) {
+ if (!(split instanceof FileSplit)) {
+ return;
+ }
+ Path filePath = ((FileSplit) split).getPath();
+ // A native parquet log file name ends in .parquet too. The realtime path
never hands a log-only
+ // split here today, so the second clause only makes the intent explicit.
+ if
(!filePath.getName().endsWith(HoodieFileFormat.PARQUET.getFileExtension())
+ || FSUtils.isLogFile(filePath.getName())) {
+ return;
+ }
+ // Screen the raw type string before anything parses it: the TypeInfoUtils
parse below is a
+ // cost every legacy-path split of every table would otherwise pay. Only
the variant shape's
+ // marker earns the parse; the exact anchor check on the parsed types is
below.
+ String rawIoColumnTypes =
WHITESPACE.matcher(job.get(IOConstants.COLUMNS_TYPES, ""))
+ .replaceAll("")
+ .toLowerCase(Locale.ROOT);
+ if (!rawIoColumnTypes.contains(HIVE_VARIANT_SHAPE_MARKER)) {
+ return;
+ }
+ Set<String> requestedColumns =
Arrays.stream(HoodieColumnProjectionUtils.getReadColumnNames(job))
+ .map(name -> name.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toSet());
+ if (requestedColumns.isEmpty()) {
+ // Hive writes the FULL column-name list for `select *`:
HiveInputFormat.pushProjection
+ // fills in every table column with read.all.columns=false.
setReadAllColumns is only
+ // called by ProjectionPusher, on the JobConf it clones downstream of
getRecordReader, so
+ // that flag never reaches the conf seen here (verified in hive-exec
2.3.10, 3.1.3, 4.0.1).
+ // Empty names here therefore means a read that materializes no column
(count(*)) or a
+ // caller that never projected; read.all.columns, true when untouched,
is not a signal.
+ return;
+ }
+ List<String> ioColumns = HoodieColumnProjectionUtils.getIOColumns(job);
+ List<TypeInfo> ioColumnTypes;
+ try {
+ ioColumnTypes =
TypeInfoUtils.getTypeInfosFromTypeString(job.get(IOConstants.COLUMNS_TYPES,
""));
+ } catch (RuntimeException e) {
+ // The screen above strips whitespace and lower-cases; the TypeInfoUtils
parse tolerates
+ // neither, so a string it lets through can still fail to parse. Bail
out like the pairing
+ // check below rather than failing a read the plain parquet reader would
serve.
+ LOG.debug("Skipping the shredded variant guard for {}: {} did not
parse", filePath, IOConstants.COLUMNS_TYPES, e);
+ return;
+ }
+ if (ioColumns.size() != ioColumnTypes.size()) {
+ // The guard is best-effort: a malformed columns/columns.types pairing
must not fail
+ // reads the plain parquet reader would otherwise serve.
+ return;
+ }
+ // The anchor, per requested column: the Hive-form paths of every node
whose synced type is the
+ // exact node shape struct<metadata:binary,value:binary> that
HiveSchemaUtil.convertField emits
+ // for a VARIANT (HMS and Glue sync both pass doFormat=false, so no spaces
and no backticks).
+ // A struct carrying any further member is a plain user struct that
happens to hold those two,
+ // and is exempt here as it is in the sibling Spark guards. One route
fails open:
+ // TableSchemaResolver's footer fallback (see the comment near
TableSchemaResolver:118) strips
+ // shredding by shape at the top level only, so a variant shredded below
the top level can
+ // reach the metastore with typed_value still in its synced type; that
three-member struct
+ // reads as a user struct here, i.e. pre-PR behaviour.
+ Map<String, List<String>> variantPathsByColumn = new HashMap<>();
+ for (int i = 0; i < ioColumns.size(); i++) {
+ String columnName = ioColumns.get(i).toLowerCase(Locale.ROOT);
+ if (!requestedColumns.contains(columnName)) {
+ continue;
+ }
+ List<String> variantPaths = new ArrayList<>();
+ collectHiveVariantPaths(ioColumnTypes.get(i), columnName, variantPaths);
+ if (!variantPaths.isEmpty()) {
+ variantPathsByColumn.put(columnName, variantPaths);
+ }
+ }
+ if (variantPathsByColumn.isEmpty()) {
+ return;
+ }
+ StoragePath storagePath = convertToStoragePath(filePath);
+ HoodieStorage storage = HoodieStorageUtils.getStorage(storagePath,
HadoopFSUtils.getStorageConf(job));
+ MessageType fileSchema = new ParquetUtils().readMessageType(storage,
storagePath);
+ // The shredded groups the file holds at a path where the column's Hive
type declares a variant:
+ // the two sides are matched, so neither a file group of that shape under
a user struct nor a
+ // synced variant the file does not actually shred can flag the column on
its own.
+ List<String> shreddedPaths = new ArrayList<>();
+ for (Type field : fileSchema.getFields()) {
+ String columnName = field.getName().toLowerCase(Locale.ROOT);
+ List<String> variantPaths = variantPathsByColumn.get(columnName);
+ if (variantPaths == null) {
+ continue;
+ }
+ List<String> filePaths = new ArrayList<>();
+ collectShreddedVariantPaths(field, columnName, filePaths);
+
filePaths.stream().filter(variantPaths::contains).forEach(shreddedPaths::add);
+ }
+ List<String> offendingColumns =
HoodieColumnProjectionUtils.columnsReadingShreddedPaths(job, shreddedPaths);
+ if (!offendingColumns.isEmpty()) {
+ throw new HoodieException(String.format(
+ "Column(s) '%s' of %s hold a shredded variant (typed_value present);
the Hive reader "
+ + "cannot reconstruct shredded variants. Read the table with
Spark 4.1+, or "
+ + "rewrite it unshredded (e.g. cluster with "
+ + "hoodie.parquet.variant.write.shredding.enabled=false).",
+ String.join(", ", offendingColumns), filePath));
+ }
+ }
+
+ /**
+ * Collects into {@code variantPaths} the Hive-form dotted path of every
node at or beneath
+ * {@code type} whose Hive type is the synced variant shape (see {@link
#isVariantShapedStruct}),
+ * starting at {@code path}. Only struct members add a segment: Hive
truncates its nested column
+ * paths at a LIST or MAP column, so a list element and a map value share
their column's path.
+ */
+ private static void collectHiveVariantPaths(TypeInfo type, String path,
List<String> variantPaths) {
+ switch (type.getCategory()) {
+ case STRUCT: {
+ StructTypeInfo struct = (StructTypeInfo) type;
+ if (isVariantShapedStruct(struct)) {
+ variantPaths.add(path);
+ return;
+ }
+ List<String> memberNames = struct.getAllStructFieldNames();
+ List<TypeInfo> memberTypes = struct.getAllStructFieldTypeInfos();
+ for (int i = 0; i < memberNames.size(); i++) {
+ collectHiveVariantPaths(memberTypes.get(i), path + "." +
memberNames.get(i).toLowerCase(Locale.ROOT), variantPaths);
+ }
+ break;
+ }
+ case LIST:
+ collectHiveVariantPaths(((ListTypeInfo)
type).getListElementTypeInfo(), path, variantPaths);
+ break;
+ case MAP:
+ collectHiveVariantPaths(((MapTypeInfo) type).getMapValueTypeInfo(),
path, variantPaths);
+ break;
+ default:
+ break;
+ }
+ }
+
+ /**
+ * Whether {@code struct} is the Hive type a synced VARIANT gets: exactly
the two binary members
+ * {@code metadata} and {@code value}. A struct with a third member is a
user struct - including
+ * one whose third member is named typed_value, which only the footer
fallback of
+ * TableSchemaResolver can produce for a nested shredded variant.
+ */
+ private static boolean isVariantShapedStruct(StructTypeInfo struct) {
+ List<String> memberNames = struct.getAllStructFieldNames();
+ if (memberNames.size() != 2) {
+ return false;
+ }
+ List<String> lowered = memberNames.stream().map(name ->
name.toLowerCase(Locale.ROOT)).collect(Collectors.toList());
+ return lowered.contains(HoodieSchema.Variant.VARIANT_METADATA_FIELD)
+ && lowered.contains(HoodieSchema.Variant.VARIANT_VALUE_FIELD)
+ &&
struct.getAllStructFieldTypeInfos().stream().allMatch(HoodieParquetInputFormat::isBinary);
+ }
+
+ /** Whether {@code type} is the Hive {@code binary} primitive. */
+ private static boolean isBinary(TypeInfo type) {
+ return type instanceof PrimitiveTypeInfo &&
serdeConstants.BINARY_TYPE_NAME.equals(type.getTypeName());
+ }
+
+ /**
+ * Collects into {@code shreddedPaths} the Hive-form dotted path of every
shredded variant group
+ * at or beneath {@code type}: a group carrying both {@code typed_value} and
{@code metadata}.
+ * The shape is checked at every group before descending, so a shredded
element is recorded at
+ * its collection column's own path on either list layout. The walk stops at
the first shredded
+ * group on a branch - everything below it belongs to that variant.
+ *
+ * <p>Paths are lower-cased parquet field names joined by "." starting at
{@code path}, minus the
+ * levels a Hive dotted path never names, because Hive truncates its nested
column paths at a
+ * LIST or MAP column: the collection's repeated level, the synthetic level
between a LIST and
+ * its element, and a map entry's key and value. Only struct members append
a segment. Which
+ * levels those are is decided structurally, by parquet's own
backward-compatibility rule rather
+ * than by level names, so a struct element's member that happens to be
called {@code element},
+ * {@code key} or {@code value} keeps its segment: under a LIST group with a
single child, that
+ * child is the synthetic level only when it is a group with exactly one
non-repeated field whose
+ * name is neither {@code array} nor exactly {@code <list>_tuple} (the
3-level layout), and
+ * otherwise it is the element itself (the 2-level layout); under a MAP
group the single child is
+ * always the entry level, whose key and value both carry the map's own path.
+ *
+ * <p>LIST and MAP are read off OriginalType rather than the
LogicalTypeAnnotation that replaced
+ * it: parquet 1.11 and later derive one from the other, while this module
loads inside Hive,
+ * whose bundled parquet can predate the annotation class entirely (the
reason ParquetAdapter
+ * picks its implementation reflectively). A collection group shaped unlike
its annotation (no
+ * single child) is walked as a plain struct, the best-effort reading.
+ */
+ private static void collectShreddedVariantPaths(Type type, String path,
List<String> shreddedPaths) {
+ if (type.isPrimitive()) {
+ return;
+ }
+ GroupType group = type.asGroupType();
+ if (group.containsField(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD)
+ && group.containsField(HoodieSchema.Variant.VARIANT_METADATA_FIELD)) {
+ shreddedPaths.add(path);
+ return;
+ }
+ OriginalType originalType = group.getOriginalType();
+ if (originalType == OriginalType.LIST && group.getFieldCount() == 1) {
+ Type repeated = group.getType(0);
+ Type element = isSyntheticListLevel(repeated, group.getName()) ?
repeated.asGroupType().getType(0) : repeated;
+ collectShreddedVariantPaths(element, path, shreddedPaths);
+ return;
+ }
+ if (originalType == OriginalType.MAP && group.getFieldCount() == 1 &&
!group.getType(0).isPrimitive()) {
Review Comment:
Nothing in `TestHoodieParquetInputFormat` declares a `map<...>` Hive type,
so deleting this MAP arm leaves the legacy suite green while a shredded variant
under a map collects at `m.key_value.value` and never meets the `m` the Hive
side declares. `InputFormatTestUtil.writeMapShreddedVariantParquetFile` already
exists for the file-group-reader suite and would cover it in one leg here.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLegacyParquetReadPath.scala:
##########
@@ -509,6 +511,66 @@ class TestLegacyParquetReadPath extends
HoodieSparkClientTestBase with ScalaAsse
}
}
+ @Test
+ def testCowSnapshotReadWithSchemaOnReadRejectsShreddedVariant(): Unit = {
+ // The shredded-variant guard is copied into all four
*LegacyHoodieParquetFileFormat versions,
+ // and buildScan on this relation is the only caller that reaches those
copies -- DefaultSource
+ // never routes a batch read to them. Reading a shredded variant back
needs Spark 4.1+
+ // (SPARK-54410), which leaves the 4.1 and 4.2 copies as the ones this
exercises.
+ assumeTrue(HoodieSparkUtils.gteqSpark4_1, "Shredded variants need Spark
4.1+ to be read back")
+
+ // Schema-on-read models a variant as a two-field {metadata, value}
record, so the merged
+ // request clips the file's typed_value away and the typed rows would come
back with a null
+ // value residual -- silent data loss (#18285). The legacy formats must
fail loudly instead
+ // (ParquetSchemaEvolutionUtils.validateNoShreddedVariants). Forced
shredding is what puts a
+ // typed_value group under `v`; without it the file carries the unshredded
pair and there is
+ // nothing for the guard to reject.
+ val shreddedSchemaOnReadOpts = Map(
+ DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.key -> "true",
+ DataSourceWriteOptions.RECONCILE_SCHEMA.key -> "true",
+ "hoodie.parquet.variant.write.shredding.enabled" -> "true",
+ "hoodie.parquet.variant.force.shredding.schema.for.test" -> "a bigint, b
string")
+
+ // The short name is required: it resolves to the Spark 4 datasource, the
only one whose
+ // supportsDataType override accepts a VariantType column on write.
+ spark.sql(
+ """select '1' as id, 1L as ts, 'p0' as partition,
parse_json('{"a":1,"b":"b1"}') as v
+ |union all
+ |select '2' as id, 1L as ts, 'p0' as partition,
parse_json('{"a":2,"b":"b2"}') as v""".stripMargin)
+ .write.format("hudi")
+ .options(writeOpts ++ shreddedSchemaOnReadOpts)
+ .option(DataSourceWriteOptions.OPERATION.key,
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+
+ val readOpts = Map(DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.key ->
"true")
+ val metaClient = createMetaClient(spark, basePath)
+ assertTrue(BaseFileOnlyRelation(sqlContext, metaClient,
legacyReadOpts(readOpts), None).hasSchemaOnRead,
+ "The write must have recorded an InternalSchema, otherwise the guard's
branch is never taken")
+
+ // Only the relation's own buildScan is exercised here, not the
HadoopFsRelation conversion:
+ // buildScan embeds the query schema into the reader's Hadoop conf
(HoodieBaseRelation
+ // .embedInternalSchema), whereas the converted relation is read with the
plain session conf,
+ // so shouldUseInternalSchema is false there and the guard is not on that
path at all. That is
+ // also why DefaultSource keeps BaseFileOnlyRelation itself under
schema-on-read.
+ val thrown = assertThrows(classOf[Throwable]) {
+ legacyRelationDf(readOpts).select("v").collect()
+ }
+ val causes = Iterator.iterate(thrown: Throwable)(_.getCause).takeWhile(_
!= null).take(10).toSeq
+ assertTrue(causes.exists(c => c.isInstanceOf[HoodieException]
+ && String.valueOf(c.getMessage).contains("shredded variant")),
+ s"Expected the shredded-variant rejection but got: $thrown")
+
+ // The guard's empty-projection carve-out (count(*) reads no column data
and must keep working)
+ // is pinned on the file-group-reader path by the count(*) legs in
TestVariantShreddingMixedLayouts,
Review Comment:
None of the `count(*)` legs in `TestVariantShreddingMixedLayouts` run under
`hoodie.schema.on.read.enable`, so the empty-projection carve-out is unpinned
and dropping the `requiredSchema.nonEmpty` gate from all five sites leaves both
suites green. One `count(*)` inside the existing schema-on-read block of
"Schema-on-read reads of shredded variant files fail fast" would pin it, or
this sentence could drop the claim.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/VariantShreddingTestSupport.scala:
##########
@@ -0,0 +1,709 @@
+/*
+ * 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.spark.sql.hudi.dml.schema
+
+import org.apache.hudi.DataSourceReadOptions
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
+import org.apache.hudi.common.model.WriteOperationType
+import org.apache.hudi.storage.StoragePath
+import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileSystem, Path => HadoopPath}
+import org.apache.parquet.example.data.Group
+import org.apache.parquet.hadoop.{ParquetFileReader, ParquetReader}
+import org.apache.parquet.hadoop.api.ReadSupport
+import org.apache.parquet.hadoop.example.GroupReadSupport
+import org.apache.parquet.hadoop.util.HadoopInputFile
+import org.apache.parquet.schema.{GroupType, MessageType, Type}
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
+import
org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase.getLastCommitMetadata
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+/**
+ * Shared helpers for variant-shredding tests: parquet-footer layout
inspection, a row-level
+ * typed-vs-residual inspector, a shape-drift data generator, and a
write-layout toggle. Mixed
+ * into [[TestVariantDataType]] and [[TestVariantShreddingMixedLayouts]].
+ */
+trait VariantShreddingTestSupport { self: HoodieSparkSqlTestBase =>
+
+ import VariantShreddingTestSupport._
+
+ /** The `(id int, v variant, ts long)` table both suites use, with the knobs
they vary. */
+ protected def createVariantTable(tableName: String,
+ tablePath: String,
+ tableType: String,
+ props: Seq[String] = Seq.empty,
+ extraCols: String = "",
+ preCombine: Boolean = true): Unit = {
Review Comment:
No call site passes `preCombine = false`, so every table in the suite
carries `preCombineField = 'ts'` and the header's "only EVENT_TIME/COMMIT_TIME
ordering is swept" claims a mode nothing reaches. Either add a COMMIT_TIME leg
through this parameter or drop it along with the ordering claim.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala:
##########
@@ -0,0 +1,1162 @@
+/*
+ * 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.spark.sql.hudi.dml.schema
+
+import org.apache.hudi.HoodieSparkUtils
+import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
+import org.apache.hudi.core.io.storage.VariantShreddingInferenceFileWriter
+import org.apache.hudi.testutils.DataSourceTestUtils
+
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
+import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
+import
org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase.getLastCommitMetadata
+
+/**
+ * Mixed-layout variant shredding matrix: files with DIFFERENT typed_value
layouts in one table,
+ * shredded/unshredded splits between base and log files, and rows inside one
file that fell back
+ * to the residual value column, driven through compaction, clustering, merges
and every Spark
+ * read mode. Complements [[TestVariantDataType]], whose shredded tests force
ONE layout per
+ * table.
+ *
+ * Layouts are toggled per commit or table service through session confs
(session hoodie.* confs
+ * override tblproperties for SQL DML and for the
run_compaction/run_clustering procedures alike).
+ * Every test is gated on Spark 4.1+, which is exactly the set of profiles
that register #18961's
+ * per-file shredding-schema inferrer (pinned by TestVariantDataType's "A
shredding-schema
+ * inferrer is registered for every Spark version that ships one"), so an
[[Inferred]] leg here
+ * always infers rather than silently degrading to an unshredded write.
+ *
+ * Deliberately not covered here:
+ * - Custom payloads: FileGroupRecordBuffer.getProjectedTransformer
short-circuits the variant
+ * log-block projection when payload classes are present (#18674), so that
is a real,
+ * explicitly UNTESTED variant branch; PartialUpdateMode and the CUSTOM
merge mode are
+ * likewise unreached (only EVENT_TIME/COMMIT_TIME ordering is swept).
+ * - Multi-writer OCC: conflict resolution is key/instant based and never
inspects layouts; the
+ * mixed-file outcomes it can produce are the same ones pinned here.
+ */
+class TestVariantShreddingMixedLayouts extends HoodieSparkSqlTestBase with
VariantShreddingTestSupport {
+
+ import VariantShreddingTestSupport._
+ import VariantShreddingTestSupport.VariantShape._
+
+ private val SPARK_4_1_GATE = "Shredded variant read-back requires Spark 4.1
or higher"
+
+ /** One insert commit per layout; returns the completed instant of each
commit, in order. */
+ private def seedMixedLayoutTable(tableName: String,
+ tablePath: String,
+ layouts: Seq[(WriteLayout, Seq[(Range,
VariantShape)])]): Seq[String] = {
+ layouts.map { case (layout, segments) =>
+ withWriteLayout(layout) {
+ spark.sql(s"insert into $tableName ${variantSourceSql(segments)}")
+ }
+ latestCompletedInstant(tablePath)
+ }
+ }
+
+ /** scheduleAndExecute compaction; the options carry the NUM_COMMITS trigger
so one delta commit suffices. */
+ private def runCompaction(tableName: String): Unit = {
+ spark.sql(s"call run_compaction(op => 'scheduleandexecute', table =>
'$tableName', " +
+ "options => 'hoodie.compact.inline.max.delta.commits=1')")
+ }
+
+ private def runClustering(tableName: String, rowWriter: Boolean): Unit = {
+ spark.sql(s"call run_clustering(table => '$tableName', " +
+ s"options => 'hoodie.datasource.write.row.writer.enable=$rowWriter')")
+ }
+
+ //
-----------------------------------------------------------------------------------------------
+ // A. Mixed records inside one file
+ //
-----------------------------------------------------------------------------------------------
+
+ test("Forced shredding: non-matching rows fall back to the residual in the
same file") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ withVariantTable("same-file mix", "cow") { (tableName, tablePath, leg) =>
+ // One insert, one file: rows 0-9 match the forced schema exactly; 10-14
conflict on the
+ // type of a (string into a bigint slot -> per-field residual); 15-19
carry disjoint keys
+ // (root residual); 20-22 are root scalars and 23 a JSON null (no object
typed_value);
+ // 24 is a SQL NULL variant.
+ val segments = Seq(
+ (0 until 10, ObjA),
+ (10 until 15, ObjAConflict),
+ (15 until 20, ObjB),
+ (20 until 23, RootScalar),
+ (23 until 24, JsonNull),
+ (24 until 25, SqlNull))
+ withWriteLayout(Forced("a bigint, b string")) {
+ spark.sql(s"insert into $tableName ${variantSourceSql(segments)}")
+ }
+
+ val files = listDataParquetFiles(tablePath)
+ assert(files.size == 1, s"[$leg] expected exactly one data file, got
$files")
+ assertVariantLayout(tablePath, shredded = true, leg)
+
+ // Physical placement per the shredding spec: objects always materialize
typed_value;
+ // unmatched FIELDS go to the per-field residual, unmatched KEYS to the
root residual;
+ // non-objects (scalars, arrays, JSON null) live entirely in the root
residual.
+ val stats = inspectVariantRows(files.head)
+ assert(stats.rows == 25, s"[$leg] rows: $stats")
+ assert(stats.nullVariants == 1, s"[$leg] null variants: $stats")
+ assert(stats.rootTyped == 20, s"[$leg] object rows with typed_value:
$stats")
+ assert(stats.rootResidual == 9, s"[$leg] root residual rows (ObjB 5 +
scalars 3 + json null 1): $stats")
+ assert(stats.fieldTyped("a") == 10, s"[$leg] typed a: $stats")
+ assert(stats.fieldResidual("a") == 5, s"[$leg] residual a (type
conflict): $stats")
+ assert(stats.fieldTyped("b") == 15, s"[$leg] typed b: $stats")
+
+ assertVariantSegments(tableName, leg, Seq(("v", segments)))
+
+ // Update rows served from the typed slot and from the residual: the
AVRO record type
+ // reconstructs both through HoodieVariantReconstruction, SPARK natively.
+ withWriteLayout(Forced("a bigint, b string")) {
+ spark.sql(s"""update $tableName set v =
parse_json('{"a":100,"b":"bu"}'), ts = 1001 where id = 20""")
+ spark.sql(s"""update $tableName set v =
parse_json('{"a":101,"b":"bv"}'), ts = 1001 where id = 5""")
+ }
+ checkAnswer(s"select id, cast(v as string), ts from $tableName where id
in (5, 12, 20) order by id")(
+ Seq(5, """{"a":101,"b":"bv"}""", 1001),
+ Seq(12, """{"a":"s12","b":"b12"}""", 1000),
+ Seq(20, """{"a":100,"b":"bu"}""", 1001)
+ )
+ assertVariantLayout(tablePath, shredded = true, leg)
+ }
+ }
+
+ //
-----------------------------------------------------------------------------------------------
+ // B. Mixed files inside one table
+ //
-----------------------------------------------------------------------------------------------
+
+ test("Each commit keeps its own layout; snapshot, time travel, incremental
and RO read them all") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // Read-mode test: layouts are writer-side and every layout is written
identically by both
+ // record types, so the sweep would only re-run the same reads. SPARK
pinned.
+ withVariantTable("mixed-files", "cow", props =
Seq(NEW_FILE_GROUP_PER_COMMIT),
+ recordTypes = Seq(HoodieRecordType.SPARK)) { (tableName, tablePath, leg)
=>
+ // Four commits, four layouts, one file each (small.file.limit=0 keeps
every commit in its
+ // own file group). The last commit infers {c, d} from its own ObjB rows.
+ val instants = seedMixedLayoutTable(tableName, tablePath, Seq(
+ (Unshredded, Seq((0 until 2, ObjA))),
+ (Forced("a bigint, b string"), Seq((2 until 4, ObjA))),
+ (Forced("b string"), Seq((4 until 6, ObjA))),
+ (Inferred, Seq((6 until 8, ObjB)))))
+
+ assertLayoutsByInstant(baseLayouts(tablePath), leg)(
+ instants(0) -> None,
+ instants(1) -> Some(Seq("a", "b")),
+ instants(2) -> Some(Seq("b")),
+ instants(3) -> Some(Seq("c", "d")))
+
+ // Snapshot reads every layout.
+ assertVariantSegments(tableName, leg, Seq(("v", Seq(
+ (0 until 6, ObjA), (6 until 8, ObjB)))))
+
+ // Time travel at the second commit sees only the first two layouts.
+ checkAnswer(s"select id, cast(v as string) from $tableName timestamp as
of '${instants(1)}' order by id")(
+ Seq(0, """{"a":0,"b":"b0"}"""),
+ Seq(1, """{"a":1,"b":"b1"}"""),
+ Seq(2, """{"a":2,"b":"b2"}"""),
+ Seq(3, """{"a":3,"b":"b3"}""")
+ )
+
+ // Incremental over the full range returns the latest state of all eight
keys, values
+ // intact (a count alone would pass even if v reconstructed as all-null).
+ val incRows = incrementalIdAndVariant(tablePath)
+ assert(incRows.length == 8, s"[$leg] incremental over the full range
should see all rows")
+ incRows.foreach { row =>
+ val id = row.getInt(0)
+ val expected = if (id < 6) s"""{"a":$id,"b":"b$id"}""" else
s"""{"c":$id,"d":true}"""
+ assert(row.getString(1) == expected,
+ s"[$leg] incremental id=$id: expected $expected, got
${row.getString(1)}")
+ }
+
+ // Read-optimized on COW equals the snapshot, values intact.
+ checkAnswer(s"select id, cast(v as string) from hudi_query('$tableName',
'read_optimized') " +
+ "where id in (0, 6) order by id")(
+ Seq(0, """{"a":0,"b":"b0"}"""),
+ Seq(6, """{"c":6,"d":true}""")
+ )
+ }
+ }
+
+ test("Small-file bin-pack rewrites the file under the layout of the incoming
commit") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // Default small.file.limit on purpose: each insert bin-packs into the
first file group
+ // and rewrites it (HoodieConcatHandle -> HoodieMergeHelper on the AVRO
record type).
+ // The value round-trip of that merge is owned by TestVariantDataType's
small-file test;
+ // this one exists for the per-instant LAYOUT pin below.
+ withVariantTable("bin-pack layout flip", "cow") { (tableName, tablePath,
leg) =>
+ withWriteLayout(Forced("a bigint, b string")) {
+ spark.sql(s"""insert into $tableName values (1,
parse_json('{"a":1,"b":"b1"}'), 1000)""")
+ }
+ val instant1 = latestCompletedInstant(tablePath)
+ withWriteLayout(Unshredded) {
+ spark.sql(s"""insert into $tableName values (2,
parse_json('{"a":2,"b":"b2"}'), 1000)""")
+ }
+ val instant2 = latestCompletedInstant(tablePath)
+ withWriteLayout(Forced("a bigint")) {
+ spark.sql(s"""insert into $tableName values (3,
parse_json('{"a":3,"b":"b3"}'), 1000)""")
+ }
+ val instant3 = latestCompletedInstant(tablePath)
+
+ assertSingleFileGroup(tablePath, leg)
+ // The rewrite re-derives the layout from the CURRENT write config; the
input file's
+ // layout is never consulted. Older file versions keep their own layouts.
+ assertLayoutsByInstant(baseLayouts(tablePath), leg)(
+ instant1 -> Some(Seq("a", "b")),
+ instant2 -> None,
+ instant3 -> Some(Seq("a")))
+
+ checkAnswer(s"select id, cast(v as string), ts from $tableName order by
id")(
+ Seq(1, """{"a":1,"b":"b1"}""", 1000),
+ Seq(2, """{"a":2,"b":"b2"}""", 1000),
+ Seq(3, """{"a":3,"b":"b3"}""", 1000)
+ )
+ }
+ }
+
+ //
-----------------------------------------------------------------------------------------------
+ // C. MOR compaction over base/log layout splits
+ //
-----------------------------------------------------------------------------------------------
+
+ test("MOR compaction merges logs of three layouts and re-derives the base
layout per service run") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // INMEMORY sends MOR inserts to log files; compaction runs via the
procedure so each run
+ // can happen under its own layout confs.
+ withVariantTable("compaction layout split", "mor", props = Seq(
+ "hoodie.index.type = 'INMEMORY'", "hoodie.compact.inline = 'false'")) {
(tableName, tablePath, leg) =>
+ withWriteLayout(Forced("a bigint, b string")) {
+ spark.sql(s"""insert into $tableName values (1,
parse_json('{"a":1,"b":"b1"}'), 1000)""")
+ }
+ val instant1 = latestCompletedInstant(tablePath)
+ withWriteLayout(Unshredded) {
+ spark.sql(s"""insert into $tableName values (2,
parse_json('{"a":2,"b":"b2"}'), 1000), """ +
+ """(3, parse_json('{"a":3,"b":"b3"}'), 1000), (4,
parse_json('{"a":4,"b":"b4"}'), 1000)""")
+ }
+ val instant2 = latestCompletedInstant(tablePath)
+ withWriteLayout(Inferred) {
+ spark.sql(s"""insert into $tableName values (5,
parse_json('{"c":5,"d":true}'), 1000), """ +
+ """(6, parse_json('{"c":6,"d":true}'), 1000)""")
+ }
+ val instant3 = latestCompletedInstant(tablePath)
+
+ assertResult(true)(DataSourceTestUtils.isLogFileOnly(tablePath))
+ // On the default table version the data logs are native parquet, each
with the layout of
+ // its own commit. (The SPARK withRecordType leg sets the parquet log
block format, the
+ // AVRO leg avro blocks, but write version >= 10 writes native log FILES
either way.)
+ assertLayoutsByInstant(nativeLogLayouts(tablePath), leg)(
+ instant1 -> Some(Seq("a", "b")),
+ instant2 -> None,
+ instant3 -> Some(Seq("c", "d")))
+
+ // Merge-on-read snapshot over the three-layout split, before any base
file exists.
+ checkAnswer(s"select id, cast(v as string) from $tableName order by id")(
+ Seq(1, """{"a":1,"b":"b1"}"""),
+ Seq(2, """{"a":2,"b":"b2"}"""),
+ Seq(3, """{"a":3,"b":"b3"}"""),
+ Seq(4, """{"a":4,"b":"b4"}"""),
+ Seq(5, """{"c":5,"d":true}"""),
+ Seq(6, """{"c":6,"d":true}""")
+ )
+
+ // Compaction 1 under Inferred: reads all three log layouts, infers the
base layout from
+ // the merged rows.
+ withWriteLayout(Inferred) {
+ runCompaction(tableName)
+ }
+ assertResult(false)(DataSourceTestUtils.isLogFileOnly(tablePath))
+ assertCompactionCount(tablePath, 1, leg)
+ val base1 = baseLayouts(tablePath)
+ assertAllShredded(base1, shredded = true, s"$leg compacted base under
Inferred")
+ // 6 rows: a and b on 4 (66 percent), c and d on 2 (33 percent) - all
clear the 10
+ // percent inference bar.
+ base1.foreach(l => assert(l.typedFields.toSet == Set("a", "b", "c", "d"),
+ s"[$leg] inferred typed_value should carry all four keys:
${l.typedFields}"))
+ checkAnswer(s"select id, cast(v as string) from $tableName where id in
(1, 5) order by id")(
+ Seq(1, """{"a":1,"b":"b1"}"""),
+ Seq(5, """{"c":5,"d":true}""")
+ )
+ checkAnswer(s"select id, cast(v as string) from hudi_query('$tableName',
'read_optimized') " +
+ "where id in (1, 5) order by id")(
+ Seq(1, """{"a":1,"b":"b1"}"""),
+ Seq(5, """{"c":5,"d":true}""")
+ )
+
+ // Round 2: updates under two further layouts, compaction under
Unshredded. The service
+ // reads a shredded base plus mixed logs and must strip typed_value on
the way out.
+ withWriteLayout(Forced("a bigint")) {
+ spark.sql(s"""update $tableName set v =
parse_json('{"a":22,"b":"b22"}'), ts = 1001 where id = 2""")
+ }
+ withWriteLayout(Unshredded) {
+ spark.sql(s"""update $tableName set v =
parse_json('{"a":33,"b":"b33"}'), ts = 1001 where id = 3""")
+ }
+ // A delete block (no data column) between the differently-shredded
logs: the merged read
+ // and the following compaction must step over it without a layout to
anchor on.
+ withWriteLayout(Forced("a bigint")) {
+ spark.sql(s"delete from $tableName where id = 6")
+ }
+ // Merge-on-read over shredded base + {a}-shredded log + unshredded log
+ delete block.
+ checkAnswer(s"select id, cast(v as string) from $tableName where id in
(2, 3, 5) order by id")(
+ Seq(2, """{"a":22,"b":"b22"}"""),
+ Seq(3, """{"a":33,"b":"b33"}"""),
+ Seq(5, """{"c":5,"d":true}""")
+ )
+ withWriteLayout(Unshredded) {
+ runCompaction(tableName)
+ }
+ assertCompactionCount(tablePath, 2, leg)
+ val compact2Instant = latestCompletedInstant(tablePath)
+ val base2 = baseLayouts(tablePath).filter(_.instantTime ==
compact2Instant)
+ assertAllShredded(base2, shredded = false, s"$leg base of the compaction
under Unshredded")
+ checkAnswer(s"select id, cast(v as string) from $tableName where id in
(2, 3) order by id")(
+ Seq(2, """{"a":22,"b":"b22"}"""),
+ Seq(3, """{"a":33,"b":"b33"}""")
+ )
+
+ // Round 3: compaction under Inferred again, this time reading an
UNSHREDDED base plus a
+ // shredded log.
+ withWriteLayout(Inferred) {
+ spark.sql(s"""update $tableName set v =
parse_json('{"a":44,"b":"b44"}'), ts = 1001 where id = 4""")
+ runCompaction(tableName)
+ }
+ assertCompactionCount(tablePath, 3, leg)
+ val compact3Instant = latestCompletedInstant(tablePath)
+ val base3 = baseLayouts(tablePath).filter(_.instantTime ==
compact3Instant)
+ assertAllShredded(base3, shredded = true, s"$leg base of the second
compaction under Inferred")
+
+ checkAnswer(s"select id, cast(v as string) from $tableName order by id")(
+ Seq(1, """{"a":1,"b":"b1"}"""),
+ Seq(2, """{"a":22,"b":"b22"}"""),
+ Seq(3, """{"a":33,"b":"b33"}"""),
+ Seq(4, """{"a":44,"b":"b44"}"""),
+ Seq(5, """{"c":5,"d":true}""")
+ )
+ // Incremental over the full range sees the latest value of every LIVE
key (id 6 deleted),
+ // values intact - a bare count would pass with v all-null.
+ val incRows = incrementalIdAndVariant(tablePath)
+ assert(incRows.map(r => (r.getInt(0), r.getString(1))).toSeq == Seq(
+ (1, """{"a":1,"b":"b1"}"""),
+ (2, """{"a":22,"b":"b22"}"""),
+ (3, """{"a":33,"b":"b33"}"""),
+ (4, """{"a":44,"b":"b44"}"""),
+ (5, """{"c":5,"d":true}""")
+ ), s"[$leg] incremental over the full range, got: ${incRows.mkString(",
")}")
+ }
+ }
+
+ test("Table version 9 legacy log blocks stay unshredded and compact onto a
shredded base") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ withVariantTable("table version 9", "mor", props = Seq(
+ "hoodie.write.table.version = '9'",
+ "hoodie.index.type = 'INMEMORY'",
+ "hoodie.compact.inline = 'false'")) { (tableName, tablePath, leg) =>
+ withWriteLayout(Inferred) {
+ spark.sql(s"""insert into $tableName values (1,
parse_json('{"key":"value1"}'), 1000)""")
+ spark.sql(s"""insert into $tableName values (2,
parse_json('{"key":"value2"}'), 1000)""")
+ }
+ assertResult(true)(DataSourceTestUtils.isLogFileOnly(tablePath))
+ // Write version 9 writes the legacy inline log format (avro blocks on
the AVRO record
+ // type leg, inline parquet data blocks on the SPARK leg), never native
parquet log files;
+ // neither inline form shreds, so the shredded layout materializes only
at compaction.
+ assert(nativeLogLayouts(tablePath).isEmpty,
+ s"[$leg] table version 9 must not write native parquet log files")
+
+ withWriteLayout(Inferred) {
+ runCompaction(tableName)
+ }
+ assertResult(false)(DataSourceTestUtils.isLogFileOnly(tablePath))
+ val base1 = baseLayouts(tablePath)
+ assertAllShredded(base1, shredded = true, s"$leg compacted base")
+ base1.foreach(l => assert(l.typedFields == Seq("key"),
+ s"[$leg] typed_value should carry key: ${l.typedFields}"))
+
+ // Legacy log over the shredded base, then a second compaction reads
base + legacy log.
+ withWriteLayout(Inferred) {
+ spark.sql(s"""update $tableName set v =
parse_json('{"key":"v1-updated"}'), ts = 1001 where id = 1""")
+ }
+ checkAnswer(s"select id, cast(v as string) from $tableName order by id")(
+ Seq(1, """{"key":"v1-updated"}"""),
+ Seq(2, """{"key":"value2"}""")
+ )
+ withWriteLayout(Inferred) {
+ runCompaction(tableName)
+ }
+ checkAnswer(s"select id, cast(v as string) from $tableName order by id")(
+ Seq(1, """{"key":"v1-updated"}"""),
+ Seq(2, """{"key":"value2"}""")
+ )
+ }
+ }
+
+ //
-----------------------------------------------------------------------------------------------
+ // D. Clustering over heterogeneous inputs
+ //
-----------------------------------------------------------------------------------------------
+
+ test("Clustering rewrites heterogeneous files into the configured layout") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ Seq(true, false).foreach { rowWriter =>
+ val recordTypes = clusteringRecordTypes(rowWriter)
+ Seq(Unshredded, Inferred).foreach { outLayout =>
+ // The unshredded output cell only re-pins the unshredded rewrite,
which the evolution
+ // test's clustering leg already sweeps over both record types; SPARK
alone here, so this
+ // test runs 5 clustering jobs instead of 6.
+ val cellRecordTypes = if (outLayout == Unshredded)
Seq(HoodieRecordType.SPARK) else recordTypes
+ withVariantTable(s"clustering rowWriter=$rowWriter out=$outLayout",
"cow",
+ props = Seq(NEW_FILE_GROUP_PER_COMMIT), recordTypes =
cellRecordTypes) { (tableName, tablePath, leg) =>
+ val instants = seedMixedLayoutTable(tableName, tablePath, Seq(
+ (Forced("a bigint, b string"), Seq((0 until 2, ObjA))),
+ (Unshredded, Seq((2 until 4, ObjA))),
+ (Inferred, Seq((4 until 6, ObjB)))))
+
+ withWriteLayout(outLayout) {
+ runClustering(tableName, rowWriter)
+ }
+ val clusteringInstant = completedClusteringInstant(tablePath, leg)
+ val outFiles = baseLayouts(tablePath).filter(_.instantTime ==
clusteringInstant)
+ assert(outFiles.nonEmpty, s"[$leg] clustering should have written
base files")
+ if (outLayout == Unshredded) {
+ outFiles.foreach(l => assert(!l.isShredded,
+ s"[$leg] clustering under Unshredded must write unshredded
output: ${l.path}"))
+ } else {
+ // 6 rows: a, b on 4 and c, d on 2 - all clear the 10 percent bar.
+ outFiles.foreach(l => assert(l.typedFields.toSet == Set("a", "b",
"c", "d"),
+ s"[$leg] inferred output typed_value should carry all keys:
${l.typedFields}"))
+ }
+
+ // Values survive the rewrite; the pre-clustering slice stays
readable via time travel.
+ assertVariantSegments(tableName, leg, Seq(("v", Seq(
+ (0 until 4, ObjA), (4 until 6, ObjB)))))
+ checkAnswer(s"select id, cast(v as string) from $tableName " +
+ s"timestamp as of '${instants(2)}' where id in (0, 4) order by
id")(
+ Seq(0, """{"a":0,"b":"b0"}"""),
+ Seq(4, """{"c":4,"d":true}""")
+ )
+ }
+ }
+ }
+ }
+
+ test("MOR clustering folds log files of another layout into the rewritten
base") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ Seq(true, false).foreach { rowWriter =>
+ val recordTypes = clusteringRecordTypes(rowWriter)
+ // No INMEMORY index: the first insert creates a base file, the update
goes to a log.
+ withVariantTable(s"mor clustering rowWriter=$rowWriter", "mor",
+ props = Seq("hoodie.compact.inline = 'false'"), recordTypes =
recordTypes) { (tableName, tablePath, leg) =>
+ withWriteLayout(Forced("a bigint, b string")) {
+ spark.sql(s"""insert into $tableName values (1,
parse_json('{"a":1,"b":"b1"}'), 1000), """ +
+ """(2, parse_json('{"a":2,"b":"b2"}'), 1000)""")
+ }
+ withWriteLayout(Unshredded) {
+ spark.sql(s"""update $tableName set v =
parse_json('{"a":10,"b":"b10"}'), ts = 1001 where id = 1""")
+ }
+ // The slice going into clustering: a shredded base plus an unshredded
native log.
+ assertAllShredded(baseLayouts(tablePath), shredded = true, s"$leg
pre-clustering base")
+ assertAllShredded(nativeLogLayouts(tablePath), shredded = false,
s"$leg pre-clustering log")
+
+ withWriteLayout(Inferred) {
+ runClustering(tableName, rowWriter)
+ }
+ val clusteringInstant = completedClusteringInstant(tablePath, leg)
+ val outFiles = baseLayouts(tablePath).filter(_.instantTime ==
clusteringInstant)
+ assert(outFiles.nonEmpty, s"[$leg] clustering should have written base
files")
+ outFiles.foreach(l => assert(l.isShredded,
+ s"[$leg] clustering under Inferred must write shredded output:
${l.path}"))
+
+ // The clustered base carries the merged (updated) row.
+ checkAnswer(s"select id, cast(v as string) from $tableName order by
id")(
+ Seq(1, """{"a":10,"b":"b10"}"""),
+ Seq(2, """{"a":2,"b":"b2"}""")
+ )
+ }
+ }
+ }
+
+ //
-----------------------------------------------------------------------------------------------
+ // E. Read modes over mixed layouts
+ //
-----------------------------------------------------------------------------------------------
+
+ test("variant_get filters and projections resolve per file across mixed
layouts") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // Read-mode test; SPARK pinned (see the mixed-files test above).
+ Seq("true", "false").foreach { pushIntoScan =>
+ withSQLConf("spark.sql.variant.pushVariantIntoScan" -> pushIntoScan) {
+ withVariantTable(s"cow pushVariantIntoScan=$pushIntoScan", "cow",
+ props = Seq(NEW_FILE_GROUP_PER_COMMIT), recordTypes =
Seq(HoodieRecordType.SPARK)) {
+ (tableName, tablePath, leg) =>
+ // $.a is typed in file 1, residual (unshredded) in file 2, a
type-conflicted residual
+ // in file 3 and absent in file 4.
+ seedMixedLayoutTable(tableName, tablePath, Seq(
+ (Forced("a bigint"), Seq((0 until 10, ObjA))),
+ (Unshredded, Seq((10 until 20, ObjA))),
+ (Forced("a bigint"), Seq((20 until 30, ObjAConflict))),
+ (Inferred, Seq((30 until 40, ObjB)))))
+
+ // Typed and residual rows answer alike; the string a declines the
cast, the missing
+ // a returns null.
+ val aValues = spark.sql(
+ s"select id, try_variant_get(v, '$$.a', 'bigint') from $tableName
order by id").collect()
+ assert(aValues.length == 40, s"[$leg] row count")
+ aValues.foreach { row =>
+ val id = row.getInt(0)
+ val expected: Any = if (id < 20) id.toLong else null
+ val actual = if (row.isNullAt(1)) null else row.getLong(1)
+ assert(actual == expected, s"[$leg] id=$id: expected $expected,
got $actual")
+ }
+
+ checkAnswer(
+ s"select count(*) from $tableName where try_variant_get(v, '$$.a',
'bigint') > 5")(Seq(14))
+ checkAnswer(
+ s"select id from $tableName where variant_get(v, '$$.b', 'string')
= 'b25'")(Seq(25))
+ checkAnswer(
+ s"select count(*) from $tableName where try_variant_get(v, '$$.d',
'boolean')")(Seq(10))
+ checkAnswer(s"select count(*) from $tableName where v is
null")(Seq(0))
+ assertVariantSegments(tableName, leg, Seq(("v", Seq(
+ (0 until 20, ObjA), (20 until 30, ObjAConflict), (30 until 40,
ObjB)))))
+ }
+ }
+ }
+
+ // MOR: the same path is typed in the base, then updated through an
unshredded log and a
+ // shredded log; the merged read serves each row from a different physical
slot.
+ Seq("true", "false").foreach { pushIntoScan =>
+ withSQLConf("spark.sql.variant.pushVariantIntoScan" -> pushIntoScan) {
+ withVariantTable(s"mor pushVariantIntoScan=$pushIntoScan", "mor",
+ props = Seq("hoodie.compact.inline = 'false'"), recordTypes =
Seq(HoodieRecordType.SPARK)) {
+ (tableName, tablePath, leg) =>
+ withWriteLayout(Forced("a bigint")) {
+ spark.sql(s"insert into $tableName ${variantSourceSql(Seq((0 until
10, ObjA)))}")
+ }
+ withWriteLayout(Unshredded) {
+ spark.sql(s"update $tableName set " +
+ s"""v = parse_json(concat('{"a":"s', id, '","b":"b', id, '"}')),
ts = 1001 """ +
+ "where id >= 5")
+ }
+ withWriteLayout(Forced("a bigint")) {
+ spark.sql(s"update $tableName set " +
+ s"""v = parse_json(concat('{"a":', 100 + id, ',"b":"b', id,
'"}')), ts = 1002 """ +
+ "where id < 3")
+ }
+
+ val aValues = spark.sql(
+ s"select id, try_variant_get(v, '$$.a', 'bigint') from $tableName
order by id").collect()
+ assert(aValues.length == 10, s"[$leg] row count")
+ aValues.foreach { row =>
+ val id = row.getInt(0)
+ val expected: Any = if (id < 3) 100L + id else if (id < 5)
id.toLong else null
+ val actual = if (row.isNullAt(1)) null else row.getLong(1)
+ assert(actual == expected, s"[$leg] id=$id: expected $expected,
got $actual")
+ }
+ checkAnswer(
+ s"select count(*) from $tableName where try_variant_get(v, '$$.a',
'bigint') > 100")(Seq(2))
+ checkAnswer(
+ s"select id from $tableName where variant_get(v, '$$.b', 'string')
= 'b7'")(Seq(7))
+ }
+ }
+ }
+ }
+
+ test("Schema-on-read reads of shredded variant files fail fast") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ withVariantTable("schema-on-read", "cow", recordTypes =
Seq(HoodieRecordType.SPARK)) {
+ (tableName, tablePath, leg) =>
+ withWriteLayout(Forced("a bigint")) {
+ spark.sql(s"""insert into $tableName values (1, parse_json('{"a":1}'),
1000)""")
+ }
+
+ withSQLConf("hoodie.schema.on.read.enable" -> "true") {
+ // Committing a schema-on-read DDL stores the internal schema; reads
under
+ // hoodie.schema.on.read.enable then request the internal-schema form
of the variant
+ // ({metadata, value}), which clips typed_value away. With
PushVariantIntoScan disabled,
+ // that would return silent nulls for the typed rows - the guard must
fire instead
+ // (#18285 tracks real reconstruction under schema-on-read).
+ spark.sql(s"alter table $tableName add columns (note string)")
+ withSQLConf("spark.sql.variant.pushVariantIntoScan" -> "false") {
+ checkNestedExceptionContains(
+ () => spark.sql(s"select id, cast(v as string), note from
$tableName").collect())(
+ "shredded variant")
+ }
+ // Under the default PushVariantIntoScan rewrite the read fails
through the guard as
+ // well: pruning treats the rewritten ordinal-named struct as the
variant column itself
+ // (SparkInternalSchemaConverter.isVariantRewriteStruct), so the guard
sees the request
+ // and rejects it up front instead of an engine-internal pruning error
or codegen NPE.
+ // Pinned on the rewrite arm's own wording: both messages carry
"cannot reconstruct", so
+ // matching on that alone would stay green with the rewrite arm gone.
Fix is #18285.
+ checkNestedExceptionContains(
+ () => spark.sql(s"select id, cast(v as string), note from
$tableName").collect())(
+ "pushVariantIntoScan")
+ }
+
+ // Known #18285 residue, documented rather than pinned: the
schema-on-read DDL also
+ // rewrites the CATALOG schema through the internal-schema converter,
which has no VARIANT
+ // arm, so the catalog column degrades to a plain struct<metadata,value>
(the resolved
+ // avro table schema keeps its variant logical type). Plain reads of the
table after the
+ // DDL request that struct and fail in Spark before any Hudi hook.
+ }
+
+ // The guard recurses: a NESTED shredded variant (struct<inner: variant>,
written by the
+ // bulk-insert row writer, the one production writer that shreds below the
top level) fails
+ // fast too, instead of slipping past a top-level-only walk.
+ withNestedVariantTable("nested schema-on-read", recordTypes =
Seq(HoodieRecordType.SPARK)) {
+ (tableName, tablePath, leg) =>
+ assertVariantLayout(tablePath, shredded = true, s"$leg setup", column =
"s.inner")
+
+ withSQLConf("hoodie.schema.on.read.enable" -> "true") {
+ spark.sql(s"alter table $tableName add columns (note string)")
+ withSQLConf("spark.sql.variant.pushVariantIntoScan" -> "false") {
+ checkNestedExceptionContains(
+ () => spark.sql(s"select id, cast(s.inner as string), note from
$tableName").collect())(
+ "shredded variant")
+ }
+ }
+ }
+ }
+
+ test("Inline compaction and clustering under schema-on-read fail fast on the
variant column") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // Hudi's own base-file reads request the same full-variant shape as
Spark's PushVariantIntoScan
+ // rewrite (SparkFileFormatInternalRowReaderContext), and a table
service's reader context
+ // carries the table's internal schema once one is committed
(SparkReaderContextFactory puts
+ // the table path and the valid commits on the conf), so an inline service
under a schema-on-read
+ // write reaches the guard's rewrite arm on a variant column that was
never shredded. It cannot be
+ // served until #18285 - the merged internal-schema request comes back as
{metadata, value} where
+ // the restore projection expects the ordinal struct - and before the
guard the same read died
+ // inside pruning ("cannot prune col: v.0"). What is pinned here is that
the failure names this
+ // route rather than a Spark conf the service never set. Upserts are
unaffected (the merge
+ // handle's base-file read never enters the reader's schema-on-read
branch), and so are
+ // run_compaction / run_clustering, whose clients carry no internal schema.
+ def writeThroughDataFrame(tableName: String, tablePath: String, tableType:
String, id: Int,
+ serviceOptions: (String, String)*): Unit = {
+ var writer = spark.sql(s"""select $id as id, parse_json('{"a":$id}') as
v, 2000L as ts, cast(null as string) as note""")
+ .write.format("hudi")
+ .option("hoodie.table.name", tableName)
+ .option("hoodie.datasource.write.recordkey.field", "id")
+ .option("hoodie.datasource.write.precombine.field", "ts")
+ .option("hoodie.datasource.write.operation", "upsert")
+ .option("hoodie.datasource.write.table.type", tableType)
+ .option("hoodie.schema.on.read.enable", "true")
+ serviceOptions.foreach { case (key, value) => writer =
writer.option(key, value) }
+ writer.mode("append").save(tablePath)
+ }
+ def seedWithCommittedInternalSchema(tableName: String): Unit = {
+ withSQLConf("hoodie.schema.on.read.enable" -> "true") {
+ spark.sql(s"""insert into $tableName values (1, parse_json('{"a":1}'),
1000)""")
+ // The schema-on-read DDL is what commits the internal schema; the
insert alone does not.
+ spark.sql(s"alter table $tableName add columns (note string)")
+ }
+ }
+
+ withVariantTable("inline clustering under schema-on-read", "cow",
recordTypes = Seq(HoodieRecordType.SPARK)) {
Review Comment:
With the default small-file limit the upsert of a new key on this one-row
COW table bin-packs into the seed file, so nothing here separates the inline
clustering read from the write's own base-file read. Asserting a completed
replacecommit, or running the same write without the clustering options and
expecting it to pass, would pin the route.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
