gengliangwang opened a new pull request, #59961: URL: https://github.com/apache/spark/pull/59961
### What changes were proposed in this pull request? `NativePartitionReader` imports every batch of a native data source (#59424) into the same Arrow vectors, which Spark reads through `ReusedArrowColumnVector`. This PR adds `NativeColumnVector`, a `ReusedArrowColumnVector` for the top-level columns of the reader, which reads their values at the addresses of the buffers of each batch: - After each import, the reader calls `refresh()` on each column, which reads the addresses of the buffers of the batch, and whether it has nulls. The vector then reads booleans, integers, floats, dates, timestamps, decimals, and `Utf8` and `Utf8View` strings with `Platform`, and the nulls from the validity bitmap, or none when the batch has no null. - It returns the decimals that fit their precision in a long, of a scale from 0 to their precision, as `Decimal.createUnsafe(unscaled, precision, scale)`, as Spark's own column vectors return them. The other decimals, which do not fit their precision or a long, or are read at another scale, or have a negative scale, whose integral conversions long-backed decimals do not support, still read through `ArrowColumnVector`, and fail or read in full as before. - Binary values, structs, lists and maps, and the fields of structs read through `ArrowColumnVector`, as before. ### Why are the changes needed? `ArrowColumnVector` reads each value through Arrow Java, which by default checks on each access the bounds and the reference count of the buffer, and whether the value is null: for a value of a nullable `BIGINT` column, `isNullAt` and `getLong` make three checked reads of buffers. The reader does not need these checks. Spark reads only the rows of a batch, whose buffers Arrow Java sizes by the lengths of the arrays, which the reader checks before it imports them; the buffers stay allocated until the reader imports the next batch or closes; and Spark reads a value of a nullable column only after `isNullAt`, while the reader checks that the columns that the schema declares not nullable have no null. A decimal of at most 18 digits also becomes a BigDecimal-backed `Decimal`, which `Decimal.+` then adds as BigDecimals, since it only adds two decimals as longs when both are long-backed. Reading a batch of 8192 rows of nullable columns without nulls, as the generated code reads them (`isNullAt`, then the value), on Java 17, with each column vector in its own JVM so that the call sites see one class, as they do in the code of a query (a local benchmark, not part of this PR): | Columns | `ReusedArrowColumnVector` | `NativeColumnVector` | | --- | --- | --- | | `BIGINT`, `INT`, `DECIMAL(15,2)` and `STRING`, summed | 43.7 ns/row | 16.0 ns/row | | `BIGINT`, summed | 4.6 ns/row | 0.3 ns/row | | `DECIMAL(15,2)`, summed with `Decimal.+` | 22.7 ns/row | 14.0 ns/row | | `STRING` views of 64 bytes, their `numBytes` summed | 9.4 ns/row | 0.5 ns/row | End to end, the native data source framework of master cannot run the Parquet source of the Rust SDK of #59692, whose ABI is that of 74aa0dffc53 of #59424, before predicate pushdown was left out. On that framework, with the same column vector, the queries of that source take this fraction of the time of Spark's own Parquet reader, which alternates with it in the same session, with `local[4]` on a shared 32-thread Linux host and Java 25: | Query | Before | After | | --- | --- | --- | | `sum` of 4 `DECIMAL(15,2)` columns, 30M rows | 2.29 | 0.96 | | `sum` of 3 `BIGINT` and 1 `INT` columns, 30M rows | 1.26 | 0.68 | | TPC-H Q1 on a table with the columns of TPC-H lineitem, 30M rows | 1.04 | 0.98 | | `count(*), sum(l_quantity) ... WHERE l_shipmode = 'AIR'`, same table | 1.05 | 0.74 | | `count(payload)` of a 64-character string, 50M rows | 0.79 | 0.54 | | `sum(id)`, 50M rows | 0.94 | 0.70 | The queries that drop most rows in the source change less, since Spark reads few values: `TPC-H Q6` 0.71 and 0.67, `l_comment LIKE 'ab%'` 0.51 and 0.52. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? - The new `NativeColumnVectorSuite` compares `NativeColumnVector` with `ArrowColumnVector` on the same Arrow vectors: booleans, integers, floats with NaN and -0.0, dates and timestamps, with nulls over several bytes of the bitmap; strings and string views, in the views and in data buffers, over several batches; decimals of every precision from 1 to 18, at the bounds of their precision, out of it and out of a long, read at other precisions and scales, and of a negative scale with `spark.sql.legacy.allowNegativeScaleOfDecimal` on and off, with the checks of `ArrowColumnVectorSuite`; batches with other buffers, with and without nulls, empty and larger; the types that still read through `ArrowColumnVector`; and that the vector is reused, so that Spark does not close it. - `build/sbt 'sql/testOnly *NativeColumnVectorSuite *ArrowColumnVectorSuite'` on Java 17: 38 passed. - `NativeDataSourceSuite`, `ColumnarDataSourceSuite` and `NativeColumnVectorSuite`, with the test JVM on Java 25, which native data sources require: 81 passed, and 2 canceled, a test for the Java versions before 22, which passed on Java 17, and one that needs the Python dependencies of pandas UDFs. The Java 25 of this host is 25.0.2, so the tests ran with the parallel collector, which JDK-8377811 does not affect, and with sbt itself on Java 17. - `build/sbt sql/checkstyle sql/scalastyle sql/Test/scalastyle`. - On the framework of 74aa0dffc53, with the same column vector, the Spark smoke scripts of the Parquet and Range examples of the Rust SDK, which compare their reads with those of Spark. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) This pull request and its description were written by Isaac. -- 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]
