Skip to content

feat: Give each native Text and CSV scan task only its own file group (use plan-data injection) #7243

Open

Description

@parthchandra

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.

Metadata

Metadata

Assignees

Labels

Type

No type

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions