voonhous commented on code in PR #19687:
URL: https://github.com/apache/hudi/pull/19687#discussion_r3863336272
##########
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:
Done in 5efe1ab: the check runs in doClustering, ahead of the readers, and
open() no longer validates. Moving it to open() was not meant to widen the
blast radius: a bad sort column fails the clustering task, which in async mode
becomes a failed commit event and a rollback of the instant, as the per-record
failure did before. testSortClusteringRejectsUnsortableColumnInsideTheTask pins
that open() succeeds and the plan yields a failed event naming the column.
##########
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:
Done in 5efe1ab: SortOperatorGen's constructor now rejects any sort field
whose LogicalType root has no typed branch in compareExpression -- ROW, ARRAY,
MAP, MULTISET, RAW, VARIANT -- by name and type, so BLOB and VECTOR are caught
as well. That replaces the shared SortUtils check on the Flink side, and since
every Flink sort route builds a SortOperatorGen, bulk insert gets the same
guard. testRejectsFieldTheSorterCannotOrder covers ROW, ARRAY and MAP.
##########
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:
Done in 5efe1ab: the flagged columns are intersected with
hive.io.file.readcolumn.names off the same conf, lower-cased, so a CUSTOM
merge's whole-table required schema no longer fails `select id`; empty names
(count(*)) flag nothing, the legacy guard's reading of that conf.
getFileRecordIteratorFlagsOnlyColumnsHiveReads pins `select *` failing, `select
id` reaching the record reader with the whole table required, and count(*)
reading; the existing legs now set the names Hive would.
##########
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:
Done in 5efe1ab: testLegacyReaderGuardSeesThroughMapEntries writes the map
fixture, pins its key_value/value layout on the footer, and expects the guard
to name `m` for the whole column and for the nested path `m`. Without the MAP
arm the walk collects m.key_value.value, which never meets the Hive-side `m`,
so the leg fails.
##########
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:
Done in 5efe1ab: the sentence names both producers separately --
SparkReaderContextFactory for the write-side services (inline compaction and
clustering), and HoodieFileGroupReaderBasedFileFormat.setSchemaEvolutionConfigs
query-side, whose conf CDCFileGroupIterator builds its reader context from.
##########
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:
Done in 5efe1ab: a blank order/options value is stored as "" rather than
dropped, so it stays the top override in createHoodieWriteClient and a sort
column from the session conf or table config cannot slip through unvalidated;
the strategies still read it as no sort. The comment says so, and a test leg
runs order => '' under a session-conf s.tags and checks the plan carries no
sort key.
##########
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:
Done in 5efe1ab: validateOrderColumns resolves against getTableSchema(true)
and resolves dotted paths with HoodieSchema.getNestedField, checking the leaf
under its path through a new SortUtils.validateSortableColumn, so
_hoodie_commit_time and s.level pass on both routes, s.tags (a MAP leaf) is
rejected as "Sorting by column 's.tags' of type MAP ..." and a missing path as
"Order column not exist". Test legs cover all four.
##########
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:
Done in 5efe1ab: a `select count(*)` sits inside the schema-on-read block of
"Schema-on-read reads of shredded variant files fail fast", after the two
failing selects. A COW count(*) takes readBaseFile into the parquet reader's
getHadoopConfClone with the unpruned query schema, so dropping the
requiredSchema.nonEmpty gate fails it on the shredded file. The
TestLegacyParquetReadPath sentence now names that leg.
--
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]
