voonhous commented on code in PR #19687:
URL: https://github.com/apache/hudi/pull/19687#discussion_r3871026049
##########
hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HiveHoodieReaderContext.java:
##########
@@ -148,6 +153,45 @@ 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 stay readable: the flagged columns are checked
against Hive's read
+ // column names, not requiredSchema, which can be wider than the query --
a CUSTOM merge reads
+ // the whole table schema for merging (no merger overrides
isProjectionCompatible), so
+ // `select id` arrives here asking for the variant column too. Hive writes
the full name list
+ // for `select *` and none for count(*). 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);
+ 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> offendingColumns =
HoodieColumnProjectionUtils.columnsReadingShreddedPaths(conf, shreddedPaths)
+ .stream()
+ .filter(requestedColumns::contains)
Review Comment:
Done in 84bea41: the flagged columns are now split by Hive's read column
names instead of filtered by them. A column Hive selected fails as before; a
column only `requiredSchema` carries fails too, naming the merger, since
`setSchemas` materializes it at `{metadata, value}` for merging. That bucket
can only be the merger's doing: `createRequestedSchema` builds an empty record
when Hive names nothing, so `count(*)` still reads, and `select id` on a file
group without log files (no widening) still reads.
`getFileRecordIteratorFailsOnShreddedVariantReadForMerging` pins `select *`,
`select id` under a whole-table required schema (fails, nothing reaches the
record reader) and `count(*)` with the empty requested schema (reads).
--
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]
