Repository navigation
Description
Describe the bug
What is the problem?
The native Text scan (#6784) and the native CSV V2 scan serialize every file group into the native plan with
addAllFilePartitions. Each task then copies that whole plan into JNI and decodes it before the native planner
picks its own group with self.partition. The work per task grows with the number of file groups, so the total
grows with file count times task count.
Measurements from the review of #6784, by @sunchao (not re-measured here):
- 1,024 two-byte files, with
maxPartitionBytesandopenCostInBytesboth set to1m, give aCometExecRDD
with 1,024 tasks and an emptycommonByKey. - The serialized plan is 85,039 bytes and contains all 1,024 groups. Keeping only the first task's group makes it
126 bytes. - In three isolated Rust trials, 1,024 full-plan decodes took 159-177 ms. Decoding single-group plans took
0.68-0.70 ms.
This matters little for the small lookup tables the native Text scan targets, but the cost has no ceiling as the
file count rises.
Where it happens
CometTextNativeScanExec.convertandCometCsvNativeScanExec.convertbuild oneTextScan/CsvScanmessage
that holds everyFilePartition.- The
TextScanandCsvScanarms innative/core/src/execution/planner.rsselect
scan.file_partitions[self.partition]after decoding the whole message.
Proposed approach
Use the per-partition plan-data injection that the native Parquet scan and the Iceberg scan already use:
- Split the
TextScanandCsvScanprotobuf messages into common data (schemas, options, object store options)
and per-partition data (one file group). - Make the exec classes implement
CometScanWithPlanData, soPlanDataInjector.findAllPlanDatacollects the
common bytes once and one entry per partition.NativeScanPlanDataInjectoris the model. - Add an injector for each scan, registered with the others, that fills in the task's own group before the plan
reaches native code. - Change the native planner arms to read the injected group instead of indexing
file_partitions.
Splitting partitions that mix object stores (CometScanUtils.splitPartitionsByStore) must keep working: the split
happens before the groups are serialized, so each task still gets one group that reads from one store.
Acceptance
- A plan for N file groups no longer carries all N groups into every task.
- Add a test, like the one in the review, that checks the size of the plan a task receives with many small files.
- Existing Text and CSV tests still pass, including the multi-store and ABFS fallback tests.
Notes
- Do both scans together. They share the shape of the problem and most of the code.
- This needs the Rust change compiled and tested. It touches the protobuf definitions, so check the
Operator
field numbers onmainbefore adding anything.
Activity