wombatu-kun commented on code in PR #19687:
URL: https://github.com/apache/hudi/pull/19687#discussion_r3859744104
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringOperator.java:
##########
@@ -353,8 +354,9 @@ private BinaryExternalSorter initSorter() {
}
private SortOperatorGen createSortOperatorGen() {
+ // Trim: the config list is user-written ("id, name"), and the column
names are looked up as given.
return new SortOperatorGen(rowType,
- conf.get(FlinkOptions.CLUSTERING_SORT_COLUMNS).split(","));
+
Arrays.stream(conf.get(FlinkOptions.CLUSTERING_SORT_COLUMNS).split(",")).map(String::trim).toArray(String[]::new));
Review Comment:
`ClusteringOperator` trims the sort columns but never calls
`SortUtils.validateSortableColumns`, so `clustering.sort.columns` naming a MAP
column still reaches `SortOperatorGen`, whose generated comparator throws
"Unsupported sort field value type" per record inside the sorter. Adding the
same check here would make the error name the column on the Flink entry point
too, or the PR summary could scope the fix to the Spark and Java clients.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunClusteringProcedure.scala:
##########
@@ -120,14 +120,17 @@ class RunClusteringProcedure extends BaseProcedure
logInfo(s"Partition selected: $selectedPartitions")
}
- // Construct sort column info
+ // Construct sort column info. Normalise once so the plan stores the same
trimmed list the
+ // strategies and partitioners work from: the procedure validates the
argument up front, and
+ // the stored value is what later services see.
orderColumns match {
case Some(o) =>
- validateOrderColumns(o.asInstanceOf[String], metaClient)
+ val normalized =
o.asInstanceOf[String].split(",").map(_.trim).mkString(",")
+ validateOrderColumns(normalized, metaClient)
confs = confs ++ Map(
- HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS.key() ->
o.asInstanceOf[String]
+ HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS.key() -> normalized
Review Comment:
The `options` argument is merged into `confs` after this, so
`run_clustering(options => 'hoodie.clustering.plan.strategy.sort.columns=...')`
overwrites the normalised value and reaches the plan without
`validateOrderColumns` or `SortUtils.validateSortableColumns`. Merging
`options` before the order-column block, or validating once after all three
merges, would make the comment above hold for both routes.
##########
hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java:
##########
@@ -233,4 +268,264 @@ 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_METADATA)) {
+ 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 a {@code _tuple} suffix (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) ?
repeated.asGroupType().getType(0) : repeated;
+ collectShreddedVariantPaths(element, path, shreddedPaths);
+ return;
+ }
+ if (originalType == OriginalType.MAP && group.getFieldCount() == 1 &&
!group.getType(0).isPrimitive()) {
+ for (Type entryMember : group.getType(0).asGroupType().getFields()) {
+ collectShreddedVariantPaths(entryMember, path, shreddedPaths);
+ }
+ return;
+ }
+ for (Type field : group.getFields()) {
+ collectShreddedVariantPaths(field, path + "." +
field.getName().toLowerCase(Locale.ROOT), shreddedPaths);
+ }
+ }
+
+ /**
+ * Whether {@code repeated}, the single child of a LIST group, is the
3-level layout's synthetic
+ * level rather than the element itself. This is parquet's
backward-compatibility rule (the one
+ * AvroSchemaConverter applies as isElementType, negated): a repeated level
that is a group with
+ * exactly one non-repeated field and a name that is not one of the legacy
element names is the
+ * synthetic level; anything else is a 2-level layout's element, whose
members are user fields.
+ */
+ private static boolean isSyntheticListLevel(Type repeated) {
+ if (repeated.isPrimitive()) {
+ return false;
+ }
+ GroupType group = repeated.asGroupType();
+ String name = repeated.getName().toLowerCase(Locale.ROOT);
+ return group.getFieldCount() == 1
+ && !group.getType(0).isRepetition(Type.Repetition.REPEATED)
+ && !"array".equals(name)
+ && !name.endsWith("_tuple");
Review Comment:
`isSyntheticListLevel` treats any repeated level ending in `_tuple` as the
element, but `AvroSchemaConverterWithTimestampNTZ.isElementType`, which the
javadoc says this mirrors, compares against `parentName + "_tuple"` - so a
shredded group under a differently-named tuple level is collected one segment
too deep and the column is never flagged. Pass the LIST group's name down and
compare it the way `ParquetSchemaEvolutionUtils.parquetListElement` already
does.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala:
##########
@@ -203,4 +212,130 @@ 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 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 =>
Review Comment:
`SparkFileFormatInternalRowReaderContext` asks
`SparkAdapter.buildFullVariantReadSchema` for the same marker-tagged
`struct<"0": variant>` on internal reads that have no catalyst schema, so on a
schema-on-read table this arm fires during compaction, clustering and CDC even
when nothing is shredded, blaming `spark.sql.variant.pushVariantIntoScan` which
was never set. Should the arm skip the shape Hudi synthesizes itself, or is
failing those services intended?
--
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]
