sunchao opened a new pull request, #5557: URL: https://github.com/apache/datafusion-comet/pull/5557
## Which issue does this PR close? Closes #5555. ## Rationale for this change Setting `spark.sql.execution.arrow.useLargeVarTypes=true` currently disables Comet's accelerated `mapInArrow` / `mapInPandas` path, even for a small batch that is already columnar. For example, a Parquet batch containing strings and binary values falls back through `CometColumnarToRow` and Spark's Arrow writer so that Python receives `large_string` and `large_binary`. This preserves Spark's requested types but reintroduces the row conversion that the accelerated path avoids. This PR keeps eligible operations accelerated while supplying real 64-bit offset buffers. Changing the IPC schema alone would be incorrect because ordinary Comet string and binary vectors have 32-bit offsets. ## What changes are included in this PR? The Python runner reads the setting from the driver-captured worker configuration and recursively advertises large string/binary fields, including fields inside structs, arrays, and maps. During serialization it widens only ordinary string/binary offset buffers. Value and validity buffers stay borrowed from the source batch, and already-large offsets are reused. Temporary offsets use the writer allocator and are released after the synchronous write, including partial allocation and write failures. The default `useLargeVarTypes=false` path does not allocate conversion buffers or traverse offsets. The planner's unconditional fallback and its now-unused SQLConf shim accessors are removed. Batch compatibility is checked against the converted field types, so an ordinary input batch and an already-large batch can share the stream schema. Empty inputs also advertise the configured types. Spark 3.x retains its existing unsupported-API fallback. The documentation now describes the supported input types and the remaining limit: widening at the Python boundary does **not** remove the 32-bit size limits of native producers that first construct ordinary arrays. This is input support; it does not claim to resolve the separate output serialization issue #5488. The existing end-to-end benchmark gains `BENCHMARK_LARGE_VAR_TYPES=true` to exercise the new path. ## How are these changes tested? All final checks used the branch rebased onto upstream main after #5368 merged, with native code rebuilt before the JVM and Python checks. - Spark 4.0 root-reactor package build: **20 focused JVM tests passed** (14 Arrow serializer tests and 6 planner tests). Coverage includes real large string/binary round trips, payloads larger than the writer's allocation limit, already-large offsets, empty batches, cross-allocator FFI ownership, injected IPC write failure, and failure while allocating a later offset buffer. - Spark 4.0.4 / JDK 17 / PyArrow 25.0.1 / pandas 3.0.5: **all 128 Python integration tests passed**. New cases assert worker-visible large types and accelerated plans for both APIs, including Unicode, binary values, nulls, nested containers, multiple small batches, union inputs with differing nested field names, chained UDFs, and empty inputs. Against the parent build, the six new accelerated cases failed their plan assertions while the six fallback cases passed. - Spark 3.5 root-reactor compatibility run: **28 tests passed**, with one expected Spark-4-only test canceled. - **24 benchmark smoke scenarios passed**: all workload/API/mode combinations with each large-type setting, using 1,000 rows, no warmup, and one iteration. These are functional smoke checks, not representative timing results. - Additional local downstream probes matched Spark for collect, limit, broadcast join, filter/project, and Comet repartition in both modes. The join and filter/project remained Spark operators; these probes do not establish native broadcast support for large output vectors. - Scala formatting/style checks, Python syntax, 204 suite-registration checks, and `git diff --check` passed. Python lint comparison found no newly introduced findings. To isolate the conversion cost, a local JVM harness called the production serializer with 8,192 rows, alternating string/binary columns containing 32-byte values and one null per seven rows. After warmup, it took the median of seven samples, alternating measurement order. The output channel discarded bytes, so these measurements isolate serializer work and **exclude Python, transport copies, and end-to-end query costs**: | Variable-width columns | Default offsets | Large offsets | Added time per batch | Peak writer allocation, default / large | | --- | ---: | ---: | ---: | ---: | | 2 | 3.0 us | 52.3 us | 49.3 us | 1 KiB / 257 KiB | | 16 | 12.6 us | 399.7 us | 387.1 us | 1 KiB / 2,049 KiB | The peak includes Arrow allocator rounding and the wrapping struct bitmap; writer allocation returned to zero after serialization. These local measurements describe the cost of offset widening, not a claimed query speedup. Spark 4.1 validation remains **unverified**: the root-reactor attempt stalled while resolving `jackson-bom:2.21.2` from the configured Maven mirror and was stopped after 90 seconds. The mirror configuration was left unchanged. -- 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]
