parthchandra opened a new issue, #7243:
URL: https://github.com/apache/datafusion-comet/issues/7243
### 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 `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:
1. Split the `TextScan` and `CsvScan` protobuf messages into common data
(schemas, options, object store options)
and per-partition data (one file group).
2. Make the exec classes implement `CometScanWithPlanData`, so
`PlanDataInjector.findAllPlanData` collects the
common bytes once and one entry per partition.
`NativeScanPlanDataInjector` is the model.
3. Add an injector for each scan, registered with the others, that fills in
the task's own group before the plan
reaches native code.
4. 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.
Related: #6784 (native Text scan), #6008.
### Steps to reproduce
_No response_
### Expected behavior
_No response_
### Additional context
_No response_
### Willingness to contribute
None
--
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]
