hudi-agent commented on code in PR #19687:
URL: https://github.com/apache/hudi/pull/19687#discussion_r3834798027
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunClusteringProcedure.scala:
##########
@@ -233,13 +233,16 @@ class RunClusteringProcedure extends BaseProcedure
}
val tableSchemaResolver = new TableSchemaResolver(metaClient)
- val fields = tableSchemaResolver.getTableSchema(false)
- .getFields.asScala.map(_.name().toLowerCase)
+ val tableSchema = tableSchemaResolver.getTableSchema(false)
+ val fields = tableSchema.getFields.asScala.map(_.name().toLowerCase)
orderColumns.split(",").foreach(col => {
- if (!fields.contains(col.toLowerCase)) {
+ if (!fields.contains(col.trim.toLowerCase)) {
throw new HoodieClusteringException("Order column not exist:" + col)
}
})
+ // The same validation the partitioners apply at execution time (see
Review Comment:
🤖 nit: `orderColumns.split(",")` is computed twice — once on line 236 for
the `foreach` existence check and again here for `validateSortableColumns`.
Could you extract it to a `val` above the loop so the split is done once and
both call sites share the same array?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/utils/SparkInternalSchemaConverter.java:
##########
@@ -82,6 +82,9 @@ public class SparkInternalSchemaConverter {
public static final String HOODIE_QUERY_SCHEMA =
"hoodie.schema.internal.querySchema";
public static final String HOODIE_TABLE_PATH = "hoodie.tablePath";
public static final String HOODIE_VALID_COMMITS_LIST =
"hoodie.valid.commits.list";
+ // Mirrors Spark's VariantMetadata.METADATA_KEY (Spark 4.x only), referenced
by literal
+ // because the class does not exist on Spark 3 classpaths.
+ private static final String SPARK_VARIANT_METADATA_KEY =
"__VARIANT_METADATA_KEY";
Review Comment:
🤖 nit: `SPARK_VARIANT_METADATA_KEY` is `private` here, which forces
`ParquetSchemaEvolutionUtils.scala` to define its own independent copy of the
same string literal. If the key ever changes in Spark it would need updating in
two places — could you move this to a shared (at least package-visible)
constant so there's a single source of truth?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java:
##########
@@ -233,4 +247,82 @@ 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.
+ */
+ private 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;
+ }
+ boolean requestsVariantShapedColumn = false;
+ for (int i = 0; i < ioColumns.size(); i++) {
+ if
(requestedColumns.contains(ioColumns.get(i).toLowerCase(Locale.ROOT))) {
+ String type = ioColumnTypes.get(i).toLowerCase(Locale.ROOT);
+ if (type.contains("metadata:binary") && type.contains("value:binary"))
{
+ requestsVariantShapedColumn = true;
+ break;
+ }
+ }
+ }
+ if (!requestsVariantShapedColumn) {
+ return;
+ }
+ StoragePath storagePath = convertToStoragePath(filePath);
+ HoodieStorage storage = HoodieStorageUtils.getStorage(storagePath,
HadoopFSUtils.getStorageConf(job));
+ HoodieSchema fileSchema =
HoodieIOFactory.getIOFactory(storage).getFileFormatUtils(storagePath).readSchema(storage,
storagePath);
+ if (fileSchema.getType() != HoodieSchemaType.RECORD) {
+ return;
+ }
+ HoodieSchema strippedSchema =
VariantSchemaUtils.stripVariantShredding(fileSchema);
Review Comment:
🤖 `stripVariantShredding` only strips nodes typed VARIANT (via
`stripVariantShreddingAt` -> `case VARIANT` -> `isShredded()`), but the
`readSchema` here goes through
`AvroSchemaConverterWithTimestampNTZ.convert(MessageType)`, which doesn't
recognize the parquet `VariantLogicalTypeAnnotation` and falls back to
plain-record conversion — so a shredded variant comes back as a plain
`{metadata, value, typed_value}` record with no VARIANT node.
`stripVariantShredding` never matches it, `strippedSchema == fileSchema` is
always true, and this guard never throws, leaving the legacy Hive path silently
nulling shredded columns. The FGR guard in `HiveHoodieReaderContext` avoids
this by anchoring on the requested variant via
`toShreddedReadSchema`/`isShreddedVariantShape`. Could you switch to
shape-based detection here (the variant group already exposes `typed_value` in
the footer schema)? @yihua
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
