spark-test commented on code in PR #59692: URL: https://github.com/apache/spark/pull/59692#discussion_r4348260148
########## examples/src/main/rust/native-parquet/README.md: ########## @@ -0,0 +1,141 @@ +<!-- +Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. +See the NOTICE file distributed with this work for additional information regarding copyright +ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. You may obtain a copy of the +License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software distributed under the License is +distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or +implied. See the License for the specific language governing permissions and limitations under the +License. +--> + +# Rust native Parquet source + +This batch-read example implements `DataSource`, `ScanBuilder`, and `PartitionReaderFactory` from +the [Rust native data source SDK][sdk] on the Parquet reader of [arrow-rs][arrow-rs]. Its +`DataSource` sets `MicroBatchStream`, `BatchWrite`, and `StreamingWrite` to `Unsupported`, and the +library exports `export_data_source!(ParquetSource)` under the name `rust_sdk_parquet`. + +The example targets C ABI version 1 from [Spark PR #59424][native-source-pr], pinned to +`74aa0dffc53463034fa6549796c4dd599e9705de`. Running it in Spark requires that framework and Java 22 +or later on the driver and executors. The standalone Rust tests do not need Spark or a JVM. + +## How it reads + +The driver lists the files of the path and splits each one into byte ranges of at most +`maxPartitionBytes`, as Spark splits the files of its own sources. A task reads the footer of its Review Comment: **Nit (P3):** `ParquetScan::plan` (lib.rs:166-180) does not split files the way Spark's file sources do. Spark sizes splits as `min(maxPartitionBytes, max(openCostInBytes, totalBytes / parallelism))` and bin-packs small files into shared partitions. This source plans fixed per-file ranges of `maxPartitionBytes`, with at least one task per file. The two readers really share only the midpoint row-group assignment. For example, 10,000 files of 1 MB become 10,000 tasks here, where Spark's reader plans a few hundred, and a 100 MB file with 16 MB row groups is read by a single task here, where Spark spreads it over several cores. The benchmark's four 512 MB files hide both effects, so readers comparing the two readers outside that regime are misled. The same sentence is in the lib.rs module doc (line 18). Please scope both to the midpoint assignment and list the missing small-file packing and parallelism-based sizing under the limits of this example. **Verification:** - **Inspection:** The README and lib.rs module doc describe split planning as fixed per-file ranges of at most maxPartitionBytes with midpoint row-group assignment, consistent with ParquetScan::plan, and the limits section names the missing small-file packing and parallelism-based sizing. ########## examples/src/main/rust/native-parquet/src/filter.rs: ########## @@ -0,0 +1,646 @@ +// Licensed to the Apache Software Foundation (ASF) under one or more +// contributor license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright ownership. +// The ASF licenses this file to You under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! The pushed predicates that the reader evaluates in its `RowFilter`. + +use std::sync::Arc; + +use arrow_arith::boolean::{and_kleene, is_not_null, is_null, not, or_kleene}; +use arrow_array::{ + ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array, Int16Array, Int32Array, + Int64Array, Int8Array, RecordBatch, Scalar, StringArray, TimestampMicrosecondArray, +}; +use arrow_ord::cmp; +use arrow_schema::{ArrowError, DataType, Schema, TimeUnit}; +use arrow_string::like; +use spark_native_source::{LiteralValue, Predicate}; + +/// A predicate on the top-level columns of a file, which evaluates to what Spark SQL computes, +/// including nulls, so that the reader drops exactly the rows that Spark drops. +#[derive(Debug)] +pub(crate) enum Filter { + /// A column compared with a literal of its type, or with null. + Compare { + column: String, + op: Comparison, + value: Option<ArrayRef>, + }, + IsNull { + column: String, + negated: bool, + }, + /// A column in a list of literals, which may contain null. + In { + column: String, + values: Vec<Option<ArrayRef>>, + }, + /// A string column matched with a string literal. + Text { + column: String, + op: TextMatch, + pattern: ArrayRef, + }, + And(Box<Filter>, Box<Filter>), + Or(Box<Filter>, Box<Filter>), + Not(Box<Filter>), +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Comparison { + Equal, + NotEqual, + NullSafeEqual, + Less, + LessOrEqual, + Greater, + GreaterOrEqual, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum TextMatch { + StartsWith, + EndsWith, + Contains, +} + +impl Filter { + /// The filter of a predicate on the columns of `schema`, or `None` if the reader does not + /// evaluate the predicate exactly as Spark does. + pub(crate) fn new(predicate: &Predicate, schema: &Schema) -> Option<Self> { + let Predicate::Function { name, children } = predicate else { + return None; + }; + let filter = |child| Self::new(child, schema).map(Box::new); + Some(match (name.as_str(), children.as_slice()) { + ("AND", [left, right]) => Self::And(filter(left)?, filter(right)?), + ("OR", [left, right]) => Self::Or(filter(left)?, filter(right)?), + ("NOT", [child]) => Self::Not(filter(child)?), + ("IS_NULL" | "IS_NOT_NULL", [child]) => Self::IsNull { + column: column(child, schema)?.0, + negated: name == "IS_NOT_NULL", + }, + ("IN", [child, values @ ..]) if !values.is_empty() => { + let (column, data_type) = comparable_column(child, schema)?; + let values = values + .iter() + .map(|value| literal(value, data_type)) + .collect::<Option<_>>()?; + Self::In { column, values } + } + ("STARTS_WITH" | "ENDS_WITH" | "CONTAINS", [child, pattern]) => { + let (column, data_type) = column(child, schema)?; + if *data_type != DataType::Utf8 { + return None; + } + let op = match name.as_str() { + "STARTS_WITH" => TextMatch::StartsWith, + "ENDS_WITH" => TextMatch::EndsWith, + _ => TextMatch::Contains, + }; + Self::Text { + column, + op, + pattern: literal(pattern, data_type)??, + } + } + (name, [left, right]) => { + let op = comparison(name)?; + // Spark may put the literal on either side. + let (column, data_type, value, op) = match comparable_column(left, schema) { + Some((column, data_type)) => (column, data_type, right, op), + None => { + let (column, data_type) = comparable_column(right, schema)?; + (column, data_type, left, op.reversed()) + } + }; + Self::Compare { + column, + op, + value: literal(value, data_type)?, + } + } + _ => return None, + }) + } + + /// Adds the names of the columns that the filter reads to `columns`. + pub(crate) fn columns<'a>(&'a self, columns: &mut Vec<&'a str>) { + match self { + Self::Compare { column, .. } + | Self::IsNull { column, .. } + | Self::In { column, .. } + | Self::Text { column, .. } => { + if !columns.contains(&column.as_str()) { + columns.push(column); + } + } + Self::And(left, right) | Self::Or(left, right) => { + left.columns(columns); + right.columns(columns); + } + Self::Not(child) => child.columns(columns), + } + } + + /// Evaluates the filter on a batch that has the columns that it reads. A row is null when + /// Spark computes null, which the reader drops, as Spark drops it, unless a NOT changes it. Review Comment: **Nit (P3):** Under the three-valued logic implemented here (`not`, `and_kleene`, `or_kleene`), NOT of null is null, so a NOT never turns a dropped null row into a kept one. The module's own test says so: "A null value makes the rows that match no value null, which NOT keeps null". Only AND or OR can resolve a null operand, when the other operand decides the result (`null OR true`, `null AND false`). The code is right, but this comment states the wrong rule for when a null row survives. A maintainer who follows it when extending `Filter` could break the exact-filter guarantee. Suggest something like: "A row is null when Spark computes null, and the reader drops it as Spark does. NOT keeps null as null; only AND and OR can resolve a null operand." **Verification:** - **Inspection:** The Filter::evaluate comment states that NOT of null is null and that only AND or OR can resolve a null operand, consistent with the in-list and logical-operator three-valued-logic test. -- 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]
