parthchandra opened a new pull request, #7491:
URL: https://github.com/apache/datafusion-comet/pull/7491
… not grow with the table
## Which issue does this PR close?
Closes # 7490.
Rationale for this change
Each native Iceberg scan task deserialized and serialized the whole scan's
delete-file metadata, even when its own files needed a tiny fraction of it. On
large merge-on-read tables this made executor heap use grow with the size of
the table, and tasks ran out of memory where Spark does not. See the issue for
the numbers.
The goal is to match the memory Spark's own Iceberg reader uses, on both the
driver and the executors.
### What changes are included in this PR?
Per-partition pools (executors)
- The five per-file pools (partition values, delete-file sets, residuals,
delete files, delete-file paths) are no longer sent in the common block. Each
partition carries the entries its own tasks use, in a new field
IcebergScan.own_pools of a new message type IcebergFilePools.
- The common block keeps only the small pools (schemas, partition specs and
types, name mappings, projected field ids) and the scan's settings.
- IcebergScanPools builds each partition's message on the driver,
renumbering task indices into its own pools. On the executor,
IcebergPlanDataInjector.inject merges it with the common block
(IcebergScanPools.forPartition).
- Native code needs no change to read these plans. It now rejects a plan
that still carries own_pools, as a safety net.
Lazy per-partition data (driver)
- New PartitionPlanData replaces Array[Array[Byte]] as the per-partition
plan data type in CometExecRDD, NativeExecContext, CometNativeShuffleInputRDD
and PlanDataInjector.findAllPlanData.
- A partition object holds a PlanDataSlice. That builds the partition's
bytes only while Spark serializes its task, through custom
writeObject/readObject. So the driver holds each partition's tasks once, and
never every partition's message at the same time. This matches when Spark
serializes Iceberg's input partitions.
- CometScanWithPlanData.partitionPlanData defaults to the scan's existing
byte arrays, so native Parquet and contrib scans (Delta, Lance) behave as
before.
Logging
- The driver log now reports the whole scan's pool sizes, the size of the
common block, and per-partition task bytes.
Behavior compared with the latest release
- Query results, fallback decisions and configs are unchanged.
- Per-task executor memory for Iceberg scan planning data no longer grows
with the table.
- Two driver log lines changed format.
- Pool entries that no task uses are no longer sent. Before, native
validated every entry in the pool, so a malformed but unused delete file could
fail a scan.
Known limits (documented in the IcebergScanPools scaladoc and the Iceberg
user guide)
- A coalesce without a shuffle combines several input partitions into one
task. That task then carries a copy of a shared delete file for every input
partition that uses it, for example with unpartitioned equality deletes.
Spark's Iceberg reader writes a shared delete file once per task. A test pins
the current behavior.
- Each partition's message is built during task serialization, which Spark
runs while holding its scheduler lock. Spark's Iceberg reader does comparable
serialization work at the same point.
- The driver keeps the scan's pools as protobuf objects for the life of the
plan, as Spark keeps Iceberg's task objects.
### How are these changes tested?
Unit tests (no Spark session)
- IcebergScanPoolsSuite:
- Randomized tests, over 200 seeds:
- every task resolves to the same entries once shipped and injected
- each partition ships exactly the entries its tasks use, once each
- the common block holds none of them
- The common block, each partition's message, and the injected plan stay
the same size as the scan grows from 2 to 10,000 partitions. That holds both
with partition-local delete files and with delete files shared across
partitions.
- Fields outside the five pools survive.
- An out-of-range index fails with an error naming the pool.
- A guard fails if a new pool or pool index field is added without
updating IcebergScanPools.
- PlanDataInjectorSuite:
- The injected plan from the real injectPlanData and serializeOperator
path does not grow with the scan.
- A partition's plan data is built only when its partition is serialized,
and arrives intact.
- A coalesced task carries one copy per input partition. This pins the
documented limit.
- CometNativeShuffleInputRDDSuite:
- The same lazy-serialization check for shuffle map tasks.
- Building an RDD's partitions builds no plan data.
- A native planner test checks that a plan still carrying own_pools is
rejected.
Integration tests, compared with Spark
- New SQL file test sql-tests/iceberg/merge_on_read_partition_layout.sql:
- A merge-on-read table partitioned by identity on five columns: a string
with non-ASCII values, a date, a decimal, a timestamp and a timestamp without
time zone.
- The values include dates and timestamps before 1970, the epoch, a leap
day, both ends of the decimal range, daylight-saving gap and overlap times, and
a partition where every value is null.
- It runs with preserve-data-grouping both on and off, in a non-UTC
session time zone.
- New CometIcebergNativeSuite tests:
- delete files shipped per partition, with grouped and ungrouped reads
- V3 deletion vectors sharing one Puffin file across partitions
- a global equality delete read across many partitions
- Existing assertions that read pools from the common block now read them
from each partition's message.
Test runs
- Unit suites pass on Scala 2.12 and 2.13.
- CometIcebergNativeSuite and CometFuzzIcebergSuite pass on Spark 3.4, apart
from one test that also fails on the base commit with Spark 3.4. That test is
unrelated to this change and is tracked in #.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
