[sunchao]

Which issue does this PR close?

Closes #7309.

Rationale for this change

Every broadcast input batch currently allocates a 1 MiB output chunk and constructs a compression codec. Thousands of tiny batches repeat both costs. This shared serializer improvement applies to every broadcast and is independent of the experimental native batching in #7258, so it can land first and be reverted independently.

What changes are included in this PR?

• Size output chunks from Arrow field and dictionary buffers, clamped to 4 KiB–1 MiB. Serialization metadata can exceed the estimate; the stream grows additional chunks as needed.
• Lazily create one compression codec per serializer iterator while keeping an independent compressed output stream for each batch.
• Add a one-row, many-column regression whose compressed schema metadata exceeds the small buffer estimate and round-trips all values across chunks.

The serialized stream format and broadcast assembly remain unchanged. No new configuration is introduced.

How are these changes tested?

Both changed Scala files compile against Spark 3.5.9 dependencies. All 16 UtilsSuite tests pass, including dictionary serialization, sliced/large-offset vectors, metadata-driven multi-chunk output and the existing broadcast string-size cap. The new test fails against the exact main serializer at d7f4fcaa58, which produces one large chunk. JVM library tracing confirms these tests use Arrow Java/compression libraries without loading Comet's native library. Formatting and git diff --check pass.

Hosted Spark 4.1 production/test compilation also passed in the scans job, but its tests did not start because Maven failed to download the Surefire JUnit4 provider. Both artifact URLs are available now. GitHub currently rejects a targeted retry while the containing workflow is still running; other CI checks remain in progress.

Performance measurements

Used the producer and complete-query harnesses from #7258 at 48e1024bfc, with native batching OFF for both serializer variants below. The baseline is main's Utils at d7f4fcaa58; the candidate overrides only compiled Utils classes from this PR (2660acf816). All other Java/native artifacts and settings are identical. Class origins and native library hashes were verified.

Fresh JVMs ran both case orders on four pinned AMD EPYC Milan CPUs, JDK 17.0.21 / Spark 3.5.9, with 4 GiB heap, warmup and at least five iterations/two measured seconds per arm. Results are ranges of arm means, computed from logged elapsed time and iteration count, not confidence intervals.

Producer input, 8,192 total rows	Main serializer (ms)	This PR (ms)
1 row/frame, INT + BIGINT	1,080.8–1,132.2	164.3–187.1
8 rows/frame, INT + BIGINT	132.5–140.5	22.21–23.34
One full 8,192-row frame	0.316–0.333	0.211–0.214
64 rows/frame plus repeated 6 KiB STRING	42.0–43.8	31.98–32.08

Values, row counts, stream counts and serialized bytes are identical between serializers. The producer includes native decoding, FFI and serialization, but excludes transport and driver assembly. These numbers measure chunk sizing and codec reuse together; they do not isolate the codec contribution.

The complete local-shuffle → broadcast join → aggregation query measured 1,098.4–1,397.8 ms on main versus 694.4–856.4 ms with this PR, with batching OFF. It uses 65,536 build rows, 128 writers, 64 reducers and 1,048,576 probe rows; every iteration regenerates inputs and includes shuffle, broadcast distribution, join and collection. All results equal [1048576,88351366032]; input streams remain 8,192. The full OFF/ON order controls are reported in #7258.

These are controlled local[4], explicit-broadcast, AQE-off measurements using the benchmark's on-heap mode and an unbounded native pool. They do not establish general workload gains or production memory safety. All eight JVM runs succeeded with unchanged artifact hashes. The shared release JNI artifact uses current source with an alternate lockfile (Arrow 59.3.1/DataFusion 55.2.0 unchanged; other versions differ because the registry lacks a submitted AWS version), LTO off and 16 codegen units. This PR itself changes no native source or dependencies.