wombatu-kun commented on code in PR #19687:
URL: https://github.com/apache/hudi/pull/19687#discussion_r3849281835
##########
hudi-common/src/main/java/org/apache/hudi/common/util/SortUtils.java:
##########
@@ -21,14 +21,72 @@
import org.apache.hudi.common.avro.HoodieAvroUtils;
import org.apache.hudi.common.model.HoodieRecord;
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.common.util.collection.FlatLists;
+import org.apache.hudi.exception.HoodieException;
+import java.util.Locale;
+import java.util.Map;
import java.util.function.Function;
+import java.util.stream.Collectors;
/**
* Utility functions used by BULK_INSERT practitioners while sorting records.
*/
public class SortUtils {
+
+ /**
+ * Rejects sort columns whose type cannot serve as a sort key. VARIANT and
MAP have no total
+ * order (Spark's RowOrdering.isOrderable is false for both), and BLOB and
VECTOR are rejected
+ * deliberately: their struct/array encodings would compare by raw bytes or
elements, which is
+ * never a meaningful sort key, and the record-based write path fails on
them outright. Without
+ * this check the failure surfaces deep in the write job (an
AnalysisException from the row
+ * partitioner, a ClassCastException from the record-based one) without
naming the column.
+ *
+ * <p>Matching is case-insensitive, mirroring Spark's column resolution.
Names absent from the
+ * schema (nested paths, meta columns on a data-only schema) are left for
the caller to handle.
+ *
+ * @param sortColumns the configured sort columns, may be null or empty
+ * @param schema schema of the data, with or without metadata fields
+ */
+ public static void validateSortableColumns(String[] sortColumns,
HoodieSchema schema) {
+ if (sortColumns == null || sortColumns.length == 0
+ || schema == null || schema.getType() != HoodieSchemaType.RECORD) {
+ return;
+ }
+ Map<String, HoodieSchemaField> fieldsByLowerName =
schema.getFields().stream()
+ .collect(Collectors.toMap(field ->
field.name().toLowerCase(Locale.ROOT), Function.identity(), (first, second) ->
first));
+ for (String sortColumn : sortColumns) {
+ HoodieSchemaField field =
fieldsByLowerName.get(sortColumn.trim().toLowerCase(Locale.ROOT));
+ if (field != null) {
+ HoodieSchemaType type = field.schema().getNonNullType().getType();
Review Comment:
This inspects only the top-level type, but `RowOrdering.isOrderable`
recurses, so a sort column of `struct<m: map<...>>` or `array<map<...>>` still
reaches the AnalysisException this check exists to replace. Either recurse the
walk the same way, or narrow the javadoc to say the check is top-level only.
##########
hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java:
##########
@@ -233,4 +249,77 @@ 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; when it does run it mirrors the per-file
readSchema the
+ * file-group-reader path already performs. That footer-derived schema
carries no variant
+ * logical type (the converter turns variant groups into plain records), so
the file side is
+ * matched by shape, anchored on the Hive type of the requested column.
+ */
+ @VisibleForTesting
+ static void validateNoShreddedVariantRead(InputSplit split, JobConf job) {
+ if (!(split instanceof FileSplit)) {
+ return;
+ }
+ Path filePath = ((FileSplit) split).getPath();
+ if
(!filePath.getName().endsWith(HoodieFileFormat.PARQUET.getFileExtension())) {
+ return;
+ }
+ Set<String> requestedColumns =
Arrays.stream(HoodieColumnProjectionUtils.getReadColumnNames(job))
+ .map(name -> name.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toSet());
+ if (requestedColumns.isEmpty()) {
+ // count(*)-style read: no column data is materialized
+ return;
+ }
+ List<String> ioColumns = HoodieColumnProjectionUtils.getIOColumns(job);
+ List<String> ioColumnTypes =
HoodieColumnProjectionUtils.getIOColumnTypes(job);
+ 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 requested columns whose synced Hive type embeds the variant
{metadata, value} shape:
+ // the anchor for the shape match on the file side below.
+ Set<String> variantColumns = new HashSet<>();
+ for (int i = 0; i < ioColumns.size(); i++) {
+ String name = ioColumns.get(i).toLowerCase(Locale.ROOT);
+ String type = ioColumnTypes.get(i).toLowerCase(Locale.ROOT);
+ if (requestedColumns.contains(name) && type.contains("metadata:binary")
&& type.contains("value:binary")) {
+ variantColumns.add(name);
+ }
+ }
+ if (variantColumns.isEmpty()) {
+ return;
+ }
+ StoragePath storagePath = convertToStoragePath(filePath);
+ HoodieStorage storage = HoodieStorageUtils.getStorage(storagePath,
HadoopFSUtils.getStorageConf(job));
+ HoodieSchema fileSchema =
HoodieIOFactory.getIOFactory(storage).getFileFormatUtils(storagePath).readSchema(storage,
storagePath);
Review Comment:
`readSchema` converts the whole footer to Avro, and `convertINT96` throws
unless `parquet.avro.readInt96AsFixed` is set, which nothing in Hudi sets, so a
single INT96 column anywhere in the file fails a read this method otherwise
takes care to leave working. Reading the raw `MessageType` and checking the
requested column's group for `typed_value` avoids both, the way the
schema-on-read guard already does.
##########
hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40HoodieParquetReadSupport.scala:
##########
@@ -92,6 +93,17 @@ object Spark40HoodieParquetReadSupport {
private def reorderVariantType(t: Type): Type = {
t match {
case group: GroupType if isVariantGroup(group) =>
+ if (group.containsField("typed_value")) {
Review Comment:
`reorderVariantFields` only enters `reorderVariantType` for top-level fields
named in `variantFieldNames`, so a variant nested in a struct is never reached
and this throw cannot fire for it - the one shape the Hive, legacy-input-format
and schema-on-read guards all recurse for. The new test misses that because it
calls `reorderVariantFields` without a catalyst schema, the `null` branch no
production caller takes.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunClusteringProcedure.scala:
##########
@@ -233,13 +233,17 @@ class RunClusteringProcedure extends BaseProcedure
}
val tableSchemaResolver = new TableSchemaResolver(metaClient)
- val fields = tableSchemaResolver.getTableSchema(false)
- .getFields.asScala.map(_.name().toLowerCase)
- orderColumns.split(",").foreach(col => {
- if (!fields.contains(col.toLowerCase)) {
+ val tableSchema = tableSchemaResolver.getTableSchema(false)
+ val fields = tableSchema.getFields.asScala.map(_.name().toLowerCase)
+ val columns = orderColumns.split(",")
+ columns.foreach(col => {
+ if (!fields.contains(col.trim.toLowerCase)) {
Review Comment:
The plan stores the raw `orderColumns` string and
`MultipleSparkJobExecutionStrategy` splits it without trimming, so `order =>
'a, b'` now clears this check and fails inside the job instead of here.
Normalising once and writing the trimmed list back into
`PLAN_STRATEGY_SORT_COLUMNS` would keep both sides consistent.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestClusteringProcedure.scala:
##########
@@ -867,6 +867,47 @@ class TestClusteringProcedure extends
HoodieSparkProcedureTestBase {
}
}
+ test("Test Call run_clustering rejects unsortable order columns") {
+ // Not gated on any Spark version: BLOB, VECTOR and MAP exist on every
supported Spark, so
+ // this covers the sort-column validation on the lanes where the variant
suite is skipped.
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val basePath = s"${tmp.getCanonicalPath}/$tableName"
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | content blob,
+ | embedding vector(4),
+ | attrs map<string, string>,
+ | ts long
+ |) using hudi
+ | options (
+ | primaryKey = 'id',
+ | orderingFields = 'ts'
+ | )
+ | location '$basePath'
+ """.stripMargin)
+ spark.sql(s"insert into $tableName values (1, 'a1', null, null, null,
1000)")
+
+ // The procedure validates the order columns up front, before any plan
is scheduled.
+ Seq("content", "embedding", "attrs").foreach { col =>
+ checkNestedExceptionContains(s"call run_clustering(table =>
'$tableName', order => '$col')")(
+ s"Sorting by column '$col'")
+ }
+ // Case-insensitive, mirroring Spark's column resolution.
+ checkNestedExceptionContains(s"call run_clustering(table =>
'$tableName', order => 'CONTENT')")(
+ "Sorting by column 'CONTENT'")
+ // The execution-time twin: configured plan-strategy sort columns skip
the procedure
+ // check and are rejected by the execution strategy and partitioner
constructors instead
+ // (SortUtils.validateSortableColumns).
+ checkNestedExceptionContains(
+ s"call run_clustering(table => '$tableName', options =>
'hoodie.clustering.plan.strategy.sort.columns=content')")(
Review Comment:
`getPartitioner` validates before it constructs the partitioner, so this leg
never reaches the new constructor checks and they could all be reverted with
the suite green. The one path only they cover is a user-defined
`RowCustomColumnsSortPartitioner`/`RDDCustomColumnsSortPartitioner` driven by
`hoodie.bulkinsert.user.defined.sort.columns` - worth one case there.
##########
hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java:
##########
@@ -151,7 +163,11 @@ public RecordReader<NullWritable, ArrayWritable>
getRecordReader(final InputSpli
// ParquetInputFormat.setFilterPredicate(job, predicate);
// clearOutExistingPredicate(job);
// }
+ // From here on the split is read by Hive's plain parquet reader, which
the file-group-reader
+ // guard in HiveHoodieReaderContext never sees; repeat the
shredded-variant fail-fast for it.
+ validateNoShreddedVariantRead(split, job);
Review Comment:
For a `BootstrapBaseFileSplit` this checks the skeleton file, which carries
only the meta columns, so it costs a footer open that can never fire; moving
the call below the bootstrap branch would skip it. Nothing in the tests builds
a `BootstrapBaseFileSplit`, so neither bootstrap leg of the guard runs today.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala:
##########
@@ -509,6 +407,16 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
spark.sql(s"""update $tableName set v =
parse_json('{"key":"value2"}'), ts = 1001 where id = 1""")
+ // Second update under the FLIPPED layout (session confs override
tblproperties):
+ // the updated file's layout now differs from the write layout, the
mixed-layout
+ // CDC case (images must survive whichever physical slot serves
them).
+ withSQLConf(
+ "hoodie.parquet.variant.write.shredding.enabled" ->
(!shredded).toString,
Review Comment:
The first insert is layout-pinned but the flipped update is not, so if the
session confs do not reach the update writer this leg silently becomes a second
copy of the other one. `assertLayoutsByInstant(baseLayouts(tablePath), leg)` on
the flipped instant would pin it.
##########
hudi-common/src/main/java/org/apache/hudi/common/util/SortUtils.java:
##########
@@ -21,14 +21,72 @@
import org.apache.hudi.common.avro.HoodieAvroUtils;
import org.apache.hudi.common.model.HoodieRecord;
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.common.util.collection.FlatLists;
+import org.apache.hudi.exception.HoodieException;
+import java.util.Locale;
+import java.util.Map;
import java.util.function.Function;
+import java.util.stream.Collectors;
/**
* Utility functions used by BULK_INSERT practitioners while sorting records.
*/
public class SortUtils {
+
+ /**
+ * Rejects sort columns whose type cannot serve as a sort key. VARIANT and
MAP have no total
+ * order (Spark's RowOrdering.isOrderable is false for both), and BLOB and
VECTOR are rejected
+ * deliberately: their struct/array encodings would compare by raw bytes or
elements, which is
+ * never a meaningful sort key, and the record-based write path fails on
them outright. Without
+ * this check the failure surfaces deep in the write job (an
AnalysisException from the row
+ * partitioner, a ClassCastException from the record-based one) without
naming the column.
+ *
+ * <p>Matching is case-insensitive, mirroring Spark's column resolution.
Names absent from the
+ * schema (nested paths, meta columns on a data-only schema) are left for
the caller to handle.
+ *
+ * @param sortColumns the configured sort columns, may be null or empty
+ * @param schema schema of the data, with or without metadata fields
+ */
+ public static void validateSortableColumns(String[] sortColumns,
HoodieSchema schema) {
+ if (sortColumns == null || sortColumns.length == 0
+ || schema == null || schema.getType() != HoodieSchemaType.RECORD) {
+ return;
+ }
+ Map<String, HoodieSchemaField> fieldsByLowerName =
schema.getFields().stream()
+ .collect(Collectors.toMap(field ->
field.name().toLowerCase(Locale.ROOT), Function.identity(), (first, second) ->
first));
+ for (String sortColumn : sortColumns) {
+ HoodieSchemaField field =
fieldsByLowerName.get(sortColumn.trim().toLowerCase(Locale.ROOT));
+ if (field != null) {
+ HoodieSchemaType type = field.schema().getNonNullType().getType();
+ if (type == HoodieSchemaType.VARIANT || type == HoodieSchemaType.MAP
Review Comment:
BLOB is a struct of atomics and VECTOR is `array<float>` in Spark, so both
are orderable and `run_clustering(order => '<blob column>')` sorts fine today;
rejecting them is a new restriction on a path this PR is not otherwise
touching. Was that intended, or should the check stay on VARIANT and MAP?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala:
##########
@@ -203,4 +212,125 @@ 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).
+ *
+ * A scan rewritten by Spark's PushVariantIntoScan (4.x) fails fast
regardless of the file's
+ * layout: the merged internal-schema request materializes the variant as
{metadata, value}
+ * while downstream codegen expects the rewrite's 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).
+ *
+ * Shared by [[ParquetSchemaEvolutionUtils.getHadoopConfClone]] and the
per-version legacy
+ * file formats, which carry a copy of the same schema-merge block. Callers
gate on a
+ * non-empty projection: empty-projection queries (count(*), select 1) read
no column data
+ * and must keep working, and the query schema is unpruned in that case.
+ */
+ def validateNoShreddedVariants(requiredSchema: StructType, querySchema:
InternalSchema, footerFileMetaData: FileMetaData): Unit = {
+ findVariantRewritePath(requiredSchema).foreach { path =>
+ throw new HoodieException(String.format(
+ "Column '%s' is a variant projected through Spark's variant rewrite "
+ + "(spark.sql.variant.pushVariantIntoScan) and the table is read
with schema-on-read "
+ + "(hoodie.schema.on.read.enable), which cannot reconstruct variants
(see issue "
+ + "#18285). Read without schema-on-read.", path))
+ }
+ val fileParquetSchema = footerFileMetaData.getSchema
+ querySchema.getRecord.fields().foreach { field =>
+ if (fileParquetSchema.containsField(field.name())) {
+ validateNoShreddedVariant(
+ field.`type`(),
fileParquetSchema.getType(fileParquetSchema.getFieldIndex(field.name())),
field.name())
+ }
+ }
+ }
+
+ /**
+ * The dotted path of the first PushVariantIntoScan rewrite struct in the
schema, if any (see
+ * SparkInternalSchemaConverter.isVariantRewriteStruct for the marker).
+ */
+ private def findVariantRewritePath(dataType: DataType, path: String = ""):
Option[String] = dataType match {
+ case struct: StructType if
SparkInternalSchemaConverter.isVariantRewriteStruct(struct) =>
+ Some(path)
+ case struct: StructType =>
+ struct.fields.foldLeft(Option.empty[String]) { (found, field) =>
+ found.orElse(findVariantRewritePath(field.dataType, concatPath(path,
field.name)))
+ }
+ case array: ArrayType => findVariantRewritePath(array.elementType,
concatPath(path, "element"))
+ case map: MapType => findVariantRewritePath(map.valueType,
concatPath(path, "value"))
+ case _ => None
+ }
+
+ private def concatPath(path: String, name: String): String =
+ if (path.isEmpty) name else path + "." + name
+
+ private def validateNoShreddedVariant(internalType: InternalType,
parquetType: ParquetType, path: String): Unit = {
+ internalType match {
+ // A variant: two fields, both carrying the sentinel negative ids
(BLOB's sentinel record
+ // has three). The parquet side decides shredded-ness.
+ case record: Types.RecordType if record.fields().size() == 2 &&
record.fields().forall(_.fieldId() < 0) =>
+ if (!parquetType.isPrimitive &&
parquetType.asGroupType().containsField("typed_value")) {
+ throw new HoodieException(String.format(
+ "Column '%s' is a shredded variant (typed_value present) and the
table is read "
+ + "with schema-on-read (hoodie.schema.on.read.enable), which
cannot reconstruct "
+ + "shredded variants (see issue #18285). Read without
schema-on-read, or rewrite "
+ + "the table unshredded (e.g. cluster with "
+ + "hoodie.parquet.variant.write.shredding.enabled=false).",
path))
+ }
+ case record: Types.RecordType if !parquetType.isPrimitive =>
+ val group = parquetType.asGroupType()
+ record.fields().foreach { field =>
+ if (group.containsField(field.name())) {
+ validateNoShreddedVariant(field.`type`(),
group.getType(field.name()), path + "." + field.name())
+ }
+ }
+ case array: Types.ArrayType =>
+
parquetListElement(parquetType).foreach(validateNoShreddedVariant(array.elementType(),
_, path + ".element"))
+ case map: Types.MapType =>
+
parquetMapValue(parquetType).foreach(validateNoShreddedVariant(map.valueType(),
_, path + ".value"))
+ case _ =>
+ }
+ }
+
+ /**
+ * Resolves the element type of a parquet LIST group, covering both the
3-level layout the
+ * Spark writer produces (group -> repeated "list" -> element) and the
2-level layout
+ * parquet-avro produces (group -> repeated element). The 3-level test
mirrors Spark's
+ * ParquetSchemaConverter.isElementType. An unrecognized shape returns None,
which stops the
+ * walk without failing the read.
+ */
+ private def parquetListElement(parquetType: ParquetType):
Option[ParquetType] = {
+ if (parquetType.isPrimitive || parquetType.asGroupType().getFieldCount !=
1) {
+ None
+ } else {
+ val repeated = parquetType.asGroupType().getType(0)
+ if (!repeated.isRepetition(ParquetType.Repetition.REPEATED)) {
+ None
+ } else if (!repeated.isPrimitive && repeated.asGroupType().getFieldCount
== 1
+ && repeated.getName != "array" &&
!repeated.getName.endsWith("_tuple")) {
Review Comment:
`isElementType` binds this to the parent name (`parentName + "_tuple"`), so
`endsWith` is more permissive and a 3-level LIST whose repeated group ends in
`_tuple` is walked as the element itself, hiding a shredded variant one level
below. Compare against `parquetType.getName + "_tuple"`, or drop the claim that
this mirrors `isElementType`.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala:
##########
@@ -203,4 +212,125 @@ 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).
+ *
+ * A scan rewritten by Spark's PushVariantIntoScan (4.x) fails fast
regardless of the file's
+ * layout: the merged internal-schema request materializes the variant as
{metadata, value}
+ * while downstream codegen expects the rewrite's 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).
+ *
+ * Shared by [[ParquetSchemaEvolutionUtils.getHadoopConfClone]] and the
per-version legacy
+ * file formats, which carry a copy of the same schema-merge block. Callers
gate on a
+ * non-empty projection: empty-projection queries (count(*), select 1) read
no column data
+ * and must keep working, and the query schema is unpruned in that case.
+ */
+ def validateNoShreddedVariants(requiredSchema: StructType, querySchema:
InternalSchema, footerFileMetaData: FileMetaData): Unit = {
+ findVariantRewritePath(requiredSchema).foreach { path =>
+ throw new HoodieException(String.format(
+ "Column '%s' is a variant projected through Spark's variant rewrite "
+ + "(spark.sql.variant.pushVariantIntoScan) and the table is read
with schema-on-read "
+ + "(hoodie.schema.on.read.enable), which cannot reconstruct variants
(see issue "
+ + "#18285). Read without schema-on-read.", path))
+ }
+ val fileParquetSchema = footerFileMetaData.getSchema
+ querySchema.getRecord.fields().foreach { field =>
+ if (fileParquetSchema.containsField(field.name())) {
Review Comment:
This resolves the footer column by the query-schema name, so after a
schema-on-read rename the file still carries the old name, `containsField` is
false and the column is skipped. Is rename meant to be covered here, or is it
left to #18285 along with real reconstruction?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala:
##########
@@ -0,0 +1,942 @@
+/*
+ * 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, HoodieSparkUtils}
+import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
+import org.apache.hudi.testutils.DataSourceTestUtils
+
+import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
+
+/**
+ * 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).
+ * Legs that need #18961's per-file shredding-schema inference substitute a
forced stand-in
+ * schema via [[inferredOr]] when no inferrer is on the classpath, so the
mixed-layout shape is
+ * preserved on every profile.
+ *
+ * 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)
+
+ withRecordType()(withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ val leg = s"same-file mix, $tableName"
+ createVariantTable(tableName, tablePath, "cow")
+
+ // 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.
+ withRecordType(Seq(HoodieRecordType.SPARK))(withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ val leg = s"mixed-files, $tableName"
+ createVariantTable(tableName, tablePath, "cow", props =
Seq(NEW_FILE_GROUP_PER_COMMIT))
+
+ // Four commits, four layouts, one file each (small.file.limit=0 keeps
every commit in its
+ // own file group). The last commit infers when an inferrer is present;
the forced stand-in
+ // yields the same {c, d} typed_value, so the expectations below hold
either way.
+ 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))),
+ (inferredOr(Forced("c bigint, d boolean")), 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 = spark.read.format("hudi")
+ .option(DataSourceReadOptions.QUERY_TYPE.key,
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
+ .option(DataSourceReadOptions.START_COMMIT.key, "000")
+ .load(tablePath)
+ .selectExpr("id", "cast(v as string)")
+ .orderBy("id")
+ .collect()
+ 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)
+
+ withRecordType()(withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ val leg = s"bin-pack layout flip, $tableName"
+ // 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.
+ createVariantTable(tableName, tablePath, "cow")
+
+ 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)
+
+ withRecordType()(withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ val leg = s"compaction layout split, $tableName"
+ // INMEMORY sends MOR inserts to log files; compaction runs via the
procedure so each run
+ // can happen under its own layout confs.
+ createVariantTable(tableName, tablePath, "mor",
+ props = Seq("hoodie.index.type = 'INMEMORY'", "hoodie.compact.inline =
'false'"))
+ val layout3 = inferredOr(Forced("c bigint, d boolean"))
+
+ 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(layout3) {
+ 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 layout3: reads all three log layouts, writes the
base under layout3.
+ withWriteLayout(layout3) {
+ runCompaction(tableName)
+ }
+ assertResult(false)(DataSourceTestUtils.isLogFileOnly(tablePath))
+ assertCompactionCount(tablePath, 1, leg)
+ val base1 = baseLayouts(tablePath)
+ assert(base1.nonEmpty && base1.forall(_.isShredded),
+ s"[$leg] compacted base must be shredded under $layout3: $base1")
+ if (inferrerPresent) {
+ // 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}"))
+ } else {
+ base1.foreach(l => assert(l.typedFields.toSet == Set("c", "d"),
+ s"[$leg] forced typed_value should carry c, d: ${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)
+ assert(base2.nonEmpty && base2.forall(!_.isShredded),
+ s"[$leg] compaction under Unshredded must write an unshredded base:
$base2")
+ 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 layout3 again, this time reading an
UNSHREDDED base plus a
+ // shredded log.
+ withWriteLayout(layout3) {
+ 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)
+ assert(base3.nonEmpty && base3.forall(_.isShredded),
+ s"[$leg] compaction under $layout3 must write a shredded base again:
$base3")
+
+ 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 = spark.read.format("hudi")
+ .option(DataSourceReadOptions.QUERY_TYPE.key,
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
+ .option(DataSourceReadOptions.START_COMMIT.key, "000")
+ .load(tablePath)
+ .selectExpr("id", "cast(v as string)")
+ .orderBy("id")
+ .collect()
+ 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)
+
+ withRecordType()(withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ val leg = s"table version 9, $tableName"
+ createVariantTable(tableName, tablePath, "mor",
+ props = Seq(
+ "hoodie.write.table.version = '9'",
+ "hoodie.index.type = 'INMEMORY'",
+ "hoodie.compact.inline = 'false'"))
+ val layout = inferredOr(Forced("key string"))
+
+ withWriteLayout(layout) {
+ 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(layout) {
+ runCompaction(tableName)
+ }
+ assertResult(false)(DataSourceTestUtils.isLogFileOnly(tablePath))
+ val base1 = baseLayouts(tablePath)
+ assert(base1.nonEmpty && base1.forall(_.isShredded),
+ s"[$leg] compacted base must be shredded: $base1")
+ 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(layout) {
+ 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(layout) {
+ 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)
+
+ // The row-writer path is record-type independent; the RDD path writes
through the
+ // record-type file writer factories, so it sweeps both.
+ Seq(true, false).foreach { rowWriter =>
+ val recordTypes = if (rowWriter) {
+ Seq(HoodieRecordType.SPARK)
+ } else {
+ Seq(HoodieRecordType.AVRO, HoodieRecordType.SPARK)
+ }
+ Seq(Unshredded, inferredOr(Forced("a bigint"))).foreach { outLayout =>
+ withRecordType(recordTypes)(withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ val leg = s"clustering rowWriter=$rowWriter out=$outLayout,
$tableName"
+ createVariantTable(tableName, tablePath, "cow", props =
Seq(NEW_FILE_GROUP_PER_COMMIT))
+
+ val instants = seedMixedLayoutTable(tableName, tablePath, Seq(
+ (Forced("a bigint, b string"), Seq((0 until 2, ObjA))),
+ (Unshredded, Seq((2 until 4, ObjA))),
+ (inferredOr(Forced("c bigint, d boolean")), 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")
+ outLayout match {
+ case Unshredded =>
+ outFiles.foreach(l => assert(!l.isShredded,
+ s"[$leg] clustering under Unshredded must write unshredded
output: ${l.path}"))
+ case Forced(_) =>
+ outFiles.foreach(l => assert(l.typedFields == Seq("a"),
+ s"[$leg] forced output typed_value should be {a}:
${l.typedFields}"))
+ case Inferred =>
+ // 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("Clustering sort on a variant column is rejected with a clear error") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ withRecordType(Seq(HoodieRecordType.SPARK))(withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ createVariantTable(tableName, tablePath, "cow")
+ withWriteLayout(Forced("a bigint")) {
+ spark.sql(s"""insert into $tableName values (1, parse_json('{"a":1}'),
1000)""")
+ }
+
+ // The procedure's order parameter is validated up front...
+ checkNestedExceptionContains(
+ s"call run_clustering(table => '$tableName', order => 'v')")(
+ "Sorting by column 'v'")
+ // ...and the config-driven sort columns are validated by the execution
strategy and the
+ // partitioner constructors (SortUtils.validateSortableColumns), so the
inline/async paths
+ // get the same error instead of an AnalysisException (row partitioner)
or
+ // ClassCastException (RDD partitioner) deep in the job.
+ checkNestedExceptionContains(
+ s"call run_clustering(table => '$tableName', " +
+ "options => 'hoodie.clustering.plan.strategy.sort.columns=v')")(
+ "Sorting by column 'v'")
+ })
+ }
+
+ 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 = if (rowWriter) {
+ Seq(HoodieRecordType.SPARK)
+ } else {
+ Seq(HoodieRecordType.AVRO, HoodieRecordType.SPARK)
+ }
+ withRecordType(recordTypes)(withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ val leg = s"mor clustering rowWriter=$rowWriter, $tableName"
+ // No INMEMORY index: the first insert creates a base file, the update
goes to a log.
+ createVariantTable(tableName, tablePath, "mor",
+ props = Seq("hoodie.compact.inline = 'false'"))
+ val outLayout = inferredOr(Forced("c bigint"))
+
+ 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.
+ val preBase = baseLayouts(tablePath)
+ assert(preBase.nonEmpty && preBase.forall(_.isShredded),
+ s"[$leg] pre-clustering base must be shredded: $preBase")
+ val preLogs = nativeLogLayouts(tablePath)
+ assert(preLogs.nonEmpty && preLogs.forall(!_.isShredded),
+ s"[$leg] pre-clustering log must be unshredded: $preLogs")
+
+ 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")
+ outFiles.foreach(l => assert(l.isShredded,
+ s"[$leg] clustering under $outLayout 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 =>
+ withRecordType(Seq(HoodieRecordType.SPARK))(withTempDir { tmp =>
+ withSQLConf("spark.sql.variant.pushVariantIntoScan" -> pushIntoScan) {
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ val leg = s"cow pushVariantIntoScan=$pushIntoScan, $tableName"
+ createVariantTable(tableName, tablePath, "cow", props =
Seq(NEW_FILE_GROUP_PER_COMMIT))
+
+ // $.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))),
+ (inferredOr(Forced("c bigint, d boolean")), 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 =>
+ withRecordType(Seq(HoodieRecordType.SPARK))(withTempDir { tmp =>
+ withSQLConf("spark.sql.variant.pushVariantIntoScan" -> pushIntoScan) {
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ val leg = s"mor pushVariantIntoScan=$pushIntoScan, $tableName"
+ createVariantTable(tableName, tablePath, "mor",
+ props = Seq("hoodie.compact.inline = 'false'"))
+
+ 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)
+
+ withRecordType(Seq(HoodieRecordType.SPARK))(withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+ val leg = s"schema-on-read, $tableName"
+ createVariantTable(tableName, tablePath, "cow")
+ 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.
+ // Both legs' messages share "cannot reconstruct"; the real fix is
#18285.
+ checkNestedExceptionContains(
+ () => spark.sql(s"select id, cast(v as string), note from
$tableName").collect())(
+ "cannot reconstruct")
Review Comment:
Both guard messages contain `cannot reconstruct` and this table is shredded,
so this leg stays green with the `PushVariantIntoScan` arm removed entirely.
Asserting `pushVariantIntoScan` would pin it, and an unshredded companion would
cover the case only that arm can catch.
--
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]
