You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
feat: Give each native Text and CSV scan task only its own file group (use plan-data injection) #7243
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 maxPartitionBytes and openCostInBytes both set to 1m, give a CometExecRDD
with 1,024 tasks and an empty commonByKey.
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.convert and CometCsvNativeScanExec.convert build one TextScan / CsvScan message
that holds every FilePartition.
The TextScan and CsvScan arms in native/core/src/execution/planner.rs select 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 TextScan and CsvScan protobuf messages into common data (schemas, options, object store options)
and per-partition data (one file group).
Make the exec classes implement CometScanWithPlanData, so PlanDataInjector.findAllPlanData collects the
common bytes once and one entry per partition. NativeScanPlanDataInjector is 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 on main before adding anything.
Activity