parthchandra opened a new issue, #7490: URL: https://github.com/apache/datafusion-comet/issues/7490
### Describe the bug With spark.comet.scan.icebergNative.enabled=true, the driver serializes a native Iceberg scan as two parts: - A common block, with the scan's deduplicated pools: delete files, delete-file paths, delete-file sets, partition values and residuals. - One block per partition, holding that partition's file scan tasks. The common block holds every pool entry for the whole scan, and it reaches executors twice: 1. It rides in the stage's task binary, so every running task deserializes its own copy. 2. IcebergPlanDataInjector.inject copies it into each partition's native plan. Then PlanDataInjector.serializeOperator allocates one byte array for that plan, per partition. So the cost of each partition's plan grows with the whole scan's delete metadata, not with that partition's own files. On a bucketed merge-on-read table with about 1M buckets and about 3M position delete files: - the common block was about 800 MB - the largest partition's own tasks were about 2 KB - every partition still serialized a plan of about 800 MB With 2 cores per executor, tasks failed with java.lang.OutOfMemoryError: Java heap space in PlanDataInjector.serializeOperator, at a heap size Spark itself runs the same query in. A run with a much larger heap succeeded, but peaked near its limit. The native side pays too. parse_file_scan_tasks_from_common converts every delete file and clones every delete-file set in the pool, once per partition plan. That's O(scan size) per task, and O(partitions × scan size) per stage. ### Steps to reproduce 1. Create an Iceberg v2 table partitioned by bucket(N, id) with merge-on-read deletes and write.delete.granularity=partition. 2. Write many buckets, then run a DELETE that touches every bucket, so there are many position delete files. 3. Read it with spark.sql.iceberg.planning.preserve-data-grouping=true and the native Iceberg scan enabled. 4. Watch the driver log line IcebergScan: ... Common pools (unique/bytes): ... and the executor heap. The common block grows with the total number of delete files, and each task's injected plan is at least that large. ### Expected behavior Each task should carry only the plan data its own files need, as Spark's Iceberg reader does: it ships each task's delete files with the task. The cost of scan planning data per task should not grow with the size of the table. ### Additional context Spark's Iceberg reader (SparkInputPartition) serializes each input partition's task group, including its DeleteFiles, per task. It also serializes them lazily, when the task launches. ### 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]
