gengliangwang opened a new pull request, #59692: URL: https://github.com/apache/spark/pull/59692
### What changes were proposed in this pull request? This draft adds `rust_sdk_parquet`, a batch-read Parquet source on the Rust SDK of #59520 and the Parquet reader of [arrow-rs](https://github.com/apache/arrow-rs) 58.4. It is stacked on #59520, which depends on #59424: only the last commit, 4163ea57b74, is new here, and the other commits are those of #59520. The source is a new workspace member under `examples/src/main/rust/native-parquet`: - **Splits.** The driver lists the files of a path, skipping the names that start with `_` or `.` as Spark does, and splits each file into byte ranges of at most `maxPartitionBytes`, 128 MiB by default. A task reads the footer of its file and the row groups whose middle falls in its range, as Spark assigns row groups to splits, so that each row group belongs to one task. - **Projection.** A task decodes only the top-level columns that Spark reads, and returns them in the order of the negotiated schema. - **Row filter.** The pushed predicates that the reader evaluates exactly as Spark does become an arrow-rs `RowFilter`: the reader decodes the columns of a predicate first, and the other columns only for the rows that pass. Spark's own vectorized reader can skip row groups and pages by their statistics, but decodes every row of the pages that it reads, and filters afterwards. The reader evaluates `=`, `<>`, `<=>`, `<`, `<=`, `>`, `>=`, and `IN` on top-level boolean, integer, string, binary, date, timestamp, and decimal columns, `STARTS_WITH`, `ENDS_WITH`, and `CONTAINS` on strings, `IS_NULL` and `IS_NOT_NULL` on any column, and `AND`, `OR`, and `NOT` of these, with three-valued logic, so that it drops exactly the rows that Spark drops. Comparisons of floating-point columns remain for Spark: Arrow compares floats by the IEEE 754 total order, in which -0.0 is less than 0.0 and some NaNs are less than all numbers, while Spark equates the zeros and orders NaN above all numbers. A push ed limit caps the rows of each task, so the source does not report it as pushed. - **Types.** As Spark does without `mergeSchema`, the schema is that of one file, the first by path, with nullable columns at every level. The source reads the default arrow-rs conversion of the Parquet schema, ignoring any Arrow schema embedded in the files, except for INT96 columns, Spark's default encoding of timestamps: an arrow-rs schema hint reads them as microseconds in UTC, also in the years that nanoseconds overflow. The hint applies only to the leaves whose physical type is INT96, since for an INT64 column arrow-rs would reinterpret nanoseconds as microseconds. The source rejects the types that Spark converts or rejects, such as millisecond and nanosecond timestamps and unsigned integers, when Spark asks for the schema. - **Options.** `path`, `batchSize` (8192), `maxPartitionBytes`, and `rowFilter` (true), which compares the reader with and without its row filter. `smoke.py` checks the source against Spark's own Parquet reader, and `benchmark.py` compares their times. The SDK README links the example. The lockfile adds `parquet` 58.4 and its dependencies, with the snappy, zstd, gzip, and lz4 codecs. The code of the SDK and of the other examples is unchanged. ### Why are the changes needed? To find out whether a Parquet source on this API can be faster than Spark's own reader, and to exercise the SDK with an existing Rust library rather than generated data. The physical plans have the same shape: the scan is a leaf of the same stage, and `ColumnarToRow` starts the same whole-stage codegen pipeline as for Spark's vectorized reader, so the source can only make the scan itself faster. The benchmark reads 50M rows of `id BIGINT`, `k INT` with 100 values, and a 64-character string `payload`: 2.0 GB of snappy Parquet in 4 files, which both readers read in 16 tasks. It runs with `local[4]` on a shared 32-core Linux host, on Spark built from the pinned framework with Java 25, with a release build of the library. Each time is the median of 5 runs after 2 warm-up runs, with the files in the page cache, and includes planning the query again. | Query | Spark's reader | `rust_sdk_parquet` | without `RowFilter` | | --- | --- | --- | --- | | `SELECT sum(id)` | 0.281 s | 0.279 s | 0.231 s | | `SELECT sum(id), sum(k), max(length(payload))` | 4.716 s | 4.721 s | 4.706 s | | `SELECT sum(id), max(length(payload)) ... WHERE k < 10` | 1.808 s | 1.670 s | 1.894 s | | `SELECT sum(id), max(length(payload)) ... WHERE k = 7` | 1.500 s | 1.262 s | 1.638 s | With the bounds and null checks of Arrow Java disabled, by `-Darrow.enable_unsafe_memory_access=true -Darrow.enable_null_check_for_get=false`, a JVM-wide setting that Spark does not change: | Query | Spark's reader | `rust_sdk_parquet` | without `RowFilter` | | --- | --- | --- | --- | | `SELECT sum(id)` | 0.256 s | 0.185 s | 0.176 s | | `SELECT sum(id), sum(k), max(length(payload))` | 4.838 s | 4.365 s | 4.356 s | | `SELECT sum(id), max(length(payload)) ... WHERE k < 10` | 1.767 s | 1.161 s | 1.266 s | | `SELECT sum(id), max(length(payload)) ... WHERE k = 7` | 1.424 s | 0.837 s | 0.937 s | - With Spark's default settings, the source matches Spark's reader on full scans, and is 8% and 16% faster on the queries that select 10% and 1% of the rows, thanks to its row filter: without it, the source is 5% and 9% slower on them. - Spark reads the Arrow vectors that the framework imports through `ArrowColumnVector`, whose reads Arrow Java checks by default. Without these checks, the source is 10% to 41% faster than Spark's reader, so reading values from Arrow vectors is the main cost that the framework of #59424 could reduce, for example with column vectors that read the imported buffers without the checks. - On this shared host, runs differ by up to about 15%: an earlier run with 20M rows found the source 15% slower than Spark's reader on the full scan, and 8% and 16% faster on the selective queries. - The source reports no statistics, so Spark does not plan to broadcast its tables in joins, and the framework has no runtime filtering. In queries with joins, these matter more than the speed of decoding. ### Does this PR introduce _any_ user-facing change? No change to Spark. It adds an unpublished example connector of the experimental Rust SDK, which runs in a Spark distribution that contains #59424, with Java 22 or later. ### How was this patch tested? From `sql/native/rust`: ```sh cargo build --workspace --locked cargo test --workspace --locked cargo clippy --workspace --all-targets --locked -- -D warnings cargo fmt --all -- --check RUSTDOCFLAGS='-D warnings' cargo doc --workspace --no-deps --locked ``` All 97 tests passed, including the 77 of #59520 and 20 new tests of the source. The filter tests cover three-valued logic for comparisons, `IN` lists with null under `NOT`, `AND`, and `OR`, literals on the left of comparisons, byte-wise string and binary comparisons, the types of literals, including integers out of the range of their column and decimals of another scale, and the predicates that remain for Spark: floats, nested fields, unknown functions, and conjunctions with any of these. The reader tests cover the types that Spark writes, with nested nullability, the order of projected columns and the batch size, byte ranges that read each row group once, the row filter with and without the option, empty projections that count filtered rows, per-task limits, INT96 columns at the top level and in a struct outside the range of nanoseconds, the rejection of other timestamp units and unsigned integers, the listing of directories, a file that lacks a column, cancellation, and options. The smoke script passed with `spark-submit --master 'local[2]'` on Spark built from the pinned framework with Maven and Java 25. It writes 20,000 rows of 19 columns with Spark three times: with INT96 timestamps, with small row groups read in splits of 64 KiB, and with microsecond timestamps in a single file. On each, 33 queries return what Spark's own reader returns, with and without the row filter, including filters on NaN and -0.0, on nested fields, and on every pushed predicate kind. It also checks the pushed predicates in the formatted plan, a limit, and the rejection of millisecond timestamps. The library exports only `spark_ds_init`, and the Python checks for syntax, ASCII, line length, and whitespace passed. **Not run:** clusters, other JVMs and platforms, object stores, and files written by other writers than Spark and arrow-rs. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Isaac (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]
