yihua opened a new pull request, #831:
URL: https://github.com/apache/hudi-rs/pull/831
## Description
closes #830
A file slice is read by one reader today. Two things want a slice cut along
its base file's row groups:
- An engine that plans splits by byte range (Spark with Hudi 1.3 reports
copy-on-write base files as splittable, and drives Velox through the C++
bridge) hands hudi-rs a slice plus a split's start and length. The bridge read
the whole slice, so a file cut into N splits came back N times, and the caller
had to plan one split per file.
- A DataFusion query over a table with fewer slices than the session has
partitions leaves cores idle: one 1 GB merge-on-read slice with two log files
merged in about 2.8 s on one core while the plain parquet scan of its base file
finished in 14 ms on fourteen.
A `BaseFileRange` names a run of row groups and the base-row positions they
hold. `FileGroupReader::base_file_ranges` cuts a slice into ranges of about
equal bytes from the footer, and `read_file_slice_range_stream` merges one
range: the engine decodes only those row groups, and the log scan keeps only
the records whose `RECORD_POSITIONS` fall inside them, so the ranges of a slice
together return exactly what the whole read returns. Once every block's header
has been read, a scan that knows all blocks carry positions steps over the Avro
records of other ranges by their length prefix without decoding them. When some
block carries no positions the merge falls back to keys, which needs every
record, so the first range reads the whole slice and the other ranges return
nothing.
For the byte-range caller, `read_file_slice_from_paths_split` (and its
stream twin, exposed through the cxx bridge as
`read_file_slice_from_paths_split`) selects the row groups whose midpoint falls
in `[start, start + length)`, the rule Spark's and Velox's parquet readers
apply, so the splits of one file select every row group exactly once. A slice
with log files merges across the whole base file, so it accepts only a split
covering the whole file and fails loudly otherwise, as does a base file without
row groups; a split holding no midpoint returns an empty batch with the slice's
schema.
The DataFusion scan splits slices only when there are fewer than the target
partitions. It follows the session's
`datafusion.optimizer.repartition_file_scans` (off turns cutting off) and
applies `repartition_file_min_size` as a floor on each range's bytes
(DataFusion applies it to the whole scan), weighs a slice by its base file
alone, and cuts a slice with log files only when
`hoodie.merge.use.record.positions` is on (or the read is read-optimized); a
slice without log files is cut regardless. Under file group reader version 1
the first range reads the slice whole and the others return nothing. On a 1 GB
slice with two log files of 2M positioned updates spread over its 7.3M rows,
`COUNT(*)` through DataFusion on fourteen partitions went from 4.3 s to 0.83 s
with the same rows as the single-reader read. Ranges are cut by base-file
bytes, so a table whose updates concentrate in a few row groups still waits for
the range that holds them; Java reads a file group in one task and has t
he same limit.
Two things worth a reviewer's eye. First, this changes the default plan for
every copy-on-write table with fewer slices than `target_partitions`: such
slices are now read in row-group ranges, which costs a footer read per slice at
plan time and one per range at read time, in exchange for using the idle
partitions. Second, the diff is well over the 1,000-line guideline: about half
of it is tests, including a writer in the test crate for a merge-on-read slice
with many row groups and positioned log blocks, since no shipped fixture has
either.
## How are the changes test-covered
- [ ] N/A
- [x] Automated tests (unit and/or integration tests)
- [ ] Manual tests
- [ ] Details are described below
Ranges of a slice with two positioned log files (updates and deletes) union
to the whole read with no row returned twice and each range moving fewer bytes;
a block without positions, or positions turned off, hands the whole slice to
the first range and nothing to the others; a slice without log files is read a
range of row groups at a time; the row-group splitter is covered on uneven,
tiny and empty groups; the decode-time skip is off when any block lacks
positions or is an HFile block; through DataFusion, one slice on four
partitions plans four partitions and returns the single-partition result. For
byte splits: the midpoint rule on a laid-out file (two halves select every
group once, a footer-only split selects nothing), two byte halves of a
synthetic base-only slice union to the whole file with the stream and eager
reads agreeing, an empty split keeps the schema, and a partial split of a slice
with log files is rejected while the covering split equals the merged read.
--
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]
