sunchao commented on code in PR #24526: URL: https://github.com/apache/datafusion/pull/24526#discussion_r3858424729
########## datafusion/pruning/src/string_in_list.rs: ########## @@ -0,0 +1,219 @@ +// 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. + +use std::fmt::{self, Display, Formatter}; +use std::hash::{Hash, Hasher}; +use std::sync::Arc; + +use arrow::array::{Array, AsArray, BooleanArray}; +use arrow::compute::cast; +use arrow::datatypes::{DataType, Schema}; +use arrow::record_batch::RecordBatch; +use datafusion_common::{Result, assert_eq_or_internal_err}; +use datafusion_physical_expr::{PhysicalExpr, PhysicalExprRef}; +use datafusion_physical_plan::ColumnarValue; + +/// Tests whether a sorted string domain intersects an inclusive statistics interval. +/// This expression is used only for pruning; the original IN remains the row filter. +#[derive(Debug, Eq)] +pub(crate) struct StringInListPruningExpr { + min: PhysicalExprRef, + max: PhysicalExprRef, + values: Arc<[String]>, +} + +impl StringInListPruningExpr { + pub(crate) fn new( + min: PhysicalExprRef, + max: PhysicalExprRef, + mut values: Vec<String>, + ) -> Self { + values.sort_unstable(); + values.dedup(); + Self { + min, + max, + values: values.into(), + } + } +} + +impl PartialEq for StringInListPruningExpr { + fn eq(&self, other: &Self) -> bool { + self.min.eq(&other.min) && self.max.eq(&other.max) && self.values == other.values + } +} + +impl Hash for StringInListPruningExpr { + fn hash<H: Hasher>(&self, state: &mut H) { + self.min.hash(state); + self.max.hash(state); + self.values.hash(state); + } +} + +impl Display for StringInListPruningExpr { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + write!( + f, + "IN_SET_INTERSECTS({}, {}, {} values)", + self.min, + self.max, + self.values.len() + ) + } +} + +fn has_oversized_string_buffer(array: &dyn Array, limit: usize) -> bool { + match array.data_type() { + DataType::Utf8 => array.as_string::<i32>().values().len() >= limit, + DataType::LargeUtf8 => array.as_string::<i64>().values().len() >= limit, + DataType::Dictionary(_, _) => has_oversized_string_buffer( + array.as_any_dictionary().values().as_ref(), + limit, + ), + _ => false, + } +} + +impl PhysicalExpr for StringInListPruningExpr { + fn data_type(&self, _input_schema: &Schema) -> Result<DataType> { + Ok(DataType::Boolean) + } + + fn nullable(&self, _input_schema: &Schema) -> Result<bool> { + Ok(true) + } + + fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> { + // Normalize Utf8, LargeUtf8, Utf8View, and dictionary-encoded statistics. + let min = self.min.evaluate(batch)?.into_array(batch.num_rows())?; + let max = self.max.evaluate(batch)?.into_array(batch.num_rows())?; + // A short string slice can retain a buffer too large for Utf8View's + // u32 offsets. Avoid a panic in the cast and keep pruning conservative. + if has_oversized_string_buffer(min.as_ref(), u32::MAX as usize) + || has_oversized_string_buffer(max.as_ref(), u32::MAX as usize) + { + return Ok(ColumnarValue::Array(Arc::new(BooleanArray::new_null( + batch.num_rows(), + )))); + } + // Dictionary values can be NULL behind valid keys. Preserve their + // validity even if the view cast only carries the key nulls. + let min_nulls = min.logical_nulls(); + let max_nulls = max.logical_nulls(); + let min = cast(&min, &DataType::Utf8View)?; + let max = cast(&max, &DataType::Utf8View)?; + let min = min.as_string_view(); + let max = max.as_string_view(); + let matches: BooleanArray = (0..batch.num_rows()) + .map(|i| { + let min = (min.is_valid(i) + && min_nulls.as_ref().is_none_or(|nulls| nulls.is_valid(i))) + .then(|| min.value(i).as_bytes()); + let max = (max.is_valid(i) + && max_nulls.as_ref().is_none_or(|nulls| nulls.is_valid(i))) + .then(|| max.value(i).as_bytes()); + match (min, max) { + (Some(min), Some(max)) => { + if min > max { + return None; + } + let index = self.values.partition_point(|v| v.as_bytes() < min); Review Comment: Yes—the compact expression reads the same gated min/max columns as the per-value expressions; it does not access raw Parquet metadata. The [row-group adapter](https://github.com/apache/datafusion/blob/168b7b5d569a5b8848f630cf0e2d251502878b1e/datafusion/datasource-parquet/src/row_group_filter.rs#L554-L604) withholds bounds for unusable footer ordering and masks deprecated byte-array bounds to NULL. The [page adapter](https://github.com/apache/datafusion/blob/168b7b5d569a5b8848f630cf0e2d251502878b1e/datafusion/datasource-parquet/src/page_filter.rs#L543-L617) also withholds untrusted bounds, and [runtime row-group pruning](https://github.com/apache/datafusion/blob/168b7b5d569a5b8848f630cf0e2d251502878b1e/datafusion/datasource-parquet/src/push_decoder.rs#L230-L263) reuses the row-group adapter. The distinction is that we suppress unusable bounds, not the entire predicate: the compact expression can still run with NULL bounds and return UNKNOWN, keeping the container unless independent statistics, such as an all-null count, safely exclude it. The [22-value page regression](https://github.com/apache/datafusion/blob/168b7b5d569a5b8848f630cf0e2d251502878b1e/datafusion/datasource-parquet/src/statistics_order_tests.rs#L363-L427) explicitly asserts `IN_SET_INTERSECTS` and verifies that the matching `az` row survives missing/unknown column order. Agreed about making the ordering assumption explicit: Rust string ordering and these byte comparisons both use unsigned lexicographic UTF-8 byte order. Generic `PruningStatistics` providers retain the existing responsibility to supply bounds in the comparison order, or mark them unavailable. A short comment here would make that contract clearer. ########## datafusion/pruning/src/string_in_list.rs: ########## @@ -0,0 +1,219 @@ +// 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. + +use std::fmt::{self, Display, Formatter}; +use std::hash::{Hash, Hasher}; +use std::sync::Arc; + +use arrow::array::{Array, AsArray, BooleanArray}; +use arrow::compute::cast; +use arrow::datatypes::{DataType, Schema}; +use arrow::record_batch::RecordBatch; +use datafusion_common::{Result, assert_eq_or_internal_err}; +use datafusion_physical_expr::{PhysicalExpr, PhysicalExprRef}; +use datafusion_physical_plan::ColumnarValue; + +/// Tests whether a sorted string domain intersects an inclusive statistics interval. +/// This expression is used only for pruning; the original IN remains the row filter. +#[derive(Debug, Eq)] +pub(crate) struct StringInListPruningExpr { + min: PhysicalExprRef, + max: PhysicalExprRef, + values: Arc<[String]>, +} + +impl StringInListPruningExpr { + pub(crate) fn new( + min: PhysicalExprRef, + max: PhysicalExprRef, + mut values: Vec<String>, + ) -> Self { + values.sort_unstable(); + values.dedup(); + Self { + min, + max, + values: values.into(), + } + } +} + +impl PartialEq for StringInListPruningExpr { + fn eq(&self, other: &Self) -> bool { + self.min.eq(&other.min) && self.max.eq(&other.max) && self.values == other.values + } +} + +impl Hash for StringInListPruningExpr { + fn hash<H: Hasher>(&self, state: &mut H) { + self.min.hash(state); + self.max.hash(state); + self.values.hash(state); + } +} + +impl Display for StringInListPruningExpr { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + write!( + f, + "IN_SET_INTERSECTS({}, {}, {} values)", + self.min, + self.max, + self.values.len() + ) + } +} + +fn has_oversized_string_buffer(array: &dyn Array, limit: usize) -> bool { + match array.data_type() { + DataType::Utf8 => array.as_string::<i32>().values().len() >= limit, + DataType::LargeUtf8 => array.as_string::<i64>().values().len() >= limit, + DataType::Dictionary(_, _) => has_oversized_string_buffer( + array.as_any_dictionary().values().as_ref(), + limit, + ), + _ => false, + } +} + +impl PhysicalExpr for StringInListPruningExpr { + fn data_type(&self, _input_schema: &Schema) -> Result<DataType> { + Ok(DataType::Boolean) + } + + fn nullable(&self, _input_schema: &Schema) -> Result<bool> { + Ok(true) + } + + fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> { + // Normalize Utf8, LargeUtf8, Utf8View, and dictionary-encoded statistics. + let min = self.min.evaluate(batch)?.into_array(batch.num_rows())?; + let max = self.max.evaluate(batch)?.into_array(batch.num_rows())?; + // A short string slice can retain a buffer too large for Utf8View's + // u32 offsets. Avoid a panic in the cast and keep pruning conservative. + if has_oversized_string_buffer(min.as_ref(), u32::MAX as usize) + || has_oversized_string_buffer(max.as_ref(), u32::MAX as usize) + { + return Ok(ColumnarValue::Array(Arc::new(BooleanArray::new_null( + batch.num_rows(), + )))); + } + // Dictionary values can be NULL behind valid keys. Preserve their + // validity even if the view cast only carries the key nulls. + let min_nulls = min.logical_nulls(); + let max_nulls = max.logical_nulls(); + let min = cast(&min, &DataType::Utf8View)?; + let max = cast(&max, &DataType::Utf8View)?; + let min = min.as_string_view(); + let max = max.as_string_view(); + let matches: BooleanArray = (0..batch.num_rows()) + .map(|i| { + let min = (min.is_valid(i) + && min_nulls.as_ref().is_none_or(|nulls| nulls.is_valid(i))) + .then(|| min.value(i).as_bytes()); + let max = (max.is_valid(i) + && max_nulls.as_ref().is_none_or(|nulls| nulls.is_valid(i))) + .then(|| max.value(i).as_bytes()); + match (min, max) { + (Some(min), Some(max)) => { + if min > max { + return None; + } + let index = self.values.partition_point(|v| v.as_bytes() < min); + Some(self.values.get(index).is_some_and(|v| v.as_bytes() <= max)) + } + // A single known bound can still exclude the whole domain. + (Some(min), None) Review Comment: Yes, that is the intended reading. With only `max`, the possible interval is unbounded below, so a nonempty sorted domain is disjoint exactly when `values.first() > max`. Symmetrically, with only `min`, it is disjoint exactly when `values.last() < min`. Equality must keep the container; otherwise we return UNKNOWN. Gaps between requested values cannot exclude an interval without its other endpoint. The [interval regression](https://github.com/apache/datafusion/blob/168b7b5d569a5b8848f630cf0e2d251502878b1e/datafusion/pruning/src/pruning_predicate.rs#L3646-L3721) checks both exclusions and equality at the endpoints for lists of 20, 21, 256, and 10,000 values. Agreed that spelling out the unbounded-interval interpretation would improve the comment. ########## datafusion/pruning/src/string_in_list.rs: ########## @@ -0,0 +1,219 @@ +// 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. + +use std::fmt::{self, Display, Formatter}; +use std::hash::{Hash, Hasher}; +use std::sync::Arc; + +use arrow::array::{Array, AsArray, BooleanArray}; +use arrow::compute::cast; +use arrow::datatypes::{DataType, Schema}; +use arrow::record_batch::RecordBatch; +use datafusion_common::{Result, assert_eq_or_internal_err}; +use datafusion_physical_expr::{PhysicalExpr, PhysicalExprRef}; +use datafusion_physical_plan::ColumnarValue; + +/// Tests whether a sorted string domain intersects an inclusive statistics interval. +/// This expression is used only for pruning; the original IN remains the row filter. +#[derive(Debug, Eq)] +pub(crate) struct StringInListPruningExpr { + min: PhysicalExprRef, + max: PhysicalExprRef, + values: Arc<[String]>, +} + +impl StringInListPruningExpr { + pub(crate) fn new( + min: PhysicalExprRef, + max: PhysicalExprRef, + mut values: Vec<String>, + ) -> Self { + values.sort_unstable(); + values.dedup(); + Self { + min, + max, + values: values.into(), + } + } +} + +impl PartialEq for StringInListPruningExpr { + fn eq(&self, other: &Self) -> bool { + self.min.eq(&other.min) && self.max.eq(&other.max) && self.values == other.values + } +} + +impl Hash for StringInListPruningExpr { + fn hash<H: Hasher>(&self, state: &mut H) { + self.min.hash(state); + self.max.hash(state); + self.values.hash(state); + } +} + +impl Display for StringInListPruningExpr { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + write!( + f, + "IN_SET_INTERSECTS({}, {}, {} values)", + self.min, + self.max, + self.values.len() + ) + } +} + +fn has_oversized_string_buffer(array: &dyn Array, limit: usize) -> bool { + match array.data_type() { + DataType::Utf8 => array.as_string::<i32>().values().len() >= limit, + DataType::LargeUtf8 => array.as_string::<i64>().values().len() >= limit, + DataType::Dictionary(_, _) => has_oversized_string_buffer( + array.as_any_dictionary().values().as_ref(), + limit, + ), + _ => false, + } +} + +impl PhysicalExpr for StringInListPruningExpr { + fn data_type(&self, _input_schema: &Schema) -> Result<DataType> { + Ok(DataType::Boolean) + } + + fn nullable(&self, _input_schema: &Schema) -> Result<bool> { Review Comment: Yes at the final container-decision level, although these cases are in `pruning_predicate.rs`, not the Parquet integration module: - [`large_string_in_list_handles_unicode_and_unknown_bounds`](https://github.com/apache/datafusion/blob/168b7b5d569a5b8848f630cf0e2d251502878b1e/datafusion/pruning/src/pruning_predicate.rs#L3809-L3864) includes inverted bounds `min="z", max="m"` and asserts that the container is kept. - [`large_string_in_list_preserves_dictionary_nulls`](https://github.com/apache/datafusion/blob/168b7b5d569a5b8848f630cf0e2d251502878b1e/datafusion/pruning/src/pruning_predicate.rs#L3744-L3806) checks missing bounds from NULL keys/values and inverted bounds `min="zz", max=""`; those containers are also kept. Both call `PruningPredicate::prune` and assert the final Boolean decisions, so they cover UNKNOWN becoming “keep,” rather than only checking the expression's NULL result. There is not a dedicated full Parquet scan test with an inverted footer. Separately, the [compact page-order regression](https://github.com/apache/datafusion/blob/168b7b5d569a5b8848f630cf0e2d251502878b1e/datafusion/datasource-parquet/src/statistics_order_tests.rs#L363-L427) reads a real Parquet fixture and verifies that matching rows survive missing/unknown statistics ordering. -- 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]
