diff --git a/.github/workflows/audit.yml b/.github/workflows/audit.yml index a6eec722ed3e1..bca3390370175 100644 --- a/.github/workflows/audit.yml +++ b/.github/workflows/audit.yml @@ -45,7 +45,7 @@ jobs: steps: - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 - name: Install cargo-audit - uses: taiki-e/install-action@0631aa6515c7d545823c67cfae7ef4fc7f490154 # v2.81.8 + uses: taiki-e/install-action@15449e3094499af05d8d964a1c884208e4b8b595 # v2.81.11 with: tool: cargo-audit - name: Run audit check diff --git a/.github/workflows/breaking_changes_detector.yml b/.github/workflows/breaking_changes_detector.yml index 264ba16e3c8d0..a0d7bf6520ee5 100644 --- a/.github/workflows/breaking_changes_detector.yml +++ b/.github/workflows/breaking_changes_detector.yml @@ -89,7 +89,7 @@ jobs: - name: Install cargo-semver-checks if: steps.changed_crates.outputs.packages != '' - uses: taiki-e/install-action@0631aa6515c7d545823c67cfae7ef4fc7f490154 # v2.81.8 + uses: taiki-e/install-action@15449e3094499af05d8d964a1c884208e4b8b595 # v2.81.11 with: tool: cargo-semver-checks diff --git a/.github/workflows/dev.yml b/.github/workflows/dev.yml index 48f5ea8939c8c..43da64c569162 100644 --- a/.github/workflows/dev.yml +++ b/.github/workflows/dev.yml @@ -64,7 +64,7 @@ jobs: source ci/scripts/utils/tool_versions.sh echo "LYCHEE_VERSION=${LYCHEE_VERSION}" >> "$GITHUB_ENV" - name: Install lychee - uses: taiki-e/install-action@0631aa6515c7d545823c67cfae7ef4fc7f490154 # v2.81.8 + uses: taiki-e/install-action@15449e3094499af05d8d964a1c884208e4b8b595 # v2.81.11 with: tool: lychee@${{ env.LYCHEE_VERSION }} - name: Run markdown link check diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 770e705dddc90..cc48a54f27ec4 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -429,7 +429,7 @@ jobs: sudo apt-get update -qq sudo apt-get install -y -qq clang - name: Setup wasm-pack - uses: taiki-e/install-action@0631aa6515c7d545823c67cfae7ef4fc7f490154 # v2.81.8 + uses: taiki-e/install-action@15449e3094499af05d8d964a1c884208e4b8b595 # v2.81.11 with: tool: wasm-pack - name: Run tests with headless mode @@ -773,7 +773,7 @@ jobs: - name: Setup Rust toolchain uses: ./.github/actions/setup-builder - name: Install cargo-msrv - uses: taiki-e/install-action@0631aa6515c7d545823c67cfae7ef4fc7f490154 # v2.81.8 + uses: taiki-e/install-action@15449e3094499af05d8d964a1c884208e4b8b595 # v2.81.11 with: tool: cargo-msrv diff --git a/Cargo.lock b/Cargo.lock index df6b263adb009..9e70bf18ae5b7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3721,9 +3721,9 @@ dependencies = [ [[package]] name = "insta-cmd" -version = "0.6.0" +version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ffeeefa927925cced49ccb01bf3e57c9d4cd132df21e576eb9415baeab2d3de6" +checksum = "bffdf4af1db390cf0401535d7c1303cd079a074d28d8473b026fdb6559c41403" dependencies = [ "insta", "serde", @@ -4803,9 +4803,9 @@ dependencies = [ [[package]] name = "prost" -version = "0.14.3" +version = "0.14.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d2ea70524a2f82d518bce41317d0fae74151505651af45faf1ffbd6fd33f0568" +checksum = "528ac67416ff8646872a3c02cad9cc4ee5dc9f9540c9b10771855c95cb2e5ae1" dependencies = [ "bytes", "prost-derive", @@ -4813,9 +4813,9 @@ dependencies = [ [[package]] name = "prost-build" -version = "0.14.3" +version = "0.14.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7" +checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ "heck", "itertools 0.14.0", @@ -4832,9 +4832,9 @@ dependencies = [ [[package]] name = "prost-derive" -version = "0.14.3" +version = "0.14.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b" +checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", "itertools 0.14.0", @@ -4845,9 +4845,9 @@ dependencies = [ [[package]] name = "prost-types" -version = "0.14.3" +version = "0.14.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8991c4cbdb8bc5b11f0b074ffe286c30e523de90fee5ba8132f1399f23cb3dd7" +checksum = "f94967dc7688f3054c7fac87473ffae4cc4c3904800e2d9f5b857246d8963b0a" dependencies = [ "prost", ] diff --git a/datafusion-cli/Cargo.toml b/datafusion-cli/Cargo.toml index 441ae00c11db0..62eedafe798d4 100644 --- a/datafusion-cli/Cargo.toml +++ b/datafusion-cli/Cargo.toml @@ -75,7 +75,7 @@ workspace = true [dev-dependencies] ctor = { workspace = true } insta = { workspace = true } -insta-cmd = "0.6.0" +insta-cmd = "0.7.0" rstest = { workspace = true } testcontainers-modules = { workspace = true, features = ["minio"] } # Makes sure `test_display_pg_json` behaves in a consistent way regardless of diff --git a/datafusion/common/src/heap_size.rs b/datafusion/common/src/heap_size.rs index 802f9d3883222..869946d82414f 100644 --- a/datafusion/common/src/heap_size.rs +++ b/datafusion/common/src/heap_size.rs @@ -326,47 +326,6 @@ impl DFHeapSize for Fields { } } -impl DFHeapSize for StructArray { - fn heap_size(&self, _: &mut DFHeapSizeCtx) -> usize { - self.get_array_memory_size() - } -} - -impl DFHeapSize for LargeListArray { - fn heap_size(&self, _: &mut DFHeapSizeCtx) -> usize { - self.get_array_memory_size() - } -} - -impl DFHeapSize for LargeListViewArray { - fn heap_size(&self, _: &mut DFHeapSizeCtx) -> usize { - self.get_array_memory_size() - } -} - -impl DFHeapSize for ListArray { - fn heap_size(&self, _: &mut DFHeapSizeCtx) -> usize { - self.get_array_memory_size() - } -} - -impl DFHeapSize for ListViewArray { - fn heap_size(&self, _: &mut DFHeapSizeCtx) -> usize { - self.get_array_memory_size() - } -} - -impl DFHeapSize for FixedSizeListArray { - fn heap_size(&self, _: &mut DFHeapSizeCtx) -> usize { - self.get_array_memory_size() - } -} -impl DFHeapSize for MapArray { - fn heap_size(&self, _: &mut DFHeapSizeCtx) -> usize { - self.get_array_memory_size() - } -} - impl DFHeapSize for Box { fn heap_size(&self, ctx: &mut DFHeapSizeCtx) -> usize { size_of::() + self.as_ref().heap_size(ctx) @@ -469,6 +428,29 @@ impl_zero_heap_size!( DateTime, ); +/// Implement [`DFHeapSize`] for Arrow arrays types. +macro_rules! impl_array_heap_size { + ($($t:ty),+ $(,)?) => { + $( + impl DFHeapSize for $t { + fn heap_size(&self, _: &mut DFHeapSizeCtx) -> usize { + self.get_array_memory_size() + } + } + )+ + }; +} + +impl_array_heap_size!( + StructArray, + LargeListArray, + LargeListViewArray, + ListArray, + ListViewArray, + FixedSizeListArray, + MapArray, +); + #[cfg(test)] mod tests { use super::*; @@ -696,4 +678,61 @@ mod tests { let field = Field::new("temperature", DataType::Float64, true); assert!(size(&field) > 0); } + + #[test] + fn test_list_array() { + use arrow::array::types::Int32Type; + + let array = ListArray::from_iter_primitive::(vec![ + Some(vec![Some(1), Some(2), Some(3)]), + Some(vec![Some(4)]), + ]); + assert_eq!(size(&array), array.get_array_memory_size()); + assert!(size(&array) > 0); + + let large = + LargeListArray::from_iter_primitive::(vec![Some(vec![ + Some(1), + Some(2), + ])]); + assert_eq!(size(&large), large.get_array_memory_size()); + assert!(size(&large) > 0); + } + + #[test] + fn test_struct_array() { + use arrow::array::Int32Array; + + let array = StructArray::from(vec![( + Arc::new(Field::new("a", DataType::Int32, true)), + Arc::new(Int32Array::from(vec![1, 2, 3])) as _, + )]); + assert_eq!(size(&array), array.get_array_memory_size()); + assert!(size(&array) > 0); + } + + #[test] + fn test_fixed_size_list_array() { + use arrow::array::Int32Array; + + let values = Arc::new(Int32Array::from(vec![1, 2, 3, 4])); + let field = Arc::new(Field::new("item", DataType::Int32, true)); + let array = FixedSizeListArray::new(field, 2, values, None); + assert_eq!(size(&array), array.get_array_memory_size()); + assert!(size(&array) > 0); + } + + #[test] + fn test_map_array() { + use arrow::array::{Int32Builder, MapBuilder, StringBuilder}; + + let mut builder = + MapBuilder::new(None, StringBuilder::new(), Int32Builder::new()); + builder.keys().append_value("key"); + builder.values().append_value(1); + builder.append(true).unwrap(); + let array = builder.finish(); + assert_eq!(size(&array), array.get_array_memory_size()); + assert!(size(&array) > 0); + } } diff --git a/datafusion/datasource-parquet/src/row_filter.rs b/datafusion/datasource-parquet/src/row_filter.rs index f19dbd6c6fa63..8ce359942cc4f 100644 --- a/datafusion/datasource-parquet/src/row_filter.rs +++ b/datafusion/datasource-parquet/src/row_filter.rs @@ -72,6 +72,7 @@ use arrow::array::BooleanArray; use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; use arrow::error::{ArrowError, Result as ArrowResult}; use arrow::record_batch::RecordBatch; +use datafusion_functions::core::file_row_index::FileRowIndexFunc; use datafusion_functions::core::getfield::GetFieldFunc; use parquet::arrow::ProjectionMask; use parquet::arrow::arrow_reader::{ArrowPredicate, RowFilter}; @@ -260,6 +261,9 @@ struct PushdownChecker<'schema> { non_primitive_columns: bool, /// Does the expression reference any columns not present in the file schema? projected_columns: bool, + /// Does the expression references a ScalarUDF that requires some rewrite + /// and therefore can't be pushed down into the row-filter. + has_unpushable_udfs: bool, /// Indices into the file schema of columns required to evaluate the expression. /// Does not include struct columns accessed via `get_field`. required_columns: Vec, @@ -276,6 +280,7 @@ impl<'schema> PushdownChecker<'schema> { Self { non_primitive_columns: false, projected_columns: false, + has_unpushable_udfs: false, required_columns: Vec::new(), struct_field_accesses: Vec::new(), allow_list_columns, @@ -372,7 +377,7 @@ impl<'schema> PushdownChecker<'schema> { #[inline] fn prevents_pushdown(&self) -> bool { - self.non_primitive_columns || self.projected_columns + self.non_primitive_columns || self.projected_columns || self.has_unpushable_udfs } /// Consumes the checker and returns sorted, deduplicated column indices @@ -484,6 +489,13 @@ impl TreeNodeVisitor<'_> for PushdownChecker<'_> { return Ok(recursion); } + if ScalarFunctionExpr::try_downcast_func::(node.as_ref()) + .is_some() + { + self.has_unpushable_udfs = true; + return Ok(TreeNodeRecursion::Jump); + } + Ok(TreeNodeRecursion::Continue) } } diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 8228cd273eae6..840b86dcb875d 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -26,6 +26,9 @@ use crate::opener::ParquetMorselizer; use crate::opener::build_pruning_predicates; use crate::opener::build_virtual_columns_state; use crate::row_filter::can_expr_be_pushed_down_with_schemas; +use arrow_schema::Fields; +use arrow_schema::extension::ExtensionType; +use arrow_schema::{DataType, Field}; use datafusion_common::config::ConfigOptions; #[cfg(feature = "parquet_encryption")] use datafusion_common::config::EncryptionFactoryOptions; @@ -40,9 +43,14 @@ use datafusion_common::config::TableParquetOptions; use datafusion_datasource::TableSchema; use datafusion_datasource::file::FileSource; use datafusion_datasource::file_scan_config::FileScanConfig; +use datafusion_functions::core::file_row_index::FileRowIndexFunc; +use datafusion_physical_expr::expressions::Column; use datafusion_physical_expr::projection::ProjectionExprs; use datafusion_physical_expr::{EquivalenceProperties, conjunction}; -use datafusion_physical_expr_adapter::DefaultPhysicalExprAdapterFactory; +use datafusion_physical_expr_adapter::expr_references_scalar_udf; +use datafusion_physical_expr_adapter::{ + DefaultPhysicalExprAdapterFactory, rewrite_file_row_index_projection, +}; use datafusion_physical_expr_common::physical_expr::PhysicalExpr; use datafusion_physical_expr_common::physical_expr::fmt_sql; use datafusion_physical_plan::DisplayFormatType; @@ -60,6 +68,7 @@ use datafusion_execution::parquet_encryption::EncryptionFactory; use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr}; use itertools::Itertools; use object_store::ObjectStore; +use parquet::arrow::RowNumber; #[cfg(feature = "parquet_encryption")] use parquet::encryption::decrypt::FileDecryptionProperties; @@ -669,7 +678,28 @@ impl FileSource for ParquetSource { projection: &ProjectionExprs, ) -> datafusion_common::Result>> { let mut source = self.clone(); - source.projection = self.projection.try_merge(projection)?; + + // If there's no reference to `FileRowIndexFunc` in the projection, we can just merge + // both projections as-is, there's no need to modify the projection first. + if !projection.iter().any(|projection_expr| { + expr_references_scalar_udf::(&projection_expr.expr) + }) { + source.projection = self.projection.try_merge(projection)?; + return Ok(Some(Arc::new(source))); + } + + // If we can find a reference to `FileRowIndexFunc`, we add it as a virtual column + // or re-use an existing one in the table's schema. + let (table_schema, row_index_col) = + table_schema_with_row_index_col(self.table_schema()); + + source.table_schema = table_schema; + source.projection = rewrite_file_row_index_projection( + &self.projection, + projection, + &row_index_col, + )?; + Ok(Some(Arc::new(source))) } @@ -952,15 +982,12 @@ impl FileSource for ParquetSource { reversed_eq_properties.ordering_satisfy(order.iter().cloned())?; let sort_order = LexOrdering::new(order.iter().cloned()); let column_in_file_schema = sort_order.as_ref().is_some_and(|s| { - s.first() - .expr - .downcast_ref::() - .is_some_and(|col| { - self.table_schema - .file_schema() - .field_with_name(col.name()) - .is_ok() - }) + s.first().expr.downcast_ref::().is_some_and(|col| { + self.table_schema + .file_schema() + .field_with_name(col.name()) + .is_ok() + }) }); if !column_in_file_schema && !reversed_satisfies { @@ -989,6 +1016,69 @@ impl FileSource for ParquetSource { } } +/// Returns the a [`TableSchema`] containing a [`RowNumber`] virtual column and a [`Column`] expression referencing its row index column. +/// The expression is then merged into a projection. +/// +/// - If the schema already has a virtual column with the [`RowNumber`] type, it returns the schema unchanged. +/// - If the schema doesn't have the appropriate virtual column, it returns a modified schema with the virtual column appended to it. +fn table_schema_with_row_index_col(table_schema: &TableSchema) -> (TableSchema, Column) { + // If we can find a virtual column with the `RowNumber` type, we just return the schema + // and create the appropriate `column` we're going to use + if let Some((idx, field)) = + table_schema + .virtual_columns() + .iter() + .enumerate() + .find(|(_, field)| { + field + .extension_type_name() + .is_some_and(|name| name == RowNumber::NAME) + }) + { + let virtual_offset = table_schema.file_schema().fields().len() + + table_schema.table_partition_cols().len(); + + return ( + table_schema.clone(), + Column::new(field.name(), virtual_offset + idx), + ); + } + + // The hidden field is shared across all files in this scan, but it must + // have a unique table-schema name because later rewrites resolve it by + // column name and index. + let base_row_index_name = "__datafusion_file_row_index"; + let mut row_index_name = base_row_index_name.to_string(); + let mut suffix = 0; + while table_schema + .table_schema() + .field_with_name(&row_index_name) + .is_ok() + { + suffix += 1; + row_index_name = format!("{base_row_index_name}_{suffix}"); + } + + let row_index_table_idx = table_schema.table_schema().fields().len(); + let row_index_field = Arc::new( + Field::new(&row_index_name, DataType::Int64, true).with_extension_type(RowNumber), + ); + ( + TableSchema::builder(Arc::clone(table_schema.file_schema())) + .with_table_partition_cols(table_schema.table_partition_cols().clone()) + .with_virtual_columns( + table_schema + .virtual_columns() + .iter() + .cloned() + .chain([row_index_field]) + .collect::(), + ) + .build(), + Column::new(&row_index_name, row_index_table_idx), + ) +} + #[cfg(test)] mod tests { use super::*; @@ -1622,7 +1712,9 @@ mod tests { use datafusion_common::config::ConfigOptions; use datafusion_datasource::TableSchema; use datafusion_expr::{col, lit as logical_lit}; + use datafusion_functions::core::expr_fn::file_row_index; use datafusion_physical_expr::planner::logical2physical; + use datafusion_physical_expr_adapter::rewrite_file_row_index_expr; use datafusion_physical_plan::filter_pushdown::PushedDown; use parquet::arrow::RowNumber; @@ -1652,13 +1744,20 @@ mod tests { .or(col("value").eq(logical_lit(4i64))), full_schema, ); + let (_, row_index_col) = table_schema_with_row_index_col(source.table_schema()); + let row_index = rewrite_file_row_index_expr( + logical2physical(&file_row_index().gt(logical_lit(2i64)), full_schema), + row_index_col.name(), + row_index_col.index(), + ) + .expect("file_row_index should rewrite to the row_number virtual column"); let config = ConfigOptions::default(); let prop = source - .try_pushdown_filters(vec![pushable, virtual_only, mixed], &config) + .try_pushdown_filters(vec![pushable, virtual_only, mixed, row_index], &config) .expect("try_pushdown_filters must not error"); - assert_eq!(prop.filters.len(), 3); + assert_eq!(prop.filters.len(), 4); assert!( matches!(prop.filters[0], PushedDown::Yes), "file-column filter should be pushable" @@ -1672,5 +1771,10 @@ mod tests { "filter mixing a virtual column with a file column must not be \ pushed down (row filter would silently drop it)" ); + assert!( + matches!(prop.filters[3], PushedDown::No), + "file_row_index() rewrites to a virtual column and must not be \ + pushed down" + ); } } diff --git a/datafusion/functions/src/core/file_row_index.rs b/datafusion/functions/src/core/file_row_index.rs new file mode 100644 index 0000000000000..7b2667a8b8768 --- /dev/null +++ b/datafusion/functions/src/core/file_row_index.rs @@ -0,0 +1,96 @@ +// 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. + +//! Implementation of the `file_row_index` scalar function. + +use arrow::datatypes::DataType; +use datafusion_common::utils::take_function_args; +use datafusion_common::{Result, exec_err}; +use datafusion_doc::Documentation; +use datafusion_expr::{ + ColumnarValue, ExpressionPlacement, ScalarFunctionArgs, ScalarUDFImpl, Signature, + Volatility, +}; +use datafusion_macros::user_doc; + +/// Scalar UDF implementation for `file_row_index()`. +/// +/// File sources that can expose per-file row indexes rewrite this placeholder +/// function into a source-provided physical expression. Direct evaluation +/// returns an error because there is no file context outside a scan. +#[user_doc( + doc_section(label = "Other Functions"), + description = r#"Returns the zero-based row offset within the source file +that produced the current row. + +The value is scoped to one file, so rows from different files in the same scan +can have the same row index. This function is intended to be rewritten at +file-scan time. If the input file is not known (for example, if this function +is evaluated outside a file scan, or was not pushed down into one), direct +evaluation returns an error. +"#, + syntax_example = "file_row_index()", + sql_example = r#"```sql +SELECT file_row_index() FROM t; +```"# +)] +#[derive(Debug, PartialEq, Eq, Hash)] +pub struct FileRowIndexFunc { + signature: Signature, +} + +impl Default for FileRowIndexFunc { + fn default() -> Self { + Self::new() + } +} + +impl FileRowIndexFunc { + pub fn new() -> Self { + Self { + signature: Signature::nullary(Volatility::Volatile), + } + } +} + +impl ScalarUDFImpl for FileRowIndexFunc { + fn name(&self) -> &str { + "file_row_index" + } + + fn signature(&self) -> &Signature { + &self.signature + } + + fn return_type(&self, args: &[DataType]) -> Result { + let [] = take_function_args(self.name(), args)?; + Ok(DataType::Int64) + } + + fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result { + let [] = take_function_args(self.name(), args.args)?; + exec_err!("file_row_index() is source dependent and cannot be evaluated directly") + } + + fn placement(&self, _args: &[ExpressionPlacement]) -> ExpressionPlacement { + ExpressionPlacement::MoveTowardsLeafNodes + } + + fn documentation(&self) -> Option<&Documentation> { + self.doc() + } +} diff --git a/datafusion/functions/src/core/mod.rs b/datafusion/functions/src/core/mod.rs index 5657f9d88810c..4665eca99ebef 100644 --- a/datafusion/functions/src/core/mod.rs +++ b/datafusion/functions/src/core/mod.rs @@ -28,6 +28,7 @@ pub mod arrowtypeof; pub mod cast_to_type; pub mod coalesce; pub mod expr_ext; +pub mod file_row_index; pub mod getfield; pub mod greatest; mod greatest_least_utils; @@ -67,6 +68,7 @@ make_udf_function!(version::VersionFunc, version); make_udf_function!(arrow_metadata::ArrowMetadataFunc, arrow_metadata); make_udf_function!(with_metadata::WithMetadataFunc, with_metadata); make_udf_function!(arrow_field::ArrowFieldFunc, arrow_field); +make_udf_function!(file_row_index::FileRowIndexFunc, file_row_index); pub mod expr_fn { use datafusion_expr::{Expr, Literal}; @@ -143,6 +145,9 @@ pub mod expr_fn { union_tag, "Returns the name of the currently selected field in the union", arg1 + ),( + file_row_index, + "Returns the offset of the row within its source file", )); #[doc = "Returns the value of the field with the given name from the struct"] @@ -196,5 +201,6 @@ pub fn functions() -> Vec> { union_tag(), version(), r#struct(), + file_row_index(), ] } diff --git a/datafusion/physical-expr-adapter/src/lib.rs b/datafusion/physical-expr-adapter/src/lib.rs index ea4db19ee110e..fa14bc8b4d150 100644 --- a/datafusion/physical-expr-adapter/src/lib.rs +++ b/datafusion/physical-expr-adapter/src/lib.rs @@ -29,5 +29,6 @@ pub mod schema_rewriter; pub use schema_rewriter::{ BatchAdapter, BatchAdapterFactory, DefaultPhysicalExprAdapter, DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter, PhysicalExprAdapterFactory, - replace_columns_with_literals, + expr_references_scalar_udf, replace_columns_with_literals, + rewrite_file_row_index_expr, rewrite_file_row_index_projection, }; diff --git a/datafusion/physical-expr-adapter/src/schema_rewriter.rs b/datafusion/physical-expr-adapter/src/schema_rewriter.rs index 56502ab8731a7..f287caf32ecda 100644 --- a/datafusion/physical-expr-adapter/src/schema_rewriter.rs +++ b/datafusion/physical-expr-adapter/src/schema_rewriter.rs @@ -25,16 +25,19 @@ use std::hash::Hash; use std::sync::Arc; use arrow::array::RecordBatch; -use arrow::datatypes::{DataType, FieldRef, SchemaRef}; +use arrow::datatypes::{DataType, Field, FieldRef, SchemaRef}; use datafusion_common::{ DataFusionError, Result, ScalarValue, exec_err, metadata::FieldMetadata, nested_struct::validate_data_type_compatibility, - tree_node::{Transformed, TransformedResult, TreeNode}, + tree_node::{Transformed, TransformedResult, TreeNode, TreeNodeRecursion}, +}; +use datafusion_expr::ScalarUDFImpl; +use datafusion_functions::core::{ + file_row_index::FileRowIndexFunc, getfield::GetFieldFunc, }; -use datafusion_functions::core::getfield::GetFieldFunc; use datafusion_physical_expr::PhysicalExprSimplifier; -use datafusion_physical_expr::projection::{ProjectionExprs, Projector}; +use datafusion_physical_expr::projection::{ProjectionExpr, ProjectionExprs, Projector}; use datafusion_physical_expr::{ ScalarFunctionExpr, expressions::{self, CastExpr, Column}, @@ -81,6 +84,114 @@ where .data() } +/// Return true if `expr` references scalar UDF `T`. +/// +/// This matches the concrete [`ScalarUDFImpl`] type rather than the function +/// name, so unrelated UDFs with the same name are not treated as matches. +pub fn expr_references_scalar_udf( + expr: &Arc, +) -> bool { + let mut found = false; + + expr.apply(|node| { + if ScalarFunctionExpr::try_downcast_func::(node.as_ref()).is_some() { + found = true; + return Ok(TreeNodeRecursion::Stop); + } + Ok(TreeNodeRecursion::Continue) + }) + .expect("Infallible traversal of PhysicalExpr tree failed"); + + found +} + +/// Rewrite occurrences of scalar UDF `T` in `expr` using `replacement`. +/// +/// The rewrite matches the concrete [`ScalarUDFImpl`] type rather than the +/// function name. `replacement` is called with each matching +/// [`ScalarFunctionExpr`] after its children have been rewritten. +fn rewrite_scalar_udf( + expr: Arc, + mut replacement: F, +) -> Result> +where + T: ScalarUDFImpl, + F: FnMut(&ScalarFunctionExpr) -> Result>, +{ + expr.transform_up(|node| { + if let Some(scalar_fn) = ScalarFunctionExpr::try_downcast_func::(node.as_ref()) + { + Ok(Transformed::yes(replacement(scalar_fn)?)) + } else { + Ok(Transformed::no(node)) + } + }) + .map(|transformed| transformed.data) +} + +/// Rewrite `file_row_index()` in `expr` to read from a source-provided +/// row-index column. +/// +/// `row_index_idx` is the index of `row_index_name` in the schema that the +/// rewritten expression will be evaluated against. The rewrite uses ordinary +/// physical expressions: a [`Column`] that reads the source row-index values +/// wrapped in a [`CastExpr`] that exposes the public `file_row_index: Int64` +/// return field without source-specific extension metadata. +pub fn rewrite_file_row_index_expr( + expr: Arc, + row_index_name: &str, + row_index_idx: usize, +) -> Result> { + rewrite_scalar_udf::(expr, |_| { + let source = Arc::new(Column::new(row_index_name, row_index_idx)); + let target_field = Arc::new(Field::new("file_row_index", DataType::Int64, true)); + Ok(Arc::new(CastExpr::new_with_target_field( + source, + target_field, + None, + ))) + }) +} + +/// Rewrite `file_row_index()` in a pushed projection to read from a +/// source-provided row-index column. +/// +/// +/// For example if `row_index_column` is `__datafusion_row_idx` this function rewrites all +/// instances of `file_row_index()` to `__datafusion_row_index` column references. +/// +/// `base_projection` is the current projection already pushed into a source. +/// The row-index source column is appended to that base projection if it is not +/// already present. `projection` is rewritten to read from the projected +/// row-index column and then merged on top of the extended base projection. +pub fn rewrite_file_row_index_projection( + base_projection: &ProjectionExprs, + projection: &ProjectionExprs, + row_index_col: &Column, +) -> Result { + let mut base_exprs = base_projection.as_ref().to_vec(); + let row_index_projection_idx = + base_projection.projected_column_position(row_index_col); + + // If the column doesn't exist in the projection yet + if row_index_projection_idx.is_none() { + base_exprs.push(ProjectionExpr { + expr: Arc::new(row_index_col.clone()), + alias: row_index_col.name().to_owned(), + }); + } + + let rewritten_projection = projection.clone().try_map_exprs(|expr| { + rewrite_file_row_index_expr( + expr, + row_index_col.name(), + row_index_projection_idx.unwrap_or(base_exprs.len() - 1), + ) + })?; + + ProjectionExprs::new(base_exprs).try_merge(&rewritten_projection) +} + /// Trait for adapting [`PhysicalExpr`] expressions to match a target schema. /// /// This is used in file scans to rewrite expressions so that they can be @@ -631,8 +742,8 @@ mod tests { RecordBatchOptions, StringArray, StringViewArray, StructArray, }; use arrow::datatypes::{Field, Fields, Schema}; - use datafusion_common::{assert_contains, record_batch}; - use datafusion_expr::Operator; + use datafusion_common::{assert_contains, config::ConfigOptions, record_batch}; + use datafusion_expr::{Operator, ScalarUDF}; use datafusion_physical_expr::expressions::{Column, Literal, col}; fn assert_cast_expr(expr: &Arc) -> &CastExpr { @@ -648,6 +759,88 @@ mod tests { assert_eq!(inner_col.index(), index); } + fn file_row_index_expr() -> Arc { + Arc::new(ScalarFunctionExpr::new( + "file_row_index", + Arc::new(ScalarUDF::from(FileRowIndexFunc::new())), + vec![], + Arc::new(Field::new("file_row_index", DataType::Int64, true)), + Arc::new(ConfigOptions::default()), + )) + } + + #[test] + fn test_rewrite_scalar_udf_replaces_nested_typed_udf() -> Result<()> { + let expr = Arc::new(expressions::BinaryExpr::new( + file_row_index_expr(), + Operator::Plus, + expressions::lit(ScalarValue::Int64(Some(1))), + )) as Arc; + + let rewritten = rewrite_scalar_udf::(expr, |_| { + Ok(expressions::lit(ScalarValue::Int64(Some(7)))) + })?; + + let binary = rewritten + .downcast_ref::() + .expect("rewritten expression should remain binary"); + assert_eq!(binary.op(), &Operator::Plus); + + let left = binary + .left() + .downcast_ref::() + .expect("left side should be rewritten to a literal"); + assert_eq!(left.value(), &ScalarValue::Int64(Some(7))); + + let right = binary + .right() + .downcast_ref::() + .expect("right side should remain the original literal"); + assert_eq!(right.value(), &ScalarValue::Int64(Some(1))); + Ok(()) + } + + #[test] + fn test_rewrite_file_row_index_expr_to_source_column() -> Result<()> { + let expr = rewrite_file_row_index_expr( + file_row_index_expr(), + "__datafusion_file_row_index", + 2, + )?; + + let cast_expr = expr + .downcast_ref::() + .expect("file row index expression should be a cast"); + assert_eq!(cast_expr.cast_type(), &DataType::Int64); + let target_field = cast_expr.target_field(); + assert_eq!(target_field.name(), "file_row_index"); + assert_eq!(target_field.data_type(), &DataType::Int64); + assert!(target_field.is_nullable()); + assert!(target_field.metadata().is_empty()); + + let source = cast_expr + .expr() + .downcast_ref::() + .expect("source column"); + assert_eq!(source.name(), "__datafusion_file_row_index"); + assert_eq!(source.index(), 2); + + let input_schema = Schema::new(vec![ + Field::new("value", DataType::Int64, true), + Field::new("__datafusion_file_row_index", DataType::Int64, false) + .with_metadata(HashMap::from([( + "source".to_string(), + "virtual".to_string(), + )])), + ]); + let return_field = expr.return_field(&input_schema)?; + assert_eq!(return_field.name(), "file_row_index"); + assert_eq!(return_field.data_type(), &DataType::Int64); + assert!(return_field.is_nullable()); + assert!(return_field.metadata().is_empty()); + Ok(()) + } + fn stale_index_cast_schemas() -> (SchemaRef, SchemaRef) { let physical_schema = Arc::new(Schema::new(vec![ Field::new("b", DataType::Binary, true), diff --git a/datafusion/physical-expr/src/projection.rs b/datafusion/physical-expr/src/projection.rs index cee95685e8440..1f6a6eb08fb78 100644 --- a/datafusion/physical-expr/src/projection.rs +++ b/datafusion/physical-expr/src/projection.rs @@ -661,7 +661,7 @@ impl ProjectionExprs { for proj_expr in self.exprs.iter() { let expr = &proj_expr.expr; let col_stats = if let Some(col) = expr.downcast_ref::() { - stats.column_statistics[col.index()].clone() + column_statistics_at(&stats.column_statistics, col.index()) } else if let Some(literal) = expr.downcast_ref::() { // Handle literal expressions (constants) by calculating proper statistics let data_type = expr.data_type(output_schema)?; @@ -725,6 +725,60 @@ impl ProjectionExprs { stats.column_statistics = column_statistics; Ok(stats) } + + /// Returns the output position of `column` if this projection contains it. + /// + /// This only matches projection expressions that are exactly [`Column`] expressions. + /// Computed expressions, even if they reference `column`, do not match. The + /// comparison uses [`Column`] equality, so both the name and index must match. + /// If the same column appears more than once, this returns the first matching + /// position. + /// + /// # Example + /// + /// ```rust + /// use datafusion_common::ScalarValue; + /// use datafusion_physical_expr::expressions::{Column, Literal}; + /// use datafusion_physical_expr::projection::{ProjectionExpr, ProjectionExprs}; + /// use std::sync::Arc; + /// + /// let projection = ProjectionExprs::new([ + /// ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b"), + /// ProjectionExpr::new( + /// Arc::new(Literal::new(ScalarValue::Int32(Some(42)))), + /// "answer", + /// ), + /// ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a"), + /// ]); + /// + /// assert_eq!( + /// projection.projected_column_position(&Column::new("b", 1)), + /// Some(0) + /// ); + /// assert_eq!( + /// projection.projected_column_position(&Column::new("a", 0)), + /// Some(2) + /// ); + /// + /// // The literal projection is not a Column expression. + /// assert_eq!( + /// projection.projected_column_position(&Column::new("answer", 1)), + /// None + /// ); + /// + /// // Columns not present in the projection also return None. + /// assert_eq!( + /// projection.projected_column_position(&Column::new("c", 2)), + /// None + /// ); + /// ``` + pub fn projected_column_position(&self, column: &Column) -> Option { + self.iter().position(|expr| { + expr.expr + .downcast_ref::() + .is_some_and(|projected| projected == column) + }) + } } /// Propagate column statistics through CAST projections. Other expressions @@ -736,7 +790,7 @@ fn project_column_statistics_through_expr( column_stats: &[ColumnStatistics], ) -> ColumnStatistics { if let Some(col) = expr.downcast_ref::() { - return column_stats[col.index()].clone(); + return column_statistics_at(column_stats, col.index()); } let Some(cast_expr) = expr.downcast_ref::() else { return ColumnStatistics::new_unknown(); @@ -760,6 +814,16 @@ fn project_column_statistics_through_expr( } } +fn column_statistics_at( + column_stats: &[ColumnStatistics], + index: usize, +) -> ColumnStatistics { + column_stats + .get(index) + .cloned() + .unwrap_or_else(ColumnStatistics::new_unknown) +} + impl<'a> IntoIterator for &'a ProjectionExprs { type Item = &'a ProjectionExpr; type IntoIter = std::slice::Iter<'a, ProjectionExpr>; @@ -2202,6 +2266,43 @@ pub(crate) mod tests { Schema::new(vec![field_0, field_1, field_2]) } + #[test] + fn test_projected_column_position_returns_output_position() { + let projection = ProjectionExprs::new([ + ProjectionExpr::new(Arc::new(Column::new("col2", 2)), "col2"), + ProjectionExpr::new(Arc::new(Column::new("col0", 0)), "col0"), + ]); + + assert_eq!( + projection.projected_column_position(&Column::new("col2", 2)), + Some(0) + ); + assert_eq!( + projection.projected_column_position(&Column::new("col0", 0)), + Some(1) + ); + } + + #[test] + fn test_projected_column_position_returns_none_for_non_column_or_missing() { + let projection = ProjectionExprs::new([ + ProjectionExpr::new( + Arc::new(Literal::new(ScalarValue::Int64(Some(42)))), + "col1", + ), + ProjectionExpr::new(Arc::new(Column::new("col0", 0)), "col0"), + ]); + + assert_eq!( + projection.projected_column_position(&Column::new("col1", 1)), + None + ); + assert_eq!( + projection.projected_column_position(&Column::new("col2", 2)), + None + ); + } + #[test] fn test_stats_projection_columns_only() { let source = get_stats(); @@ -2913,6 +3014,57 @@ pub(crate) mod tests { byte_size: Precision::Absent, } ); + + Ok(()) + } + + #[test] + fn test_project_statistics_missing_column_stats_are_unknown() -> Result<()> { + let mut input_stats = get_stats(); + let input_schema = get_schema(); + input_stats.column_statistics.truncate(2); + + // The schema has col2, but the statistics do not. This can happen for + // source-provided virtual columns that are available at execution time + // but not represented in file-level statistics. + let projection = ProjectionExprs::new(vec![ + ProjectionExpr { + expr: Arc::new(Column::new("col2", 2)), + alias: "virtual_col".to_string(), + }, + ProjectionExpr { + expr: Arc::new(CastExpr::new( + Arc::new(Column::new("col2", 2)), + DataType::Float64, + None, + )), + alias: "casted_virtual_col".to_string(), + }, + ProjectionExpr { + expr: Arc::new(Column::new("col0", 0)), + alias: "physical_col".to_string(), + }, + ]); + + let output_stats = projection.project_statistics( + input_stats, + &projection.project_schema(&input_schema)?, + )?; + + assert_eq!(output_stats.column_statistics.len(), 3); + assert_eq!( + output_stats.column_statistics[0], + ColumnStatistics::new_unknown() + ); + assert_eq!( + output_stats.column_statistics[1], + ColumnStatistics::new_unknown() + ); + assert_eq!( + output_stats.column_statistics[2].max_value, + Precision::Exact(ScalarValue::Int64(Some(21))) + ); + Ok(()) } diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 6e73c4d2e0157..aa624e050c9d2 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -844,6 +844,14 @@ impl HashJoinExec { return false; } + // Bounds and membership filters derived from the build side do not + // account for null-equal matching: a probe-side NULL key evaluates + // such predicates to NULL and would be pruned, even though it can + // match a build-side NULL when nulls compare equal. + if self.null_equality == NullEquality::NullEqualsNull { + return false; + } + // `preserve_file_partitions` can report Hash partitioning for Hive-style // file groups, but those partitions are not actually hash-distributed. // Partitioned dynamic filters rely on hash routing, so disable them in @@ -6417,6 +6425,35 @@ mod tests { Ok(()) } + #[test] + fn test_dynamic_filter_pushdown_rejects_null_equal_join() -> Result<()> { + let (_, _, on) = build_schema_and_on()?; + let left = build_table(("a1", &vec![1]), ("b1", &vec![1]), ("c1", &vec![1])); + let right = build_table(("a2", &vec![1]), ("b1", &vec![1]), ("c2", &vec![1])); + + let mut session_config = SessionConfig::default(); + session_config + .options_mut() + .optimizer + .enable_join_dynamic_filter_pushdown = true; + + let join = HashJoinExec::try_new( + left, + right, + on, + None, + &JoinType::RightSemi, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNull, + false, + )?; + + assert!(!join.allow_join_dynamic_filter_pushdown(session_config.options())); + + Ok(()) + } + #[test] fn test_with_dynamic_filter_rejects_invalid_columns() -> Result<()> { let (_, _, on) = build_schema_and_on()?; diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs b/datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs index ad7312426bd18..99aef6ed82a36 100644 --- a/datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs +++ b/datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs @@ -119,8 +119,6 @@ //! factor than the pair-materialization approach. use std::cmp::Ordering; -use std::fs::File; -use std::io::BufReader; use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; @@ -134,7 +132,6 @@ use crate::{EmptyRecordBatchStream, RecordBatchStream}; use arrow::array::{Array, ArrayRef, BooleanArray, BooleanBufferBuilder, RecordBatch}; use arrow::compute::{BatchCoalescer, SortOptions, filter_record_batch, not}; use arrow::datatypes::SchemaRef; -use arrow::ipc::reader::StreamReader; use arrow::util::bit_chunk_iterator::UnalignedBitChunk; use arrow::util::bit_util::apply_bitwise_binary_op; use datafusion_common::{ @@ -259,6 +256,13 @@ pub(crate) struct BitwiseSortMergeJoinStream { inner_key_buffer: Vec, inner_key_spill: Option, + // Track the active spill_stream + spill_stream: Option, + // Whether the active spill stream has produced any batches yet. + spill_stream_has_data: bool, + // Prevents wiping out the buffer if we yield while evaluating the filter + inner_group_buffered: bool, + // True when buffer_inner_key_group returned Pending after partially // filling inner_key_buffer. On re-entry, buffer_inner_key_group // must skip clear() and resume from poll_next_inner_batch (the @@ -371,6 +375,9 @@ impl BitwiseSortMergeJoinStream { matched: BooleanBufferBuilder::new(0), inner_key_buffer: vec![], inner_key_spill: None, + spill_stream: None, + spill_stream_has_data: false, + inner_group_buffered: false, buffering_inner_pending: false, pending_boundary: None, on_outer, @@ -468,6 +475,9 @@ impl BitwiseSortMergeJoinStream { fn clear_inner_key_group(&mut self) { self.inner_key_buffer.clear(); self.inner_key_spill = None; + self.spill_stream = None; + self.spill_stream_has_data = false; + self.inner_group_buffered = false; self.inner_buffer_size = 0; } @@ -749,7 +759,10 @@ impl BitwiseSortMergeJoinStream { /// Process a key match with a filter. For each inner row in the buffered /// key group, evaluates the filter against the outer key group and ORs /// the results into the matched bitset using u64-chunked bitwise ops. - fn process_key_match_with_filter(&mut self) -> Result<()> { + fn process_key_match_with_filter( + &mut self, + cx: &mut Context<'_>, + ) -> Poll> { self.get_outer_self_cmp()?; let filter = self.filter.as_ref().unwrap(); let outer_batch = self.outer_batch.as_ref().unwrap(); @@ -785,24 +798,47 @@ impl BitwiseSortMergeJoinStream { ) .count_ones(); - // Process spilled inner batches first (read back from disk). - if let Some(spill_file) = &self.inner_key_spill { - let file = BufReader::new(File::open(spill_file.path())?); - let reader = StreamReader::try_new(file, None)?; - for batch_result in reader { - let inner_slice = batch_result?; - matched_count = eval_filter_for_inner_slice( - self.outer_is_left, - filter, - &outer_slice, - &inner_slice, - &mut self.matched, - self.outer_offset, - outer_group_len, - matched_count, - )?; - if matched_count == outer_group_len { - break; + // Process spilled inner batches first asynchronously. + if matched_count < outer_group_len + && (self.inner_key_spill.is_some() || self.spill_stream.is_some()) + { + if self.spill_stream.is_none() + && let Some(spill_file) = &self.inner_key_spill + { + let stream = self + .spill_manager + .read_spill_as_stream(spill_file.clone(), None)?; + self.spill_stream = Some(stream); + } + + while matched_count < outer_group_len { + let stream = self.spill_stream.as_mut().unwrap(); + match ready!(stream.poll_next_unpin(cx)) { + Some(Ok(inner_slice)) => { + self.spill_stream_has_data = true; + matched_count = eval_filter_for_inner_slice( + self.outer_is_left, + filter, + &outer_slice, + &inner_slice, + &mut self.matched, + self.outer_offset, + outer_group_len, + matched_count, + )?; + } + Some(Err(e)) => { + self.spill_stream = None; + self.spill_stream_has_data = false; + return Poll::Ready(Err(e)); + } + None => { + self.spill_stream = None; + if !self.spill_stream_has_data { + return Poll::Ready(internal_err!("Spill file was empty")); + } + break; + } } } } @@ -830,13 +866,17 @@ impl BitwiseSortMergeJoinStream { } self.outer_offset = outer_group_end; - Ok(()) + + self.spill_stream = None; + self.spill_stream_has_data = false; + + Poll::Ready(Ok(())) } /// Continue processing an outer key group that spans multiple outer /// batches. Returns `true` if this outer batch was fully consumed /// by the key group and the caller should load another. - fn resume_boundary(&mut self) -> Result { + fn resume_boundary(&mut self, cx: &mut Context<'_>) -> Poll> { debug_assert!( self.outer_batch.is_some(), "caller must load outer_batch first" @@ -858,7 +898,7 @@ impl BitwiseSortMergeJoinStream { }); self.emit_outer_batch()?; self.outer_batch = None; - return Ok(true); + return Poll::Ready(Ok(true)); } } } @@ -874,7 +914,15 @@ impl BitwiseSortMergeJoinStream { self.null_equality, )?; if same_key { - self.process_key_match_with_filter()?; + match self.process_key_match_with_filter(cx) { + Poll::Ready(Ok(())) => (), + Poll::Ready(Err(e)) => return Poll::Ready(Err(e)), + Poll::Pending => { + self.pending_boundary = + Some(PendingBoundary::Filtered { saved_keys }); + return Poll::Pending; + } + } let num_outer = self.outer_batch.as_ref().unwrap().num_rows(); if self.outer_offset >= num_outer { self.pending_boundary = Some(PendingBoundary::Filtered { @@ -882,14 +930,63 @@ impl BitwiseSortMergeJoinStream { }); self.emit_outer_batch()?; self.outer_batch = None; - return Ok(true); + return Poll::Ready(Ok(true)); } } self.clear_inner_key_group(); } None => {} } - Ok(false) + Poll::Ready(Ok(false)) + } + + /// Helper to process an Equal match across potential outer batch boundaries. + fn process_filtered_match_loop(&mut self, cx: &mut Context<'_>) -> Poll> { + loop { + ready!(self.process_key_match_with_filter(cx))?; + + let outer_batch = self.outer_batch.as_ref().unwrap(); + if self.outer_offset >= outer_batch.num_rows() { + let saved_keys = + slice_keys(&self.outer_key_arrays, outer_batch.num_rows() - 1); + + self.emit_outer_batch()?; + self.pending_boundary = Some(PendingBoundary::Filtered { saved_keys }); + + // Clear stale batch before polling + self.outer_batch = None; + + match ready!(self.poll_next_outer_batch(cx)) { + Err(e) => return Poll::Ready(Err(e)), + Ok(false) => { + self.pending_boundary = None; + break; + } + Ok(true) => { + let Some(PendingBoundary::Filtered { saved_keys }) = + self.pending_boundary.take() + else { + unreachable!() + }; + let same = keys_match( + &saved_keys, + &self.outer_key_arrays, + &self.sort_options, + self.null_equality, + )?; + if same { + continue; + } + break; + } + } + } else { + break; + } + } + + self.clear_inner_key_group(); // This resets inner_group_buffered to false + Poll::Ready(Ok(())) } /// Main loop: drive the merge-scan to produce output batches. @@ -911,14 +1008,21 @@ impl BitwiseSortMergeJoinStream { } return Poll::Ready(Ok(None)); } - Ok(true) => { - if self.resume_boundary()? { - continue; - } - } + Ok(true) => {} // Loaded batch, move on to checks } } + // Handles pausing while fetching a NEW outer batch. + if self.pending_boundary.is_some() && ready!(self.resume_boundary(cx))? { + continue; + } + + // Handles pausing while reading the disk stream mid-batch. + if self.inner_group_buffered { + ready!(self.process_filtered_match_loop(cx))?; + continue; + } + // 2. Ensure we have an inner batch (unless inner is exhausted). // Skip this when resuming a pending boundary — inner was already // advanced past the key group before the boundary loop started. @@ -1043,65 +1147,17 @@ impl BitwiseSortMergeJoinStream { } Ordering::Equal => { if self.filter.is_some() { + debug_assert!(!self.inner_group_buffered); // Buffer inner key group (may span batches) match ready!(self.buffer_inner_key_group(cx)) { Err(e) => return Poll::Ready(Err(e)), - Ok(_inner_exhausted) => {} + Ok(_inner_exhausted) => { + self.inner_group_buffered = true; + } } - // Process outer rows against buffered inner group // (may need to handle outer batch boundary) - loop { - self.process_key_match_with_filter()?; - - let outer_batch = self.outer_batch.as_ref().unwrap(); - if self.outer_offset >= outer_batch.num_rows() { - let saved_keys = slice_keys( - &self.outer_key_arrays, - outer_batch.num_rows() - 1, - ); - - self.emit_outer_batch()?; - debug_assert!( - !self.inner_key_buffer.is_empty() - || self.inner_key_spill.is_some(), - "Filtered pending boundary requires inner key data in buffer or spill" - ); - self.pending_boundary = - Some(PendingBoundary::Filtered { saved_keys }); - - match ready!(self.poll_next_outer_batch(cx)) { - Err(e) => return Poll::Ready(Err(e)), - Ok(false) => { - self.pending_boundary = None; - self.outer_batch = None; - break; - } - Ok(true) => { - let Some(PendingBoundary::Filtered { - saved_keys, - }) = self.pending_boundary.take() - else { - unreachable!() - }; - let same = keys_match( - &saved_keys, - &self.outer_key_arrays, - &self.sort_options, - self.null_equality, - )?; - if same { - continue; - } - break; - } - } - } else { - break; - } - } - - self.clear_inner_key_group(); + ready!(self.process_filtered_match_loop(cx))?; } else { // No filter: advance inner past key group, then // mark all outer rows with this key as matched. diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs b/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs index 9bcc749c23dce..f1a18aac762f5 100644 --- a/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs +++ b/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs @@ -23,8 +23,7 @@ use std::cmp::Ordering; use std::collections::{HashMap, VecDeque}; -use std::fs::File; -use std::io::BufReader; +use std::fmt::Debug; use std::mem::size_of; use std::ops::Range; use std::pin::Pin; @@ -47,10 +46,9 @@ use crate::{PhysicalExpr, RecordBatchStream, SendableRecordBatchStream}; use arrow::array::{types::UInt64Type, *}; use arrow::compute::{ self, BatchCoalescer, SortOptions, concat_batches, filter_record_batch, interleave, - take, take_arrays, + take_arrays, }; use arrow::datatypes::SchemaRef; -use arrow::ipc::reader::StreamReader; use datafusion_common::cast::as_uint64_array; use datafusion_common::{JoinType, NullEquality, Result, exec_err, internal_err}; use datafusion_execution::disk_manager::RefCountedTempFile; @@ -58,7 +56,7 @@ use datafusion_execution::memory_pool::MemoryReservation; use datafusion_execution::runtime_env::RuntimeEnv; use datafusion_physical_expr_common::physical_expr::PhysicalExprRef; -use futures::{Stream, StreamExt}; +use futures::{Stream, StreamExt, ready}; /// State of SMJ stream #[derive(Debug, PartialEq, Eq)] @@ -382,6 +380,11 @@ pub(super) struct MaterializingSortMergeJoinStream { /// Manages the process of spilling and reading back intermediate data pub spill_manager: SpillManager, + /// Tracks the active stream when loading spilled buffered batches back in memory + pub spill_stream: Option, + /// Tracks the number of batches currently spilled + pub spilled_batch_count: usize, + // ======================================================================== // CACHED COMPARATORS: // Pre-built comparators to avoid per-row type dispatch in hot loops. @@ -599,6 +602,16 @@ impl Stream for MaterializingSortMergeJoinStream { .filter_mask .len(); if accumulated >= self.batch_size { + // Ensure required spilled batches are restored to memory + // before processing, as this path invokes freeze_all(). + let needed = self.get_required_batch_indices( + self.buffered_data.batches.len(), + ); + if let Err(e) = ready!( + self.poll_spilled_batches(cx, &needed) + ) { + return Poll::Ready(Some(Err(e))); + } match self.process_filtered_batches()? { Poll::Ready(Some(batch)) => { return Poll::Ready(Some(Ok(batch))); @@ -684,14 +697,13 @@ impl Stream for MaterializingSortMergeJoinStream { self.state = SortMergeJoinState::Init; } SortMergeJoinState::JoinOutput => { - self.join_partial()?; + // If the batch size limit is reached, restore required spilled batches to memory and freeze. + // Guarding at the top of the loop safely handles re-entry from Poll::Pending. + if self.num_unfrozen_pairs() >= self.batch_size { + let needed = self + .get_required_batch_indices(self.buffered_data.batches.len()); + ready!(self.poll_spilled_batches(cx, &needed))?; - if self.num_unfrozen_pairs() < self.batch_size { - if self.buffered_data.scanning_finished() { - self.buffered_data.scanning_reset(); - self.state = SortMergeJoinState::EmitReadyThenInit; - } - } else { self.freeze_all()?; // Verify metadata alignment before checking if we have batches to output @@ -705,7 +717,6 @@ impl Stream for MaterializingSortMergeJoinStream { } // For non-filtered joins, only output if we have a completed batch - // (opportunistic output when target batch size is reached) if self .joined_record_batches .joined_batches @@ -720,10 +731,26 @@ impl Stream for MaterializingSortMergeJoinStream { .record_output(&self.join_metrics.baseline_metrics()); return Poll::Ready(Some(Ok(record_batch))); } + // Otherwise keep buffering (don't output yet) + continue; + } + + self.join_partial()?; + + if self.num_unfrozen_pairs() < self.batch_size + && self.buffered_data.scanning_finished() + { + self.buffered_data.scanning_reset(); + self.state = SortMergeJoinState::EmitReadyThenInit; } + // Note: If join_partial() reached the batch size, the loop repeats to freeze the data. } SortMergeJoinState::Exhausted => { + let needed = + self.get_required_batch_indices(self.buffered_data.batches.len()); + ready!(self.poll_spilled_batches(cx, &needed))?; + self.freeze_all()?; // Verify metadata alignment before final output @@ -843,6 +870,8 @@ impl MaterializingSortMergeJoinStream { reservation, runtime_env, spill_manager, + spill_stream: None, + spilled_batch_count: 0, streamed_buffered_cmp: None, buffered_equality_cmp: None, streamed_batch_counter: AtomicUsize::new(0), @@ -917,6 +946,84 @@ impl MaterializingSortMergeJoinStream { Poll::Pending } + /// Identifies which buffered batches are needed for the upcoming freeze operation + fn get_required_batch_indices(&self, buffered_freeze_count: usize) -> Vec { + let mut needed = vec![]; + // Avoid scanning if no spilled batches exist + if self.spilled_batch_count == 0 { + return needed; + } + // We need all batches that matched with streamed rows + for chunk in &self.streamed_batch.output_indices { + if let Some(idx) = chunk.buffered_batch_idx { + needed.push(idx); + } + } + + // Full Joins need to emit null-joined rows, so we need batches up to freeze_count + if self.join_type == JoinType::Full { + needed.extend(0..buffered_freeze_count); + } + + needed.sort_unstable(); + needed.dedup(); + needed + } + + /// Asynchronously reads spilled batches back into memory. + /// Only processes the required indices to avoid OOMs. + fn poll_spilled_batches( + &mut self, + cx: &mut Context<'_>, + required_indices: &[usize], + ) -> Poll> { + for &idx in required_indices { + // Guard against indices that might be out of bounds if the queue was cleared + if idx >= self.buffered_data.batches.len() { + continue; + } + + let bb = &mut self.buffered_data.batches[idx]; + + if let BufferedBatchState::Spilled(spill_file) = &bb.batch { + if self.spill_stream.is_none() { + let stream = self + .spill_manager + .read_spill_as_stream(spill_file.clone(), None)?; + self.spill_stream = Some(stream); + } + + match ready!(self.spill_stream.as_mut().unwrap().poll_next_unpin(cx)) { + Some(Ok(batch)) => { + // Transition the batch back to InMemory + bb.batch = BufferedBatchState::InMemory(batch); + self.spilled_batch_count -= 1; + // The batch is back in memory, so we must account for its size. + let newly_allocated = + bb.size_estimation.saturating_sub(bb.reserved_amount); + self.reservation.grow(newly_allocated); + bb.reserved_amount = bb.size_estimation; + + self.join_metrics + .peak_mem_used() + .set_max(self.reservation.size()); + + self.spill_stream = None; + } + Some(Err(e)) => { + self.spill_stream = None; + return Poll::Ready(Err(e)); + } + None => { + self.spill_stream = None; + return Poll::Ready(internal_err!("Spill file was empty")); + } + } + } + } + Poll::Ready(Ok(())) + } + /// Poll next streamed row fn poll_streamed_row(&mut self, cx: &mut Context) -> Poll>> { loop { @@ -931,33 +1038,41 @@ impl MaterializingSortMergeJoinStream { self.streamed_state = StreamedState::Polling; } } - StreamedState::Polling => match self.streamed.poll_next_unpin(cx)? { - Poll::Pending => { - return Poll::Pending; - } - Poll::Ready(None) => { - // Release the streamed input pipeline's resources. - let streamed_schema = self.streamed.schema(); - self.streamed = - Box::pin(EmptyRecordBatchStream::new(streamed_schema)); - self.streamed_state = StreamedState::Exhausted; + StreamedState::Polling => { + let needed = + self.get_required_batch_indices(self.buffered_data.batches.len()); + if let Err(e) = ready!(self.poll_spilled_batches(cx, &needed)) { + return Poll::Ready(Some(Err(e))); } - Poll::Ready(Some(batch)) => { - if batch.num_rows() > 0 { - self.freeze_streamed()?; - self.join_metrics.input_batches().add(1); - self.join_metrics.input_rows().add(batch.num_rows()); - self.streamed_batch = - StreamedBatch::new(batch, &self.on_streamed); - self.rebuild_streamed_buffered_cmp()?; - // Every incoming streaming batch should have its unique id - // Check `JoinedRecordBatches.self.streamed_batch_counter` documentation - self.streamed_batch_counter - .fetch_add(1, std::sync::atomic::Ordering::SeqCst); - self.streamed_state = StreamedState::Ready; + + match self.streamed.poll_next_unpin(cx)? { + Poll::Pending => { + return Poll::Pending; + } + Poll::Ready(None) => { + // Release the streamed input pipeline's resources. + let streamed_schema = self.streamed.schema(); + self.streamed = + Box::pin(EmptyRecordBatchStream::new(streamed_schema)); + self.streamed_state = StreamedState::Exhausted; + } + Poll::Ready(Some(batch)) => { + if batch.num_rows() > 0 { + self.freeze_streamed()?; + self.join_metrics.input_batches().add(1); + self.join_metrics.input_rows().add(batch.num_rows()); + self.streamed_batch = + StreamedBatch::new(batch, &self.on_streamed); + self.rebuild_streamed_buffered_cmp()?; + // Every incoming streaming batch should have its unique id + // Check `JoinedRecordBatches.self.streamed_batch_counter` documentation + self.streamed_batch_counter + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + self.streamed_state = StreamedState::Ready; + } } } - }, + } StreamedState::Ready => { return Poll::Ready(Some(Ok(()))); } @@ -997,6 +1112,7 @@ impl MaterializingSortMergeJoinStream { .unwrap(); // Operation only return None if no batches are spilled, here we ensure that at least one batch is spilled buffered_batch.batch = BufferedBatchState::Spilled(spill_file); + self.spilled_batch_count += 1; // Join key arrays remain in memory after the batch is // spilled — the comparator needs them for key boundary @@ -1036,12 +1152,25 @@ impl MaterializingSortMergeJoinStream { let head_batch = self.buffered_data.head_batch(); // If the head batch is fully processed, dequeue it and produce output of it. if head_batch.range.end == head_batch.num_rows { + // load the spilled head batch before dequeuing + let needed = self.get_required_batch_indices(1); + if let Err(e) = ready!(self.poll_spilled_batches(cx, &needed)) + { + return Poll::Ready(Some(Err(e))); + } + self.freeze_dequeuing_buffered()?; if let Some(mut buffered_batch) = self.buffered_data.batches.pop_front() { self.produce_buffered_not_matched(&mut buffered_batch)?; self.free_reservation(&buffered_batch); + if matches!( + buffered_batch.batch, + BufferedBatchState::Spilled(_) + ) { + self.spilled_batch_count -= 1; + } head_changed = true; } } else { @@ -1556,18 +1685,6 @@ impl MaterializingSortMergeJoinStream { as_uint64_array(&compute::concat(&refs)?)?.clone() }; - let spill_reservation = self.reservation.new_empty(); - if matches!( - &self.buffered_data.batches[first_batch_idx].batch, - BufferedBatchState::Spilled(_) - ) { - spill_reservation - .grow(self.buffered_data.batches[first_batch_idx].size_estimation); - self.join_metrics - .peak_mem_used() - .set_max(self.reservation.size() + spill_reservation.size()); - } - return fetch_right_columns_by_idxs( &self.buffered_data, first_batch_idx, @@ -1603,29 +1720,20 @@ impl MaterializingSortMergeJoinStream { let num_right_cols = self.buffered_schema.fields().len(); // Read each source batch once (spilled batches require disk I/O). - // Track memory for each spilled batch at the point of deserialization - // so the pool reflects actual usage as it grows. - let spill_reservation = self.reservation.new_empty(); - let mut source_data: Vec> = - Vec::with_capacity(source_batches.len()); - for &idx in &source_batches { - let bb = &self.buffered_data.batches[idx]; - match &bb.batch { - BufferedBatchState::InMemory(batch) => { - source_data.push(Some(batch.clone())); - } - BufferedBatchState::Spilled(spill_file) => { - spill_reservation.grow(bb.size_estimation); - self.join_metrics - .peak_mem_used() - .set_max(self.reservation.size() + spill_reservation.size()); - - let file = BufReader::new(File::open(spill_file.path())?); - let reader = StreamReader::try_new(file, None)?; - source_data.push(reader.into_iter().next().transpose()?); + let source_data_result: Result> = source_batches + .iter() + .map(|&idx| { + let bb = &self.buffered_data.batches[idx]; + match &bb.batch { + BufferedBatchState::InMemory(batch) => Ok(batch.clone()), + BufferedBatchState::Spilled(_) => { + internal_err!("Buffered batch should have been unspilled before fetching columns") + } } - } - } + }) + .collect(); + + let source_data = source_data_result?; let mut right_columns = Vec::with_capacity(num_right_cols); for col_idx in 0..num_right_cols { @@ -1637,14 +1745,7 @@ impl MaterializingSortMergeJoinStream { source_arrays.push(null_array.as_ref()); for data in &source_data { - match data { - Some(batch) => source_arrays.push(batch.column(col_idx).as_ref()), - None => { - return internal_err!( - "Failed to read spilled buffered batch during interleave" - ); - } - } + source_arrays.push(data.column(col_idx).as_ref()); } right_columns.push(interleave(&source_arrays, &interleave_indices)?); } @@ -1838,32 +1939,17 @@ fn fetch_right_columns_from_batch_by_idxs( buffered_indices: &UInt64Array, ) -> Result> { match &buffered_batch.batch { - // In memory batch - // In memory batch BufferedBatchState::InMemory(batch) => { - // When indices form a contiguous range (common in SMJ since the - // buffered side is scanned sequentially), use zero-copy slice. if let Some(range) = is_contiguous_range(buffered_indices) { Ok(batch.slice(range.start, range.len()).columns().to_vec()) } else { Ok(take_arrays(batch.columns(), buffered_indices, None)?) } } - // If the batch was spilled to disk, less likely - BufferedBatchState::Spilled(spill_file) => { - let mut buffered_cols: Vec = - Vec::with_capacity(buffered_indices.len()); - - let file = BufReader::new(File::open(spill_file.path())?); - let reader = StreamReader::try_new(file, None)?; - - for batch in reader { - batch?.columns().iter().for_each(|column| { - buffered_cols.extend(take(column, &buffered_indices, None)) - }); - } - - Ok(buffered_cols) + BufferedBatchState::Spilled(_) => { + internal_err!( + "Buffered batch should have been unspilled before fetching columns" + ) } } } diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs b/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs index b1fdf3ddabb5a..0347299dd0094 100644 --- a/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs +++ b/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs @@ -2459,6 +2459,24 @@ async fn overallocation_multi_batch_spill() -> Result<()> { assert!(join.metrics().unwrap().spilled_bytes().unwrap() > 0); assert!(join.metrics().unwrap().spilled_rows().unwrap() > 0); + // For Full joins, get_required_batch_indices extends 0..batches.len(), so + // poll_spilled_batches can restore all spilled batches at once via infallible + // grow(). Verify accounting tracked the transient spike and cleaned up. + let peak_mem = join + .metrics() + .and_then(|m| m.sum_by_name("peak_mem_used")) + .map(|m| m.as_usize()) + .unwrap_or(0); + assert!( + peak_mem > 0, + "peak_mem_used should be > 0 for {join_type:?} batch_size={batch_size}" + ); + assert_eq!( + runtime.memory_pool.reserved(), + 0, + "memory should be fully released after {join_type:?} completes + (batch_size={batch_size}): infallible grow during restore must be balanced" + ); // Run the test with no spill configuration as let task_ctx_no_spill = TaskContext::default().with_session_config(session_config.clone()); diff --git a/datafusion/proto-common/gen/Cargo.toml b/datafusion/proto-common/gen/Cargo.toml index f0e60819d42a8..0cbba311b2c3f 100644 --- a/datafusion/proto-common/gen/Cargo.toml +++ b/datafusion/proto-common/gen/Cargo.toml @@ -38,4 +38,4 @@ workspace = true [dependencies] # Pin these dependencies so that the generated output is deterministic pbjson-build = "=0.9.0" -prost-build = "=0.14.3" +prost-build = "=0.14.4" diff --git a/datafusion/proto-models/gen/Cargo.toml b/datafusion/proto-models/gen/Cargo.toml index 8b48dfe70e6c7..9724b63cccf3c 100644 --- a/datafusion/proto-models/gen/Cargo.toml +++ b/datafusion/proto-models/gen/Cargo.toml @@ -38,4 +38,4 @@ workspace = true [dependencies] # Pin these dependencies so that the generated output is deterministic pbjson-build = "=0.9.0" -prost-build = "=0.14.3" +prost-build = "=0.14.4" diff --git a/datafusion/sqllogictest/test_files/file_row_index.slt b/datafusion/sqllogictest/test_files/file_row_index.slt new file mode 100644 index 0000000000000..38822bebfdfd3 --- /dev/null +++ b/datafusion/sqllogictest/test_files/file_row_index.slt @@ -0,0 +1,171 @@ +# 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. + +statement ok +COPY (VALUES (10), (20), (30), (40), (50)) +TO 'test_files/scratch/file_row_index/parquet_table/data.parquet' +STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE parquet_table(column1 int) +STORED AS PARQUET +LOCATION 'test_files/scratch/file_row_index/parquet_table/'; + +query TT +EXPLAIN SELECT file_row_index(), column1 FROM parquet_table +---- +logical_plan +01)Projection: file_row_index(), parquet_table.column1 +02)--TableScan: parquet_table projection=[column1] +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/file_row_index/parquet_table/data.parquet]]}, projection=[CAST(__datafusion_file_row_index@1 AS Int64) as file_row_index(), column1], file_type=parquet + + +query II +SELECT file_row_index(), column1 FROM parquet_table ORDER BY column1 +---- +0 10 +1 20 +2 30 +3 40 +4 50 + +query III +SELECT file_row_index(), file_row_index() + 1, column1 +FROM parquet_table +ORDER BY column1 +---- +0 1 10 +1 2 20 +2 3 30 +3 4 40 +4 5 50 + + +query II +SELECT file_row_index(), column1 +FROM parquet_table +WHERE file_row_index() > 2 +ORDER BY column1 +---- +3 40 +4 50 + +# Filter on file_row_index without having it in projection + +query TT +EXPLAIN SELECT column1 FROM parquet_table WHERE file_row_index() > 2 ORDER BY column1 +---- +logical_plan +01)Sort: parquet_table.column1 ASC NULLS LAST +02)--Projection: parquet_table.column1 +03)----Filter: __datafusion_extracted_1 > Int64(2) +04)------Projection: file_row_index() AS __datafusion_extracted_1, parquet_table.column1 +05)--------TableScan: parquet_table projection=[column1] +physical_plan +01)SortExec: expr=[column1@0 ASC NULLS LAST], preserve_partitioning=[false] +02)--FilterExec: __datafusion_extracted_1@0 > 2, projection=[column1@1] +03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/file_row_index/parquet_table/data.parquet]]}, projection=[CAST(__datafusion_file_row_index@1 AS Int64) as __datafusion_extracted_1, column1], file_type=parquet + +query I +SELECT column1 FROM parquet_table WHERE file_row_index() > 2 ORDER BY column1 +---- +40 +50 + +# Filter on file_row_index without projecting it, while enabling filter pushdown + +statement ok +SET datafusion.execution.parquet.pushdown_filters = true; + +query TT +EXPLAIN SELECT column1 FROM parquet_table WHERE file_row_index() > 2 ORDER BY column1 +---- +logical_plan +01)Sort: parquet_table.column1 ASC NULLS LAST +02)--Projection: parquet_table.column1 +03)----Filter: __datafusion_extracted_1 > Int64(2) +04)------Projection: file_row_index() AS __datafusion_extracted_1, parquet_table.column1 +05)--------TableScan: parquet_table projection=[column1] +physical_plan +01)SortExec: expr=[column1@0 ASC NULLS LAST], preserve_partitioning=[false] +02)--FilterExec: __datafusion_extracted_1@0 > 2, projection=[column1@1] +03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/file_row_index/parquet_table/data.parquet]]}, projection=[CAST(__datafusion_file_row_index@1 AS Int64) as __datafusion_extracted_1, column1], file_type=parquet + +query I +SELECT column1 FROM parquet_table WHERE file_row_index() > 2 ORDER BY column1 +---- +40 +50 + +statement ok +RESET datafusion.execution.parquet.pushdown_filters; + +# Without the rewrite in ParquetSource, `file_row_index()` errors because it +# depends on file-source context + +query error file_row_index\(\) is source dependent and cannot be evaluated directly +SELECT file_row_index() + +# Testing pushdown over a source that doesn't support `file_row_index()`. + +statement ok +COPY (VALUES (10), (20), (30), (40), (50)) +TO 'test_files/scratch/file_row_index/csv_table/data.csv' +STORED AS CSV; + +statement ok +CREATE EXTERNAL TABLE csv_table(column1 int) +STORED AS CSV +LOCATION 'test_files/scratch/file_row_index/csv_table/data.csv'; + +query error file_row_index\(\) is source dependent and cannot be evaluated directly +SELECT *, file_row_index() FROM csv_table; + +# Testing a table with two files. + +statement ok +COPY (VALUES (10), (20)) +TO 'test_files/scratch/file_row_index/parquet_two_files/part-1.parquet' +STORED AS PARQUET; + +statement ok +COPY (VALUES (30), (40)) +TO 'test_files/scratch/file_row_index/parquet_two_files/part-2.parquet' +STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE parquet_two_files(column1 int) +STORED AS PARQUET +LOCATION 'test_files/scratch/file_row_index/parquet_two_files/'; + +query II +SELECT file_row_index(), column1 +FROM parquet_two_files +WHERE file_row_index() = 1 +ORDER BY column1 +---- +1 20 +1 40 + +statement ok +DROP TABLE parquet_two_files; + +statement ok +DROP TABLE parquet_table; + +statement ok +DROP TABLE csv_table; diff --git a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt index 40bfe79dcc633..b5a06bc7cb313 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt @@ -1023,6 +1023,48 @@ statement ok drop table int_probe; +######## +# Dynamic filters must not be created for null-equal joins (IS NOT DISTINCT +# FROM, INTERSECT): min/max bounds and membership filters derived from the +# build side evaluate to NULL for probe-side NULL keys and would prune rows +# that can null-match a build-side NULL. +######## + +statement ok +COPY (SELECT * FROM (VALUES (11), (22), (NULL)) v(id)) TO 'test_files/scratch/push_down_filter_parquet/nej_probe.parquet' STORED AS PARQUET; + +statement ok +COPY (SELECT * FROM (VALUES (11), (NULL)) v(id)) TO 'test_files/scratch/push_down_filter_parquet/nej_build.parquet' STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE nej_probe STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_parquet/nej_probe.parquet'; + +statement ok +CREATE EXTERNAL TABLE nej_build STORED AS PARQUET LOCATION 'test_files/scratch/push_down_filter_parquet/nej_build.parquet'; + +# The probe-side NULL key must survive to match the build-side NULL +query II rowsort +SELECT nej_build.id, nej_probe.id FROM nej_build JOIN nej_probe ON nej_build.id IS NOT DISTINCT FROM nej_probe.id +---- +11 11 +NULL NULL + +# No DynamicFilter predicate may appear on the probe side of a null-equal join +query TT +EXPLAIN SELECT nej_build.id, nej_probe.id FROM nej_build JOIN nej_probe ON nej_build.id IS NOT DISTINCT FROM nej_probe.id +---- +physical_plan +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], NullsEqual: true +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nej_build.parquet]]}, projection=[id], file_type=parquet +03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nej_probe.parquet]]}, projection=[id], file_type=parquet + +statement ok +drop table nej_build; + +statement ok +drop table nej_probe; + + # Config reset statement ok RESET datafusion.explain.physical_plan_only; diff --git a/datafusion/wasmtest/datafusion-wasm-app/package-lock.json b/datafusion/wasmtest/datafusion-wasm-app/package-lock.json index c476ea76347ab..526853d841421 100644 --- a/datafusion/wasmtest/datafusion-wasm-app/package-lock.json +++ b/datafusion/wasmtest/datafusion-wasm-app/package-lock.json @@ -2431,14 +2431,13 @@ } }, "node_modules/launch-editor": { - "version": "2.10.0", - "resolved": "https://registry.npmjs.org/launch-editor/-/launch-editor-2.10.0.tgz", - "integrity": "sha512-D7dBRJo/qcGX9xlvt/6wUYzQxjh5G1RvZPgPv8vi4KRU99DVQL/oW7tnVOCCTm2HGeo3C5HvGE5Yrh6UBoZ0vA==", + "version": "2.14.1", + "resolved": "https://registry.npmjs.org/launch-editor/-/launch-editor-2.14.1.tgz", + "integrity": "sha512-QWBrQsMpH7gPr965dsKD/3cKWiNoTjpATQf++Xq63N6sKRGMwlVXz41O1IZTMfZQgBctD/K5Zt06+/I6pP6+HA==", "dev": true, - "license": "MIT", "dependencies": { - "picocolors": "^1.0.0", - "shell-quote": "^1.8.1" + "picocolors": "^1.1.1", + "shell-quote": "^1.8.4" } }, "node_modules/loader-runner": { @@ -6028,13 +6027,13 @@ "dev": true }, "launch-editor": { - "version": "2.10.0", - "resolved": "https://registry.npmjs.org/launch-editor/-/launch-editor-2.10.0.tgz", - "integrity": "sha512-D7dBRJo/qcGX9xlvt/6wUYzQxjh5G1RvZPgPv8vi4KRU99DVQL/oW7tnVOCCTm2HGeo3C5HvGE5Yrh6UBoZ0vA==", + "version": "2.14.1", + "resolved": "https://registry.npmjs.org/launch-editor/-/launch-editor-2.14.1.tgz", + "integrity": "sha512-QWBrQsMpH7gPr965dsKD/3cKWiNoTjpATQf++Xq63N6sKRGMwlVXz41O1IZTMfZQgBctD/K5Zt06+/I6pP6+HA==", "dev": true, "requires": { - "picocolors": "^1.0.0", - "shell-quote": "^1.8.1" + "picocolors": "^1.1.1", + "shell-quote": "^1.8.4" } }, "loader-runner": { diff --git a/docs/pyproject.toml b/docs/pyproject.toml index d8fa4f9ec1775..be812d6174f25 100644 --- a/docs/pyproject.toml +++ b/docs/pyproject.toml @@ -5,9 +5,9 @@ requires-python = ">=3.11" dependencies = [ "sphinx>=9,<10", "sphinx-reredirects>=1.1,<2", - "pydata-sphinx-theme>=0.18.0,<1", + "pydata-sphinx-theme>=0.19.0,<1", "myst-parser>=5.1.0,<6", - "maturin>=1.13.3,<2", + "maturin>=1.14.0,<2", "jinja2>=3.1.6,<4", "setuptools>=82.0.1,<83", ] diff --git a/docs/source/user-guide/sql/scalar_functions.md b/docs/source/user-guide/sql/scalar_functions.md index e5cd6f3d99711..83df7b06fd224 100644 --- a/docs/source/user-guide/sql/scalar_functions.md +++ b/docs/source/user-guide/sql/scalar_functions.md @@ -5702,6 +5702,7 @@ union_tag(union_expression) - [arrow_try_cast](#arrow_try_cast) - [arrow_typeof](#arrow_typeof) - [cast_to_type](#cast_to_type) +- [file_row_index](#file_row_index) - [get_field](#get_field) - [try_cast_to_type](#try_cast_to_type) - [version](#version) @@ -5885,6 +5886,27 @@ cast_to_type(expression, reference) +-----+ ``` +### `file_row_index` + +Returns the zero-based row offset within the source file +that produced the current row. + +The value is scoped to one file, so rows from different files in the same scan +can have the same row index. This function is intended to be rewritten at +file-scan time. If the input file is not known (for example, if this function +is evaluated outside a file scan, or was not pushed down into one), direct +evaluation returns an error. + +```sql +file_row_index() +``` + +#### Example + +```sql +SELECT file_row_index() FROM t; +``` + ### `get_field` Returns a field within a map or a struct with the given key. diff --git a/snowflake_flatten_validation.sql b/snowflake_flatten_validation.sql deleted file mode 100644 index cae6f5ea59e77..0000000000000 --- a/snowflake_flatten_validation.sql +++ /dev/null @@ -1,219 +0,0 @@ --- ============================================================================ --- Snowflake LATERAL FLATTEN validation queries --- --- Run this file against a real Snowflake instance to verify that the --- Unparser-generated SQL is syntactically and semantically correct. --- --- Each section shows: --- 1. The DataFusion input (SQL parsed by the planner) --- 2. The Snowflake SQL produced by the Unparser --- --- NOTE: The Unparser emits array literals as [1, 2, 3] (DataFusion syntax). --- Snowflake requires ARRAY_CONSTRUCT(1, 2, 3). The queries below use --- ARRAY_CONSTRUCT so they can run directly on Snowflake. The exact Unparser --- output is shown in the "Unparser output:" comment above each query. --- ============================================================================ - --- ---------------------------------------------------------------------------- --- Setup: create and seed test tables --- ---------------------------------------------------------------------------- - -CREATE OR REPLACE TABLE source ( - items ARRAY -); - -INSERT INTO source SELECT PARSE_JSON('[1, 2, 3]'); -INSERT INTO source SELECT PARSE_JSON('["a", "b"]'); -INSERT INTO source SELECT NULL; - -CREATE OR REPLACE TABLE unnest_table ( - array_col ARRAY -); - -INSERT INTO unnest_table SELECT PARSE_JSON('[10, 20, 30]'); -INSERT INTO unnest_table SELECT PARSE_JSON('[40, 50]'); -INSERT INTO unnest_table SELECT NULL; - -CREATE OR REPLACE TABLE multi_array_table ( - column_a ARRAY, - column_b ARRAY -); - -INSERT INTO multi_array_table SELECT PARSE_JSON('[1, 2, 3]'), PARSE_JSON('["x", "y"]'); -INSERT INTO multi_array_table SELECT PARSE_JSON('[4]'), PARSE_JSON('["z"]'); - --- ============================================================================ --- Roundtrip tests: SQL parsed → plan → Snowflake SQL --- ============================================================================ - --- -------------------------------------------------------------------------- --- Test: snowflake_unnest_to_lateral_flatten_simple --- DataFusion input: SELECT * FROM UNNEST([1,2,3]) --- Unparser output: SELECT "_unnest_1"."VALUE" FROM LATERAL FLATTEN(INPUT => [1, 2, 3]) AS "_unnest_1" --- -------------------------------------------------------------------------- -SELECT "_unnest_1"."VALUE" -FROM LATERAL FLATTEN(INPUT => ARRAY_CONSTRUCT(1, 2, 3)) AS "_unnest_1"; - --- -------------------------------------------------------------------------- --- Test: snowflake_flatten_implicit_from --- DataFusion input: SELECT UNNEST([1,2,3]) --- Unparser output: SELECT "_unnest_1"."VALUE" FROM LATERAL FLATTEN(INPUT => [1, 2, 3]) AS "_unnest_1" --- -------------------------------------------------------------------------- -SELECT "_unnest_1"."VALUE" -FROM LATERAL FLATTEN(INPUT => ARRAY_CONSTRUCT(1, 2, 3)) AS "_unnest_1"; - --- -------------------------------------------------------------------------- --- Test: snowflake_flatten_string_array --- DataFusion input: SELECT * FROM UNNEST(['a','b','c']) --- Unparser output: SELECT "_unnest_1"."VALUE" FROM LATERAL FLATTEN(INPUT => ['a', 'b', 'c']) AS "_unnest_1" --- -------------------------------------------------------------------------- -SELECT "_unnest_1"."VALUE" -FROM LATERAL FLATTEN(INPUT => ARRAY_CONSTRUCT('a', 'b', 'c')) AS "_unnest_1"; - --- -------------------------------------------------------------------------- --- Test: snowflake_flatten_select_unnest_with_alias --- DataFusion input: SELECT UNNEST([1,2,3]) as c1 --- Unparser output: SELECT "_unnest_1"."VALUE" AS "c1" FROM LATERAL FLATTEN(INPUT => [1, 2, 3]) AS "_unnest_1" --- -------------------------------------------------------------------------- -SELECT "_unnest_1"."VALUE" AS "c1" -FROM LATERAL FLATTEN(INPUT => ARRAY_CONSTRUCT(1, 2, 3)) AS "_unnest_1"; - --- -------------------------------------------------------------------------- --- Test: snowflake_flatten_from_unnest_with_table_alias --- DataFusion input: SELECT * FROM UNNEST([1,2,3]) AS t1 (c1) --- Unparser output: SELECT "t1"."VALUE" FROM LATERAL FLATTEN(INPUT => [1, 2, 3]) AS "t1" --- -------------------------------------------------------------------------- -SELECT "t1"."VALUE" -FROM LATERAL FLATTEN(INPUT => ARRAY_CONSTRUCT(1, 2, 3)) AS "t1"; - --- ============================================================================ --- Plan-built tests: LogicalPlan → Snowflake SQL --- These use a table called "source" with an ARRAY column "items". --- ============================================================================ - --- -------------------------------------------------------------------------- --- Test: snowflake_flatten_limit_between_projection_and_unnest --- Plan: Projection → Limit → Unnest → Projection → TableScan --- Unparser output: SELECT "_unnest_1"."VALUE" AS "item" FROM "source" CROSS JOIN LATERAL FLATTEN(INPUT => "source"."items", OUTER => true) AS "_unnest_1" LIMIT 5 --- -------------------------------------------------------------------------- -SELECT "_unnest_1"."VALUE" AS "item" -FROM "source" -CROSS JOIN LATERAL FLATTEN(INPUT => "source"."items", OUTER => true) AS "_unnest_1" -LIMIT 5; - --- -------------------------------------------------------------------------- --- Test: snowflake_flatten_sort_between_projection_and_unnest --- Plan: Projection → Sort → Unnest → Projection → TableScan --- Unparser output: SELECT "_unnest_1"."VALUE" AS "item" FROM "source" CROSS JOIN LATERAL FLATTEN(INPUT => "source"."items", OUTER => true) AS "_unnest_1" ORDER BY "_unnest_1"."VALUE" ASC NULLS FIRST --- -------------------------------------------------------------------------- -SELECT "_unnest_1"."VALUE" AS "item" -FROM "source" -CROSS JOIN LATERAL FLATTEN(INPUT => "source"."items", OUTER => true) AS "_unnest_1" -ORDER BY "_unnest_1"."VALUE" ASC NULLS FIRST; - --- -------------------------------------------------------------------------- --- Test: snowflake_flatten_limit_between_projection_and_unnest_with_subquery_alias --- Plan: Projection → Limit → Unnest → SubqueryAlias → Projection → TableScan --- Unparser output: SELECT "_unnest_1"."VALUE" AS "item" FROM "source" CROSS JOIN LATERAL FLATTEN(INPUT => "source"."items", OUTER => true) AS "_unnest_1" LIMIT 10 --- -------------------------------------------------------------------------- -SELECT "_unnest_1"."VALUE" AS "item" -FROM "source" -CROSS JOIN LATERAL FLATTEN(INPUT => "source"."items", OUTER => true) AS "_unnest_1" -LIMIT 10; - --- -------------------------------------------------------------------------- --- Test: snowflake_flatten_composed_expression_wrapping_unnest --- Plan: Projection(CAST(placeholder AS Int64)) → Unnest → Projection → TableScan --- Unparser output: SELECT CAST("_unnest_1"."VALUE" AS BIGINT) AS "item_id" FROM "source" CROSS JOIN LATERAL FLATTEN(INPUT => "source"."items", OUTER => true) AS "_unnest_1" --- -------------------------------------------------------------------------- -SELECT CAST("_unnest_1"."VALUE" AS BIGINT) AS "item_id" -FROM "source" -CROSS JOIN LATERAL FLATTEN(INPUT => "source"."items", OUTER => true) AS "_unnest_1"; - --- -------------------------------------------------------------------------- --- Test: snowflake_flatten_composed_expression_with_limit --- Plan: Projection(CAST) → Limit → Unnest → Projection → TableScan --- Unparser output: SELECT CAST("_unnest_1"."VALUE" AS BIGINT) AS "item_id" FROM "source" CROSS JOIN LATERAL FLATTEN(INPUT => "source"."items", OUTER => true) AS "_unnest_1" LIMIT 5 --- -------------------------------------------------------------------------- -SELECT CAST("_unnest_1"."VALUE" AS BIGINT) AS "item_id" -FROM "source" -CROSS JOIN LATERAL FLATTEN(INPUT => "source"."items", OUTER => true) AS "_unnest_1" -LIMIT 5; - --- -------------------------------------------------------------------------- --- Test: snowflake_flatten_multi_expression_projection --- Plan: Projection([CAST AS Int64, CAST AS Utf8]) → Unnest → Projection → TableScan --- Unparser output: SELECT CAST("_unnest_1"."VALUE" AS BIGINT) AS "a", CAST("_unnest_1"."VALUE" AS VARCHAR) AS "b" FROM "source" CROSS JOIN LATERAL FLATTEN(INPUT => "source"."items", OUTER => true) AS "_unnest_1" --- -------------------------------------------------------------------------- -SELECT CAST("_unnest_1"."VALUE" AS BIGINT) AS "a", - CAST("_unnest_1"."VALUE" AS VARCHAR) AS "b" -FROM "source" -CROSS JOIN LATERAL FLATTEN(INPUT => "source"."items", OUTER => true) AS "_unnest_1"; - --- -------------------------------------------------------------------------- --- Test: snowflake_flatten_multi_expression_with_limit --- Plan: Projection([CAST, CAST]) → Limit → Unnest → Projection → TableScan --- Unparser output: SELECT CAST("_unnest_1"."VALUE" AS BIGINT) AS "a", CAST("_unnest_1"."VALUE" AS VARCHAR) AS "b" FROM "source" CROSS JOIN LATERAL FLATTEN(INPUT => "source"."items", OUTER => true) AS "_unnest_1" LIMIT 10 --- -------------------------------------------------------------------------- -SELECT CAST("_unnest_1"."VALUE" AS BIGINT) AS "a", - CAST("_unnest_1"."VALUE" AS VARCHAR) AS "b" -FROM "source" -CROSS JOIN LATERAL FLATTEN(INPUT => "source"."items", OUTER => true) AS "_unnest_1" -LIMIT 10; - --- -------------------------------------------------------------------------- --- Test: snowflake_unnest_through_subquery_alias --- Plan: Projection → Unnest → SubqueryAlias → Projection → TableScan --- Unparser output: SELECT "_unnest_1"."VALUE" AS "item" FROM "source" CROSS JOIN LATERAL FLATTEN(INPUT => "source"."items", OUTER => true) AS "_unnest_1" --- -------------------------------------------------------------------------- -SELECT "_unnest_1"."VALUE" AS "item" -FROM "source" -CROSS JOIN LATERAL FLATTEN(INPUT => "source"."items", OUTER => true) AS "_unnest_1"; - --- ============================================================================ --- Roundtrip tests with table columns --- ============================================================================ - --- -------------------------------------------------------------------------- --- Test: snowflake_flatten_unnest_from_subselect --- DataFusion input: SELECT UNNEST(array_col) FROM (SELECT array_col FROM unnest_table WHERE array_col IS NOT NULL LIMIT 3) --- Unparser output: SELECT "_unnest_1"."VALUE" FROM (SELECT "unnest_table"."array_col" FROM "unnest_table" WHERE "unnest_table"."array_col" IS NOT NULL LIMIT 3) CROSS JOIN LATERAL FLATTEN(INPUT => "unnest_table"."array_col") AS "_unnest_1" --- -------------------------------------------------------------------------- -SELECT "_unnest_1"."VALUE" -FROM ( - SELECT "unnest_table"."array_col" - FROM "unnest_table" - WHERE "unnest_table"."array_col" IS NOT NULL - LIMIT 3 -) CROSS JOIN LATERAL FLATTEN(INPUT => "unnest_table"."array_col") AS "_unnest_1"; - --- -------------------------------------------------------------------------- --- Test: snowflake_flatten_cross_join_unnest_table_column --- DataFusion input: SELECT * FROM multi_array_table CROSS JOIN UNNEST(column_a) AS a (a) --- Unparser output: SELECT "multi_array_table"."column_a", "multi_array_table"."column_b", "a"."VALUE" FROM "multi_array_table" CROSS JOIN LATERAL FLATTEN(INPUT => "multi_array_table"."column_a") AS "a" --- -------------------------------------------------------------------------- -SELECT "multi_array_table"."column_a", - "multi_array_table"."column_b", - "a"."VALUE" -FROM "multi_array_table" -CROSS JOIN LATERAL FLATTEN(INPUT => "multi_array_table"."column_a") AS "a"; - --- -------------------------------------------------------------------------- --- Test: snowflake_flatten_multiple_unnest_cross_join --- DataFusion input: SELECT a.a, b.b FROM multi_array_table --- CROSS JOIN UNNEST(column_a) AS a (a) --- CROSS JOIN UNNEST(column_b) AS b (b) --- Unparser output: SELECT "a"."VALUE", "b"."VALUE" FROM "multi_array_table" CROSS JOIN LATERAL FLATTEN(INPUT => "multi_array_table"."column_a") AS "a" CROSS JOIN LATERAL FLATTEN(INPUT => "multi_array_table"."column_b") AS "b" --- -------------------------------------------------------------------------- -SELECT "a"."VALUE", - "b"."VALUE" -FROM "multi_array_table" -CROSS JOIN LATERAL FLATTEN(INPUT => "multi_array_table"."column_a") AS "a" -CROSS JOIN LATERAL FLATTEN(INPUT => "multi_array_table"."column_b") AS "b"; - --- ============================================================================ --- Cleanup --- ============================================================================ --- DROP TABLE IF EXISTS source; --- DROP TABLE IF EXISTS unnest_table; --- DROP TABLE IF EXISTS multi_array_table; diff --git a/uv.lock b/uv.lock index 70c7cc04b3c72..bdda81c5f9777 100644 --- a/uv.lock +++ b/uv.lock @@ -240,61 +240,61 @@ wheels = [ [[package]] name = "cryptography" -version = "46.0.7" +version = "48.0.1" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "cffi", marker = "platform_python_implementation != 'PyPy'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/47/93/ac8f3d5ff04d54bc814e961a43ae5b0b146154c89c61b47bb07557679b18/cryptography-46.0.7.tar.gz", hash = "sha256:e4cfd68c5f3e0bfdad0d38e023239b96a2fe84146481852dffbcca442c245aa5", size = 750652, upload-time = "2026-04-08T01:57:54.692Z" } +sdist = { url = "https://files.pythonhosted.org/packages/12/45/870e7f4bef50e5f53b9f51d4428aee5290eedf58ba443f16b1ebb7ab8e66/cryptography-48.0.1.tar.gz", hash = "sha256:266f4ee051abb2f725b74ef8072b521ce1feacf685a3364fa6a6b45548db791a", size = 832989, upload-time = "2026-06-09T22:32:31.8Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/0b/5d/4a8f770695d73be252331e60e526291e3df0c9b27556a90a6b47bccca4c2/cryptography-46.0.7-cp311-abi3-macosx_10_9_universal2.whl", hash = "sha256:ea42cbe97209df307fdc3b155f1b6fa2577c0defa8f1f7d3be7d31d189108ad4", size = 7179869, upload-time = "2026-04-08T01:56:17.157Z" }, - { url = "https://files.pythonhosted.org/packages/5f/45/6d80dc379b0bbc1f9d1e429f42e4cb9e1d319c7a8201beffd967c516ea01/cryptography-46.0.7-cp311-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:b36a4695e29fe69215d75960b22577197aca3f7a25b9cf9d165dcfe9d80bc325", size = 4275492, upload-time = "2026-04-08T01:56:19.36Z" }, - { url = "https://files.pythonhosted.org/packages/4a/9a/1765afe9f572e239c3469f2cb429f3ba7b31878c893b246b4b2994ffe2fe/cryptography-46.0.7-cp311-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:5ad9ef796328c5e3c4ceed237a183f5d41d21150f972455a9d926593a1dcb308", size = 4426670, upload-time = "2026-04-08T01:56:21.415Z" }, - { url = "https://files.pythonhosted.org/packages/8f/3e/af9246aaf23cd4ee060699adab1e47ced3f5f7e7a8ffdd339f817b446462/cryptography-46.0.7-cp311-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:73510b83623e080a2c35c62c15298096e2a5dc8d51c3b4e1740211839d0dea77", size = 4280275, upload-time = "2026-04-08T01:56:23.539Z" }, - { url = "https://files.pythonhosted.org/packages/0f/54/6bbbfc5efe86f9d71041827b793c24811a017c6ac0fd12883e4caa86b8ed/cryptography-46.0.7-cp311-abi3-manylinux_2_28_ppc64le.whl", hash = "sha256:cbd5fb06b62bd0721e1170273d3f4d5a277044c47ca27ee257025146c34cbdd1", size = 4928402, upload-time = "2026-04-08T01:56:25.624Z" }, - { url = "https://files.pythonhosted.org/packages/2d/cf/054b9d8220f81509939599c8bdbc0c408dbd2bdd41688616a20731371fe0/cryptography-46.0.7-cp311-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:420b1e4109cc95f0e5700eed79908cef9268265c773d3a66f7af1eef53d409ef", size = 4459985, upload-time = "2026-04-08T01:56:27.309Z" }, - { url = "https://files.pythonhosted.org/packages/f9/46/4e4e9c6040fb01c7467d47217d2f882daddeb8828f7df800cb806d8a2288/cryptography-46.0.7-cp311-abi3-manylinux_2_31_armv7l.whl", hash = "sha256:24402210aa54baae71d99441d15bb5a1919c195398a87b563df84468160a65de", size = 3990652, upload-time = "2026-04-08T01:56:29.095Z" }, - { url = "https://files.pythonhosted.org/packages/36/5f/313586c3be5a2fbe87e4c9a254207b860155a8e1f3cca99f9910008e7d08/cryptography-46.0.7-cp311-abi3-manylinux_2_34_aarch64.whl", hash = "sha256:8a469028a86f12eb7d2fe97162d0634026d92a21f3ae0ac87ed1c4a447886c83", size = 4279805, upload-time = "2026-04-08T01:56:30.928Z" }, - { url = "https://files.pythonhosted.org/packages/69/33/60dfc4595f334a2082749673386a4d05e4f0cf4df8248e63b2c3437585f2/cryptography-46.0.7-cp311-abi3-manylinux_2_34_ppc64le.whl", hash = "sha256:9694078c5d44c157ef3162e3bf3946510b857df5a3955458381d1c7cfc143ddb", size = 4892883, upload-time = "2026-04-08T01:56:32.614Z" }, - { url = "https://files.pythonhosted.org/packages/c7/0b/333ddab4270c4f5b972f980adef4faa66951a4aaf646ca067af597f15563/cryptography-46.0.7-cp311-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:42a1e5f98abb6391717978baf9f90dc28a743b7d9be7f0751a6f56a75d14065b", size = 4459756, upload-time = "2026-04-08T01:56:34.306Z" }, - { url = "https://files.pythonhosted.org/packages/d2/14/633913398b43b75f1234834170947957c6b623d1701ffc7a9600da907e89/cryptography-46.0.7-cp311-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:91bbcb08347344f810cbe49065914fe048949648f6bd5c2519f34619142bbe85", size = 4410244, upload-time = "2026-04-08T01:56:35.977Z" }, - { url = "https://files.pythonhosted.org/packages/10/f2/19ceb3b3dc14009373432af0c13f46aa08e3ce334ec6eff13492e1812ccd/cryptography-46.0.7-cp311-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:5d1c02a14ceb9148cc7816249f64f623fbfee39e8c03b3650d842ad3f34d637e", size = 4674868, upload-time = "2026-04-08T01:56:38.034Z" }, - { url = "https://files.pythonhosted.org/packages/1a/bb/a5c213c19ee94b15dfccc48f363738633a493812687f5567addbcbba9f6f/cryptography-46.0.7-cp311-abi3-win32.whl", hash = "sha256:d23c8ca48e44ee015cd0a54aeccdf9f09004eba9fc96f38c911011d9ff1bd457", size = 3026504, upload-time = "2026-04-08T01:56:39.666Z" }, - { url = "https://files.pythonhosted.org/packages/2b/02/7788f9fefa1d060ca68717c3901ae7fffa21ee087a90b7f23c7a603c32ae/cryptography-46.0.7-cp311-abi3-win_amd64.whl", hash = "sha256:397655da831414d165029da9bc483bed2fe0e75dde6a1523ec2fe63f3c46046b", size = 3488363, upload-time = "2026-04-08T01:56:41.893Z" }, - { url = "https://files.pythonhosted.org/packages/7b/56/15619b210e689c5403bb0540e4cb7dbf11a6bf42e483b7644e471a2812b3/cryptography-46.0.7-cp314-cp314t-macosx_10_9_universal2.whl", hash = "sha256:d151173275e1728cf7839aaa80c34fe550c04ddb27b34f48c232193df8db5842", size = 7119671, upload-time = "2026-04-08T01:56:44Z" }, - { url = "https://files.pythonhosted.org/packages/74/66/e3ce040721b0b5599e175ba91ab08884c75928fbeb74597dd10ef13505d2/cryptography-46.0.7-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:db0f493b9181c7820c8134437eb8b0b4792085d37dbb24da050476ccb664e59c", size = 4268551, upload-time = "2026-04-08T01:56:46.071Z" }, - { url = "https://files.pythonhosted.org/packages/03/11/5e395f961d6868269835dee1bafec6a1ac176505a167f68b7d8818431068/cryptography-46.0.7-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:ebd6daf519b9f189f85c479427bbd6e9c9037862cf8fe89ee35503bd209ed902", size = 4408887, upload-time = "2026-04-08T01:56:47.718Z" }, - { url = "https://files.pythonhosted.org/packages/40/53/8ed1cf4c3b9c8e611e7122fb56f1c32d09e1fff0f1d77e78d9ff7c82653e/cryptography-46.0.7-cp314-cp314t-manylinux_2_28_aarch64.whl", hash = "sha256:b7b412817be92117ec5ed95f880defe9cf18a832e8cafacf0a22337dc1981b4d", size = 4271354, upload-time = "2026-04-08T01:56:49.312Z" }, - { url = "https://files.pythonhosted.org/packages/50/46/cf71e26025c2e767c5609162c866a78e8a2915bbcfa408b7ca495c6140c4/cryptography-46.0.7-cp314-cp314t-manylinux_2_28_ppc64le.whl", hash = "sha256:fbfd0e5f273877695cb93baf14b185f4878128b250cc9f8e617ea0c025dfb022", size = 4905845, upload-time = "2026-04-08T01:56:50.916Z" }, - { url = "https://files.pythonhosted.org/packages/c0/ea/01276740375bac6249d0a971ebdf6b4dc9ead0ee0a34ef3b5a88c1a9b0d4/cryptography-46.0.7-cp314-cp314t-manylinux_2_28_x86_64.whl", hash = "sha256:ffca7aa1d00cf7d6469b988c581598f2259e46215e0140af408966a24cf086ce", size = 4444641, upload-time = "2026-04-08T01:56:52.882Z" }, - { url = "https://files.pythonhosted.org/packages/3d/4c/7d258f169ae71230f25d9f3d06caabcff8c3baf0978e2b7d65e0acac3827/cryptography-46.0.7-cp314-cp314t-manylinux_2_31_armv7l.whl", hash = "sha256:60627cf07e0d9274338521205899337c5d18249db56865f943cbe753aa96f40f", size = 3967749, upload-time = "2026-04-08T01:56:54.597Z" }, - { url = "https://files.pythonhosted.org/packages/b5/2a/2ea0767cad19e71b3530e4cad9605d0b5e338b6a1e72c37c9c1ceb86c333/cryptography-46.0.7-cp314-cp314t-manylinux_2_34_aarch64.whl", hash = "sha256:80406c3065e2c55d7f49a9550fe0c49b3f12e5bfff5dedb727e319e1afb9bf99", size = 4270942, upload-time = "2026-04-08T01:56:56.416Z" }, - { url = "https://files.pythonhosted.org/packages/41/3d/fe14df95a83319af25717677e956567a105bb6ab25641acaa093db79975d/cryptography-46.0.7-cp314-cp314t-manylinux_2_34_ppc64le.whl", hash = "sha256:c5b1ccd1239f48b7151a65bc6dd54bcfcc15e028c8ac126d3fada09db0e07ef1", size = 4871079, upload-time = "2026-04-08T01:56:58.31Z" }, - { url = "https://files.pythonhosted.org/packages/9c/59/4a479e0f36f8f378d397f4eab4c850b4ffb79a2f0d58704b8fa0703ddc11/cryptography-46.0.7-cp314-cp314t-manylinux_2_34_x86_64.whl", hash = "sha256:d5f7520159cd9c2154eb61eb67548ca05c5774d39e9c2c4339fd793fe7d097b2", size = 4443999, upload-time = "2026-04-08T01:57:00.508Z" }, - { url = "https://files.pythonhosted.org/packages/28/17/b59a741645822ec6d04732b43c5d35e4ef58be7bfa84a81e5ae6f05a1d33/cryptography-46.0.7-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:fcd8eac50d9138c1d7fc53a653ba60a2bee81a505f9f8850b6b2888555a45d0e", size = 4399191, upload-time = "2026-04-08T01:57:02.654Z" }, - { url = "https://files.pythonhosted.org/packages/59/6a/bb2e166d6d0e0955f1e9ff70f10ec4b2824c9cfcdb4da772c7dd69cc7d80/cryptography-46.0.7-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:65814c60f8cc400c63131584e3e1fad01235edba2614b61fbfbfa954082db0ee", size = 4655782, upload-time = "2026-04-08T01:57:04.592Z" }, - { url = "https://files.pythonhosted.org/packages/95/b6/3da51d48415bcb63b00dc17c2eff3a651b7c4fed484308d0f19b30e8cb2c/cryptography-46.0.7-cp314-cp314t-win32.whl", hash = "sha256:fdd1736fed309b4300346f88f74cd120c27c56852c3838cab416e7a166f67298", size = 3002227, upload-time = "2026-04-08T01:57:06.91Z" }, - { url = "https://files.pythonhosted.org/packages/32/a8/9f0e4ed57ec9cebe506e58db11ae472972ecb0c659e4d52bbaee80ca340a/cryptography-46.0.7-cp314-cp314t-win_amd64.whl", hash = "sha256:e06acf3c99be55aa3b516397fe42f5855597f430add9c17fa46bf2e0fb34c9bb", size = 3475332, upload-time = "2026-04-08T01:57:08.807Z" }, - { url = "https://files.pythonhosted.org/packages/a7/7f/cd42fc3614386bc0c12f0cb3c4ae1fc2bbca5c9662dfed031514911d513d/cryptography-46.0.7-cp38-abi3-macosx_10_9_universal2.whl", hash = "sha256:462ad5cb1c148a22b2e3bcc5ad52504dff325d17daf5df8d88c17dda1f75f2a4", size = 7165618, upload-time = "2026-04-08T01:57:10.645Z" }, - { url = "https://files.pythonhosted.org/packages/a5/d0/36a49f0262d2319139d2829f773f1b97ef8aef7f97e6e5bd21455e5a8fb5/cryptography-46.0.7-cp38-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:84d4cced91f0f159a7ddacad249cc077e63195c36aac40b4150e7a57e84fffe7", size = 4270628, upload-time = "2026-04-08T01:57:12.885Z" }, - { url = "https://files.pythonhosted.org/packages/8a/6c/1a42450f464dda6ffbe578a911f773e54dd48c10f9895a23a7e88b3e7db5/cryptography-46.0.7-cp38-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:128c5edfe5e5938b86b03941e94fac9ee793a94452ad1365c9fc3f4f62216832", size = 4415405, upload-time = "2026-04-08T01:57:14.923Z" }, - { url = "https://files.pythonhosted.org/packages/9a/92/4ed714dbe93a066dc1f4b4581a464d2d7dbec9046f7c8b7016f5286329e2/cryptography-46.0.7-cp38-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:5e51be372b26ef4ba3de3c167cd3d1022934bc838ae9eaad7e644986d2a3d163", size = 4272715, upload-time = "2026-04-08T01:57:16.638Z" }, - { url = "https://files.pythonhosted.org/packages/b7/e6/a26b84096eddd51494bba19111f8fffe976f6a09f132706f8f1bf03f51f7/cryptography-46.0.7-cp38-abi3-manylinux_2_28_ppc64le.whl", hash = "sha256:cdf1a610ef82abb396451862739e3fc93b071c844399e15b90726ef7470eeaf2", size = 4918400, upload-time = "2026-04-08T01:57:19.021Z" }, - { url = "https://files.pythonhosted.org/packages/c7/08/ffd537b605568a148543ac3c2b239708ae0bd635064bab41359252ef88ed/cryptography-46.0.7-cp38-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:1d25aee46d0c6f1a501adcddb2d2fee4b979381346a78558ed13e50aa8a59067", size = 4450634, upload-time = "2026-04-08T01:57:21.185Z" }, - { url = "https://files.pythonhosted.org/packages/16/01/0cd51dd86ab5b9befe0d031e276510491976c3a80e9f6e31810cce46c4ad/cryptography-46.0.7-cp38-abi3-manylinux_2_31_armv7l.whl", hash = "sha256:cdfbe22376065ffcf8be74dc9a909f032df19bc58a699456a21712d6e5eabfd0", size = 3985233, upload-time = "2026-04-08T01:57:22.862Z" }, - { url = "https://files.pythonhosted.org/packages/92/49/819d6ed3a7d9349c2939f81b500a738cb733ab62fbecdbc1e38e83d45e12/cryptography-46.0.7-cp38-abi3-manylinux_2_34_aarch64.whl", hash = "sha256:abad9dac36cbf55de6eb49badd4016806b3165d396f64925bf2999bcb67837ba", size = 4271955, upload-time = "2026-04-08T01:57:24.814Z" }, - { url = "https://files.pythonhosted.org/packages/80/07/ad9b3c56ebb95ed2473d46df0847357e01583f4c52a85754d1a55e29e4d0/cryptography-46.0.7-cp38-abi3-manylinux_2_34_ppc64le.whl", hash = "sha256:935ce7e3cfdb53e3536119a542b839bb94ec1ad081013e9ab9b7cfd478b05006", size = 4879888, upload-time = "2026-04-08T01:57:26.88Z" }, - { url = "https://files.pythonhosted.org/packages/b8/c7/201d3d58f30c4c2bdbe9b03844c291feb77c20511cc3586daf7edc12a47b/cryptography-46.0.7-cp38-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:35719dc79d4730d30f1c2b6474bd6acda36ae2dfae1e3c16f2051f215df33ce0", size = 4449961, upload-time = "2026-04-08T01:57:29.068Z" }, - { url = "https://files.pythonhosted.org/packages/a5/ef/649750cbf96f3033c3c976e112265c33906f8e462291a33d77f90356548c/cryptography-46.0.7-cp38-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:7bbc6ccf49d05ac8f7d7b5e2e2c33830d4fe2061def88210a126d130d7f71a85", size = 4401696, upload-time = "2026-04-08T01:57:31.029Z" }, - { url = "https://files.pythonhosted.org/packages/41/52/a8908dcb1a389a459a29008c29966c1d552588d4ae6d43f3a1a4512e0ebe/cryptography-46.0.7-cp38-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:a1529d614f44b863a7b480c6d000fe93b59acee9c82ffa027cfadc77521a9f5e", size = 4664256, upload-time = "2026-04-08T01:57:33.144Z" }, - { url = "https://files.pythonhosted.org/packages/4b/fa/f0ab06238e899cc3fb332623f337a7364f36f4bb3f2534c2bb95a35b132c/cryptography-46.0.7-cp38-abi3-win32.whl", hash = "sha256:f247c8c1a1fb45e12586afbb436ef21ff1e80670b2861a90353d9b025583d246", size = 3013001, upload-time = "2026-04-08T01:57:34.933Z" }, - { url = "https://files.pythonhosted.org/packages/d2/f1/00ce3bde3ca542d1acd8f8cfa38e446840945aa6363f9b74746394b14127/cryptography-46.0.7-cp38-abi3-win_amd64.whl", hash = "sha256:506c4ff91eff4f82bdac7633318a526b1d1309fc07ca76a3ad182cb5b686d6d3", size = 3472985, upload-time = "2026-04-08T01:57:36.714Z" }, - { url = "https://files.pythonhosted.org/packages/63/0c/dca8abb64e7ca4f6b2978769f6fea5ad06686a190cec381f0a796fdcaaba/cryptography-46.0.7-pp311-pypy311_pp73-macosx_11_0_arm64.whl", hash = "sha256:fc9ab8856ae6cf7c9358430e49b368f3108f050031442eaeb6b9d87e4dcf4e4f", size = 3476879, upload-time = "2026-04-08T01:57:38.664Z" }, - { url = "https://files.pythonhosted.org/packages/3a/ea/075aac6a84b7c271578d81a2f9968acb6e273002408729f2ddff517fed4a/cryptography-46.0.7-pp311-pypy311_pp73-manylinux_2_28_aarch64.whl", hash = "sha256:d3b99c535a9de0adced13d159c5a9cf65c325601aa30f4be08afd680643e9c15", size = 4219700, upload-time = "2026-04-08T01:57:40.625Z" }, - { url = "https://files.pythonhosted.org/packages/6c/7b/1c55db7242b5e5612b29fc7a630e91ee7a6e3c8e7bf5406d22e206875fbd/cryptography-46.0.7-pp311-pypy311_pp73-manylinux_2_28_x86_64.whl", hash = "sha256:d02c738dacda7dc2a74d1b2b3177042009d5cab7c7079db74afc19e56ca1b455", size = 4385982, upload-time = "2026-04-08T01:57:42.725Z" }, - { url = "https://files.pythonhosted.org/packages/cb/da/9870eec4b69c63ef5925bf7d8342b7e13bc2ee3d47791461c4e49ca212f4/cryptography-46.0.7-pp311-pypy311_pp73-manylinux_2_34_aarch64.whl", hash = "sha256:04959522f938493042d595a736e7dbdff6eb6cc2339c11465b3ff89343b65f65", size = 4219115, upload-time = "2026-04-08T01:57:44.939Z" }, - { url = "https://files.pythonhosted.org/packages/f4/72/05aa5832b82dd341969e9a734d1812a6aadb088d9eb6f0430fc337cc5a8f/cryptography-46.0.7-pp311-pypy311_pp73-manylinux_2_34_x86_64.whl", hash = "sha256:3986ac1dee6def53797289999eabe84798ad7817f3e97779b5061a95b0ee4968", size = 4385479, upload-time = "2026-04-08T01:57:46.86Z" }, - { url = "https://files.pythonhosted.org/packages/20/2a/1b016902351a523aa2bd446b50a5bc1175d7a7d1cf90fe2ef904f9b84ebc/cryptography-46.0.7-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:258514877e15963bd43b558917bc9f54cf7cf866c38aa576ebf47a77ddbc43a4", size = 3412829, upload-time = "2026-04-08T01:57:48.874Z" }, + { url = "https://files.pythonhosted.org/packages/1b/bc/ee4137cbbe105652c0ee4252792b78fc8e7afa4b8e61d9d5dc05a7f45731/cryptography-48.0.1-cp311-abi3-macosx_10_9_universal2.whl", hash = "sha256:3e4a1a3232eef2e6c732827d5722db29a0cc8b27af2a4d865b094cf954be9ca1", size = 8008324, upload-time = "2026-06-09T22:31:00.702Z" }, + { url = "https://files.pythonhosted.org/packages/d5/85/6379d42181bfc713094f081360fc5784d6c816b599d45e7f082502d173ce/cryptography-48.0.1-cp311-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:32143b24adb918f078134e1e230f1eb8cc04886b92c28b5f0041aaf3e5699225", size = 4696243, upload-time = "2026-06-09T22:32:33.446Z" }, + { url = "https://files.pythonhosted.org/packages/9c/87/c85d147b53323c7eb4d850920c8901377323c2a0ff8d79c262d4fee89aa2/cryptography-48.0.1-cp311-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:f0d27a5696721ef7a672b8c810f6aded391058e0b9486e63e6d93baf765da691", size = 4713235, upload-time = "2026-06-09T22:31:40.141Z" }, + { url = "https://files.pythonhosted.org/packages/79/58/67cbf8cf1ee7c54b439ca07bbecf8362c07afc11a3724fea70f745784add/cryptography-48.0.1-cp311-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:eb86ce1af36fe65041b6db9a8bb064ee621a7e5fded0f80d475ec243477cd242", size = 4702323, upload-time = "2026-06-09T22:31:42.191Z" }, + { url = "https://files.pythonhosted.org/packages/89/c6/24266ac10c47f6cd2a865f4446062b466da1d1f10b27189eac00e61bf0c9/cryptography-48.0.1-cp311-abi3-manylinux_2_28_ppc64le.whl", hash = "sha256:b024e784ad6c077ee0147b35ea9cbfc1e34e1fd4c1dcca214c2794d73a12df08", size = 5300085, upload-time = "2026-06-09T22:31:58.703Z" }, + { url = "https://files.pythonhosted.org/packages/d2/bb/cc4b78784f97efc8c5874c2a9743708d172be6663024b34a0467885ae0c8/cryptography-48.0.1-cp311-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:3752f2dbc8f07a30aad2932c986cea495b03bb554887828225da104f732852b6", size = 4746137, upload-time = "2026-06-09T22:31:31.01Z" }, + { url = "https://files.pythonhosted.org/packages/1f/52/0c44de3f5267f8fbe8e835138017522a333436166e406f0db9b9e6e3033f/cryptography-48.0.1-cp311-abi3-manylinux_2_31_armv7l.whl", hash = "sha256:bd81490cd5801d755cf97bb68ac191f14b708470b1c7cf4580f669b9c9264cd8", size = 4333867, upload-time = "2026-06-09T22:32:28.096Z" }, + { url = "https://files.pythonhosted.org/packages/9a/2e/772d7adbfa931537bc401640b7cac9976bff689bda187833e5d63b428e49/cryptography-48.0.1-cp311-abi3-manylinux_2_34_aarch64.whl", hash = "sha256:66fd0771e7b9c6dcd44cf1120690d2338d16d72795cf40cae2786a39eba65429", size = 4701805, upload-time = "2026-06-09T22:31:38.284Z" }, + { url = "https://files.pythonhosted.org/packages/f8/a3/b06844f303873493c963caf581c04df31c7035e0c1b0f02c4814d319ec80/cryptography-48.0.1-cp311-abi3-manylinux_2_34_ppc64le.whl", hash = "sha256:3fd2ca57062b241c856670b073487d2e86c4637937ca5601e48f97bf8e11fc8f", size = 5258461, upload-time = "2026-06-09T22:31:04.187Z" }, + { url = "https://files.pythonhosted.org/packages/9f/13/8b765e2e12b07c74941caadb9d1c8fdc006c4dfbf2b8f2d610519758954d/cryptography-48.0.1-cp311-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:0ee6ea481db1ab889cba043ec1eda17bb9c1ea79db6722f779c3667f9f70322f", size = 4745488, upload-time = "2026-06-09T22:32:30.07Z" }, + { url = "https://files.pythonhosted.org/packages/2e/aa/48972bce55049b32a94f4907eda4d75fa385aad8a39506cc2fc72196ecf0/cryptography-48.0.1-cp311-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:f2ceef93cb096aa3c4cc4b5c94ca6131f9196d28c64d6111533402a9b2054d41", size = 4830256, upload-time = "2026-06-09T22:31:43.868Z" }, + { url = "https://files.pythonhosted.org/packages/47/a2/e5079a032fb85cf6005046ca92bbd78b0c82dad2b5751ab8c311659da06f/cryptography-48.0.1-cp311-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:9bd3f92d76217892b15df84ca256c2c113d386fdda7a7d8691aeeced976507c6", size = 4979117, upload-time = "2026-06-09T22:31:05.845Z" }, + { url = "https://files.pythonhosted.org/packages/b7/a0/8f50cae9c74e718ed769d63ed5c74bd0ea830c9550a74629cebd1b9c7bc7/cryptography-48.0.1-cp311-abi3-win32.whl", hash = "sha256:b9a32b876490d66c8bcc9963ef220199569748434ab01a9d6aaeabf88e7f5158", size = 3304154, upload-time = "2026-06-09T22:32:16.845Z" }, + { url = "https://files.pythonhosted.org/packages/c5/69/0572c77dbace6fef72f33755bd52ea399c71367250d366237f8691826b9e/cryptography-48.0.1-cp311-abi3-win_amd64.whl", hash = "sha256:39489bfca54c7a1f6b297efcd8bc608ab92d16c4ca631b0cad4da46724588b24", size = 3817138, upload-time = "2026-06-09T22:32:00.388Z" }, + { url = "https://files.pythonhosted.org/packages/42/06/3e768b4c3bc78201583fa35a0e18f640dd782ff41afba88f8545481a8874/cryptography-48.0.1-cp314-cp314t-macosx_10_9_universal2.whl", hash = "sha256:f817adc181390bd54f2f700107a7419040fb7c1bdf2fc26f36551a06a68c3345", size = 7989830, upload-time = "2026-06-09T22:31:07.8Z" }, + { url = "https://files.pythonhosted.org/packages/8a/13/6476736484b94041110c8340a3eb63962fea4975baea8cb4a512adb44d4d/cryptography-48.0.1-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:d5d30989c6917b478b5817902e85fddaea2261efa8648383d965381ccb9e1ac4", size = 4689201, upload-time = "2026-06-09T22:31:09.745Z" }, + { url = "https://files.pythonhosted.org/packages/79/62/65a87f34d2a431546e2509b85d55e8c90df86d668f6731da64d538512ac2/cryptography-48.0.1-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:df637c05205ea7c1d7fbcbe54bbfea648a52951155f997af13d895d0ecc96991", size = 4702822, upload-time = "2026-06-09T22:32:24.409Z" }, + { url = "https://files.pythonhosted.org/packages/7f/59/810b5204b0a9b10f4b6bc06bd551a8b609803cd931806bc3b71884b225e5/cryptography-48.0.1-cp314-cp314t-manylinux_2_28_aarch64.whl", hash = "sha256:869c3b8a53bfe27147832df48b32adadf558249d50e76cb3769d40e986b13265", size = 4694875, upload-time = "2026-06-09T22:32:08.737Z" }, + { url = "https://files.pythonhosted.org/packages/24/dc/d8ca05ffea724eec6d232ea6f18e74c269eb6bdfdcc9bfba689790d1325f/cryptography-48.0.1-cp314-cp314t-manylinux_2_28_ppc64le.whl", hash = "sha256:e361afba8918070d376df76f408a4f67fec0ee9cff81a99e48fe9a233ef59e17", size = 5290385, upload-time = "2026-06-09T22:31:15.212Z" }, + { url = "https://files.pythonhosted.org/packages/03/8c/3be6cb4da181f5bb6c19cf560c2359d60644a6b5fc5b57854e528f47b296/cryptography-48.0.1-cp314-cp314t-manylinux_2_28_x86_64.whl", hash = "sha256:d069066deead00ac7f090be101be875a06855908f7ec004c27b8fefb4acfb411", size = 4737082, upload-time = "2026-06-09T22:32:22.66Z" }, + { url = "https://files.pythonhosted.org/packages/aa/f6/d5f60a5a1434dbfd949e227fd0065d194c7e6b6ac526b17f5c06152b8231/cryptography-48.0.1-cp314-cp314t-manylinux_2_31_armv7l.whl", hash = "sha256:09f73a725d582cef64b91281a322cd798d14a33b2b6f2b7ad9531dc336d84c02", size = 4325328, upload-time = "2026-06-09T22:32:10.777Z" }, + { url = "https://files.pythonhosted.org/packages/17/b7/ba75dd947a14b6ad907b01ae8f6b5b348cdd1b48142f0063dee9e20c1d9d/cryptography-48.0.1-cp314-cp314t-manylinux_2_34_aarch64.whl", hash = "sha256:15254441469dd6bf027039453288e2072124f8b6603563f5d759e1c9b69273fa", size = 4694530, upload-time = "2026-06-09T22:31:53.105Z" }, + { url = "https://files.pythonhosted.org/packages/62/29/50d6b9e8aff12d8b67afaeb3569335e32dc83a5723e3bbded24fdac9f809/cryptography-48.0.1-cp314-cp314t-manylinux_2_34_ppc64le.whl", hash = "sha256:8ace4507d1e6533c125f4fac754f8bb8b6a74c08e92179dabd7e16571a3efbf3", size = 5245046, upload-time = "2026-06-09T22:31:25.774Z" }, + { url = "https://files.pythonhosted.org/packages/9f/04/618f4115cfc0add0838c82507aa18a346089428da8653ad38b3ff36f5cb3/cryptography-48.0.1-cp314-cp314t-manylinux_2_34_x86_64.whl", hash = "sha256:b4e391975f038e66432328639620a4aff2d307513b004f1ca06d6225bced815c", size = 4736660, upload-time = "2026-06-09T22:32:12.676Z" }, + { url = "https://files.pythonhosted.org/packages/24/9c/06e062462a0de28a3b3911322eded4c16deb9f441b1b7575d3dc59488ab5/cryptography-48.0.1-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:42fcd8e26fe555d9b3577a135f5091fefa0aa4e99129c23fb56787a1bd4ada72", size = 4822229, upload-time = "2026-06-09T22:31:17.062Z" }, + { url = "https://files.pythonhosted.org/packages/f4/be/0561971eaaee4b8a0e7d5113c536921063ab91aaf23278ac374eaf881e11/cryptography-48.0.1-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:c1400da5e32a43253392277eac7490a60e497d810a63dd5608d71bbd7af507c9", size = 4966364, upload-time = "2026-06-09T22:31:32.842Z" }, + { url = "https://files.pythonhosted.org/packages/a4/27/728c77876f12b000820b69ae490f3c4083775e79e07827e9e60be07ad209/cryptography-48.0.1-cp314-cp314t-win32.whl", hash = "sha256:0df56b056bc17c1b7d6821dfa65216e62bd232d8ab05eb3db44e71d235651471", size = 3278498, upload-time = "2026-06-09T22:31:29.154Z" }, + { url = "https://files.pythonhosted.org/packages/06/e3/79a612c6d7b1e6ee0edd43633d53035bec2cfb78c82b76f7864f39e36f34/cryptography-48.0.1-cp314-cp314t-win_amd64.whl", hash = "sha256:9de21387aa95e2a895823d0745b430bed4f33503ba9ab5e0b5311f33e37d66d2", size = 3798790, upload-time = "2026-06-09T22:31:56.697Z" }, + { url = "https://files.pythonhosted.org/packages/ca/6c/00fa2a95997164c8b2072ce327c23d4ab20809ccc323ea5fab91e53a4bba/cryptography-48.0.1-cp39-abi3-macosx_10_9_universal2.whl", hash = "sha256:4fdc69f8e4316bcf0c8c8ec1f26f285d12e8142d88d96c876a59a03be3f6ae67", size = 7987408, upload-time = "2026-06-09T22:32:20.777Z" }, + { url = "https://files.pythonhosted.org/packages/b0/d9/45f309a7e4e5f3f8f121d6d3be9e94024a7726ec598d6e08ae04edb2f04d/cryptography-48.0.1-cp39-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:48fe40804d4caa2288f24e70ca8c64c42dd826da0ad7e4f1b41b2128d679e6c8", size = 4690196, upload-time = "2026-06-09T22:31:54.74Z" }, + { url = "https://files.pythonhosted.org/packages/5f/9f/a1bc8bcc798811b8527eb374bbccf30a3f3e806829d967118222bf1125eb/cryptography-48.0.1-cp39-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:86be3b1b0b6bf09482fb50a979c508d2950ed95f5621ec77f4e385962006b83a", size = 4696782, upload-time = "2026-06-09T22:31:45.615Z" }, + { url = "https://files.pythonhosted.org/packages/66/c2/81a4fb4e4373c500bb526bc337ac5719dd31dd15b970b84a238168c6aa08/cryptography-48.0.1-cp39-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:4ab0a343c807bbcd90c971cd1ecf072937cd01847a9e002bef88fb47ac6be577", size = 4696618, upload-time = "2026-06-09T22:31:11.564Z" }, + { url = "https://files.pythonhosted.org/packages/e5/0b/aa68b221dde92d09cb29a024ede17550ee21e77a404e59fc093c82bb51e1/cryptography-48.0.1-cp39-abi3-manylinux_2_28_ppc64le.whl", hash = "sha256:9621de99d2da096006b629979efd8ae7eb2d8b822488d0c89ee4000c306c59b1", size = 5289970, upload-time = "2026-06-09T22:31:20.368Z" }, + { url = "https://files.pythonhosted.org/packages/78/13/fba657f958d2af66ea959a4ba01212632089249d34af1ae48054136344d7/cryptography-48.0.1-cp39-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:88c852a0ae366e262e5a1744b685e6a433dc8788dd2a277e418bf4904203609d", size = 4731873, upload-time = "2026-06-09T22:31:22.253Z" }, + { url = "https://files.pythonhosted.org/packages/4c/4c/9a964756d24a26b3e34dfcb16f961b89838786e6700b635b0d1e3adff4b6/cryptography-48.0.1-cp39-abi3-manylinux_2_31_armv7l.whl", hash = "sha256:43c5835e2cb98c8733d86f57d6fc879b613f5c3478607281c3e36daffc6dd8a6", size = 4330804, upload-time = "2026-06-09T22:31:36.56Z" }, + { url = "https://files.pythonhosted.org/packages/4b/0f/a10f3a6eb12950a10e3a874070283aa2dd5875b2bfd15fad8a3e17b3f13e/cryptography-48.0.1-cp39-abi3-manylinux_2_34_aarch64.whl", hash = "sha256:fe0180af5bf9236518a087e35bf2d9a347d5f5f51e63c579d683ddff424e3d46", size = 4696217, upload-time = "2026-06-09T22:31:13.351Z" }, + { url = "https://files.pythonhosted.org/packages/f3/6f/5cd12f951165ea73ef85266775d97e4c763b2474ccfd816dd69d3a18d6f8/cryptography-48.0.1-cp39-abi3-manylinux_2_34_ppc64le.whl", hash = "sha256:b7a2d1a937a738a881737cec135a38bb61470589b17515b9f73f571d0ae10401", size = 5245252, upload-time = "2026-06-09T22:32:02.193Z" }, + { url = "https://files.pythonhosted.org/packages/68/ab/8aaa12e4516ec4464033ab79b6f3b592bd5a92102467c4ace8a0d970203f/cryptography-48.0.1-cp39-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:b74ca3b8e5ecdd833bf6a002ca41b4793bb27fb8f1c06ffaf2643c9e9140e31b", size = 4731388, upload-time = "2026-06-09T22:32:04.019Z" }, + { url = "https://files.pythonhosted.org/packages/1b/24/50027ea4dca85ec1f40688f3c24fb32ccacd520583c9592c3cc95628e6fb/cryptography-48.0.1-cp39-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:2c37f2461406063b417837f5f3daab668652acd82423efcd7f0a9f04be972de1", size = 4824186, upload-time = "2026-06-09T22:32:18.707Z" }, + { url = "https://files.pythonhosted.org/packages/52/41/04cb5eb17085ade6f50cc611fb657df6a0f5885350de8764ece89c050197/cryptography-48.0.1-cp39-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:86fe77abb1bd87afb251d4d02ada7ecf53a32cee9b67d976abb2e45a13297475", size = 4964539, upload-time = "2026-06-09T22:31:18.793Z" }, + { url = "https://files.pythonhosted.org/packages/36/bf/ed70785c496e89d7e73b7cda2d21f2447fd6d4e821714b8d04ff217fed92/cryptography-48.0.1-cp39-abi3-win32.whl", hash = "sha256:6b2c0c3e6ccf3ade7750f836ef3ee36eea250cc467d45c256895573ac08cc6f1", size = 3282307, upload-time = "2026-06-09T22:30:53.162Z" }, + { url = "https://files.pythonhosted.org/packages/b3/ff/371ea7d252656ee1eb6d83eeeef3d1d0c6baf1d6497687d081ea03814670/cryptography-48.0.1-cp39-abi3-win_amd64.whl", hash = "sha256:9a49ca6c81417f6a5edb50375a60cccdd70fa0a91a5211829dbea74eba94d2ac", size = 3793408, upload-time = "2026-06-09T22:32:15.191Z" }, + { url = "https://files.pythonhosted.org/packages/a9/d3/eb4e394e587341fdad09a09101fa76478ead3a78b0ad63e55c22f0d75c02/cryptography-48.0.1-pp311-pypy311_pp73-macosx_11_0_arm64.whl", hash = "sha256:08a597acce1ff37f347400087776599e2348a3a8bc53b44120e463cd274efe4a", size = 3951747, upload-time = "2026-06-09T22:31:23.871Z" }, + { url = "https://files.pythonhosted.org/packages/e0/4a/3f43451b4f858bfceaaaffc649e6e787e8d4fb332a1d443af39ab02cc8f1/cryptography-48.0.1-pp311-pypy311_pp73-manylinux_2_28_aarch64.whl", hash = "sha256:735824ec41b7f74a7c45fb1591349333e4c696cb6c044e5f46356e560143e4cd", size = 4641226, upload-time = "2026-06-09T22:31:02.532Z" }, + { url = "https://files.pythonhosted.org/packages/73/4e/855584c2c23b09e4ce2d3b9c30e983e679cd60b068c513c6bbdb91e11782/cryptography-48.0.1-pp311-pypy311_pp73-manylinux_2_28_x86_64.whl", hash = "sha256:92a46e1d638daa264ba2971c0b0489c9409787943efae4d60ffda3d091ef832c", size = 4668958, upload-time = "2026-06-09T22:32:06.213Z" }, + { url = "https://files.pythonhosted.org/packages/42/3b/d35750e41d803d1e516fd6d6011f065424924da7af1748cef4cc9cb3ede1/cryptography-48.0.1-pp311-pypy311_pp73-manylinux_2_34_aarch64.whl", hash = "sha256:7e234ac052af99f2700826a5c29ea99d9c1b1f80341cde62d11c8154dc8e0bd9", size = 4640793, upload-time = "2026-06-09T22:32:26.331Z" }, + { url = "https://files.pythonhosted.org/packages/ca/aa/cdb7181fe865285e87e96825aaab239400f1de0c3bfba9bd9769b79f1a92/cryptography-48.0.1-pp311-pypy311_pp73-manylinux_2_34_x86_64.whl", hash = "sha256:33842cf0888951cef5bc7ac724ab844a42044c1727b967b7f8997289a0464f92", size = 4668505, upload-time = "2026-06-09T22:31:27.534Z" }, + { url = "https://files.pythonhosted.org/packages/5d/8c/ce3823c06c2804f194f9e64f0d67fa3f4094a39f2bb1a990cd03603af8fc/cryptography-48.0.1-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:6184ca7b174f28d7c703f1290d4b297217c45355f77a98f67e9b7f14549ac54a", size = 3742204, upload-time = "2026-06-09T22:31:34.773Z" }, ] [[package]]