parthchandra opened a new issue, #7523: URL: https://github.com/apache/datafusion-comet/issues/7523
### Is your feature request related to a problem or challenge? After #7490, each native Iceberg scan partition ships the delete files, delete-file paths and sets, partition values and residuals its own tasks use, in that partition's `IcebergScan.common`. The executors no longer pay for the whole scan's delete metadata. The driver still builds every partition's message eagerly. `CometIcebergNativeScan.serializePartitions` returns `Array[Array[Byte]]`, one serialized message per partition. That array is held through `perPartitionByKey` in `CometExecRDD`, and in the `CometExecPartition` objects that Spark keeps for the life of the job. Two costs follow: - **The driver holds every partition's bytes at once**, alongside the Iceberg `FileScanTask` and `DeleteFile` objects the scan already caches. - **Entries that many partitions share are serialized once per partition**, and all those copies live on the driver together. Examples are an unpartitioned equality delete, or partition-scoped deletes of an Iceberg partition split across many Spark partitions. A table with N such delete files read as P partitions holds roughly N × P serialized delete entries on the driver. Spark's Iceberg reader (`SparkInputPartition`) keeps one set of Java objects on the driver. It serializes each task's input partition only when the task launches, so it never holds every task's serialized payload at once. ### Describe the solution you'd like Build each partition's bytes when Spark serializes its task, not up front: - Replace `Array[Array[Byte]]` with a per-partition source (for example, `PartitionPlanData` with `length` and `apply(i)`) in `CometExecRDD`, `NativeExecContext`, `CometNativeShuffleInputRDD` and `PlanDataInjector.findAllPlanData`. - Let the partition object carry a small handle that builds its bytes in `writeObject`, and holds the received bytes after `readObject`. - On the driver, keep the scan's full pools in compact form (for example, the serialized pools plus an offset per entry), and parse only the entries a task uses. - Keep `CometScanWithPlanData.perPartitionData` working for native Parquet and contrib scans, by defaulting the new source to the existing byte arrays. Questions to settle first: - **Measure the driver heap** for eager messages on a layout with heavily shared deletes, before and after. A synthetic test showed a parsed `IcebergScanCommon` taking about 3.9× its serialized size, while bytes plus an offset index took about 1.1×. These are microbenchmark numbers, not driver measurements from a real job. - **Cost under the scheduler lock:** building a partition's bytes during task serialization runs inside Spark's scheduler lock (`TaskSchedulerImpl.resourceOffers`). Iceberg's reader serializes its task groups at the same point, but the per-task build cost should be measured on a scan with large partitions. - **Coalesced tasks:** a coalesce without a shuffle serializes several input partitions into one task, so shared entries repeat once per input partition. Should the lazy build deduplicate them within one task stream, as Java serialization does for Iceberg's shared `DeleteFile` objects? ### Describe alternatives you've considered - **Keep eager messages.** This is fine when each delete file belongs to one partition, as with grouped reads of partition-scoped deletes. Driver memory grows with how widely deletes are shared. - **Ship widely shared entries once, through a broadcast.** This saves driver and network bytes, but every task deserializes the broadcast copy, which is the executor cost #7490 removed. -- 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]
