Native shuffle fails with a 2GB task serialization OOM on jobs with very many partitions #5391

Describe the bug

A job with a huge number of scan partitions fails at stage submission:

Task serialization failed: java.lang.OutOfMemoryError: Required array length 2147483639 + 794 is too large
    at java.io.ByteArrayOutputStream.ensureCapacity(...)
    at org.apache.spark.serializer.JavaSerializerInstance.serialize(...)
    at org.apache.spark.scheduler.DAGScheduler.submitMissingTasks(DAGScheduler.scala:1528)

A single byte[] can't exceed ~2GB, and the driver is serializing one object bigger than that. The failing job had ~38.7M partitions (multi-petabyte dataset), native scan into a native shuffle. The same job runs fine on plain Spark, so it's something Comet does differently, not just scale.

Analysis

DAGScheduler.submitMissingTasks serializes the (RDD, ShuffleDependency) pair once into a single broadcast byte array, which must fit in ~2GB. On the native shuffle path, CometShuffleDependency.nativeShuffleSpec is a non-transient field holding a NativeExecContext whose perPartitionByKey: Map[String, Array[Array[Byte]]] carries one
serialized scan plan-data blob (the partition's file list) per map partition. With ~38.7M partitions that's ~38.7M protobufs baked into the broadcast task binary, so it exceeds 2GB — even though each write task only ever reads its own partition's slice.

Spark's FileScanRDD keeps per-partition file lists @transient; each partition's files ride in its own Partition object and are serialized separately per task, so the full list is never included in the shared broadcast blob.

Steps to reproduce

No response

Expected behavior

No response

Additional context

No response