wombatu-kun commented on code in PR #19687:
URL: https://github.com/apache/hudi/pull/19687#discussion_r3877406810
##########
hudi-common/src/main/java/org/apache/hudi/common/util/SortUtils.java:
##########
@@ -21,14 +21,127 @@
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.common.util.collection.Pair;
+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. Spark's
RowOrdering.isOrderable
+ * is false for both VARIANT and MAP, which is the binding constraint on the
row path. On the
+ * Avro path only MAP is outright uncomparable (GenericData.compare throws
"Can't compare
+ * maps!"); a variant's {metadata, value} record does compare, but by its
bytes, which is never
+ * a meaningful sort key. The walk recurses through records and array
elements just as
+ * isOrderable does, so a struct or an array that merely holds a variant or
a map at depth is
+ * rejected too, and the error names the nested member that made the column
unorderable. 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) {
+ String columnName = sortColumn.trim();
+ HoodieSchemaField field =
fieldsByLowerName.get(columnName.toLowerCase(Locale.ROOT));
+ if (field == null) {
+ continue;
Review Comment:
`validateSortableColumns` looks its names up among the top-level fields and
continues on a miss, so `hoodie.clustering.plan.strategy.sort.columns =
's.tags'` reaches `RowCustomColumnsSortPartitioner` unchecked and inline
clustering still dies on Spark's AnalysisException, while `run_clustering(order
=> 's.tags')` is caught up front. Is that split intended, or should
`resolveOrderColumn`'s segment walk move into `SortUtils` so every sink
resolves dotted paths?
##########
hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HiveHoodieReaderContext.java:
##########
@@ -148,6 +154,62 @@ 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.
+ // The flagged columns are split by Hive's read column names on the outer
conf, since the
+ // per-file copy below gets requiredSchema's names from setSchemas and
cannot tell the two
+ // apart: a column Hive selected fails as Hive-visible nulls; a column
only requiredSchema
+ // names is there for merging (a CUSTOM merge whose merger is not
projection compatible reads
+ // the whole table schema; a merger can also list it as mandatory) and
fails too, because the
+ // reader materializes it at {metadata, value} and the merger would
consume the nulls. Hive
+ // writes the full name list for `select *` and none for count(*), whose
requested schema is
+ // then empty
(HoodieFileGroupReaderBasedRecordReader.createRequestedSchema), so nothing is
+ // flagged unless merging widens it. A read whose nested column paths
+ // (hive.io.file.readNestedColumn.paths) all miss the shredded group is
not flagged either:
+ // 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);
+ Configuration conf = storage.getConf().unwrapAs(Configuration.class);
+ Set<String> requestedColumns =
Arrays.stream(HoodieColumnProjectionUtils.getReadColumnNames(conf))
+ .map(name -> name.trim().toLowerCase(Locale.ROOT))
+ .collect(Collectors.toSet());
+ List<String> shreddedColumns =
HoodieColumnProjectionUtils.columnsReadingShreddedPaths(conf, shreddedPaths);
+ Map<Boolean, List<String>> byHiveRequest = shreddedColumns.stream()
+ .collect(Collectors.partitioningBy(requestedColumns::contains));
+ List<String> hiveReads = byHiveRequest.get(true);
+ List<String> mergeOnly = byHiveRequest.get(false);
+ if (!hiveReads.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(", ", hiveReads), filePath));
+ }
+ if (!mergeOnly.isEmpty()) {
Review Comment:
Under skip-merge the file group reader builds an
`UnmergedFileGroupRecordBuffer`, whose `processNextDataRecord` is a no-op, so
no merger consumes the column - but `generateRequiredSchema` still widens to
the whole table schema on a CUSTOM merge and this arm fails a `select id` that
returns correct rows today. Gating the merge-only throw on the resolved merge
type (`hoodie.datasource.merge.type` and `hoodie.realtime.merge.skip` are both
on the conf this already reads) would keep it working.
##########
hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestHiveHoodieReaderContext.java:
##########
@@ -160,4 +180,179 @@ private static HoodieSchema getBaseSchema() {
private ArrayWritable createBaseRecord(Writable[] values) {
return new ArrayWritable(Writable.class, values);
}
+
+ @Test
+ void getFileRecordIteratorFailsFastOnShreddedVariantColumn(@TempDir
java.nio.file.Path tempDir) throws Exception {
+ // The Hive reader hands base files to a plain parquet-avro read at the
requested
+ // {metadata, value} projection; a shredded file would come back with
silent nulls (the
+ // payload of typed rows lives in typed_value, which the projection
drops). The footer is
+ // already read for schema pruning, so the shredded shape must fail fast
instead.
+ StoragePath filePath =
InputFormatTestUtil.writeVariantParquetFile(tempDir, "shredded.parquet", true);
+ HoodieSchema tableSchema = tableSchemaWithVariant();
+ HiveHoodieReaderContext readerContext = newReaderContext();
+ HoodieStorage storage = HoodieStorageUtils.getStorage(filePath,
storageConfiguration);
+ requestColumns("id", "v");
+
+ HoodieException failure = assertThrows(HoodieException.class, () ->
+ readerContext.getFileRecordIterator(filePath, 0, Long.MAX_VALUE,
tableSchema, tableSchema, storage));
+ assertTrue(failure.getMessage().contains("shredded variant") &&
failure.getMessage().contains("'v'"),
+ "The error must name the shredded variant column, got: " +
failure.getMessage());
+
+ // A query that does not project the variant column (`select id`) stays
readable.
+ HoodieSchema withoutVariant = HoodieSchema.createRecord("TestRecord",
null, null, Collections.singletonList(
+ HoodieSchemaField.of("id",
HoodieSchema.create(HoodieSchemaType.INT))));
+ requestColumns("id");
+ when(readerCreator.getRecordReader(any(), any(), any()))
+ .thenReturn((RecordReader<NullWritable, ArrayWritable>)
mock(RecordReader.class));
+ assertDoesNotThrow(() ->
+ readerContext.getFileRecordIterator(filePath, 0, Long.MAX_VALUE,
tableSchema, withoutVariant, storage));
+ }
+
+ @Test
+ void getFileRecordIteratorFailsOnShreddedVariantReadForMerging(@TempDir
java.nio.file.Path tempDir) throws Exception {
+ // The required schema can be wider than the query: a CUSTOM merge whose
merger is not
+ // projection compatible reads the whole table schema for merging, so
`select id` reaches the
+ // context asking for the variant column too. Hive's read column names
split the flagged
+ // columns: the ones Hive selected fail as Hive-visible nulls, the ones
only merging needs fail
+ // as well, since setSchemas materializes them at {metadata, value} for
the merger. count(*)
+ // names no column and its requested schema is empty, so nothing is
flagged and it reads.
+ StoragePath filePath =
InputFormatTestUtil.writeVariantParquetFile(tempDir, "shredded.parquet", true);
+ HoodieSchema tableSchema = tableSchemaWithVariant();
+ HiveHoodieReaderContext readerContext = newReaderContext();
+ HoodieStorage storage = HoodieStorageUtils.getStorage(filePath,
storageConfiguration);
+ when(readerCreator.getRecordReader(any(), any(), any()))
+ .thenReturn((RecordReader<NullWritable, ArrayWritable>)
mock(RecordReader.class));
+
+ requestColumns("id", "v");
+ HoodieException hiveVisible = assertThrows(HoodieException.class, () ->
+ readerContext.getFileRecordIterator(filePath, 0, Long.MAX_VALUE,
tableSchema, tableSchema, storage));
+ assertTrue(hiveVisible.getMessage().contains("'v'") &&
!hiveVisible.getMessage().contains("for merging"),
+ "select * must fail as a Hive-visible read of the shredded variant
column, got: " + hiveVisible.getMessage());
+
+ requestColumns("id");
+ HoodieException mergeOnly = assertThrows(HoodieException.class, () ->
+ readerContext.getFileRecordIterator(filePath, 0, Long.MAX_VALUE,
tableSchema, tableSchema, storage));
+ assertTrue(mergeOnly.getMessage().contains("'v'") &&
mergeOnly.getMessage().contains("for merging"),
+ "A variant only the merge reads must fail as a merge read, got: " +
mergeOnly.getMessage());
+ verify(readerCreator, never()).getRecordReader(any(), any(), any());
+
+ // count(*): Hive names no column, and createRequestedSchema turns that
into an empty record.
+ requestColumns();
Review Comment:
`toShreddedReadSchema` returns its argument by identity for a zero-field
record, so this leg exits at the `shreddedReadSchema != requiredSchema` gate
and never reads the Hive column names - `requestColumns("id", "v")` here would
leave it green. Either drop it, since
`getFileRecordIteratorFailsFastOnShreddedVariantColumn` already pins that
identity short-circuit, or give it a `requiredSchema` that still carries the
variant so the empty name list lands in the merge-only bucket.
--
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]
