Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/audit.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion .github/workflows/breaking_changes_detector.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
2 changes: 1 addition & 1 deletion .github/workflows/dev.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 2 additions & 2 deletions .github/workflows/rust.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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

Expand Down
20 changes: 10 additions & 10 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion datafusion-cli/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
121 changes: 80 additions & 41 deletions datafusion/common/src/heap_size.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<T: DFHeapSize> DFHeapSize for Box<T> {
fn heap_size(&self, ctx: &mut DFHeapSizeCtx) -> usize {
size_of::<T>() + self.as_ref().heap_size(ctx)
Expand Down Expand Up @@ -469,6 +428,29 @@ impl_zero_heap_size!(
DateTime<Utc>,
);

/// 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::*;
Expand Down Expand Up @@ -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::<Int32Type, _, _>(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::<Int32Type, _, _>(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);
}
}
14 changes: 13 additions & 1 deletion datafusion/datasource-parquet/src/row_filter.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -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<usize>,
Expand All @@ -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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -484,6 +489,13 @@ impl TreeNodeVisitor<'_> for PushdownChecker<'_> {
return Ok(recursion);
}

if ScalarFunctionExpr::try_downcast_func::<FileRowIndexFunc>(node.as_ref())
.is_some()
{
self.has_unpushable_udfs = true;
return Ok(TreeNodeRecursion::Jump);
}

Ok(TreeNodeRecursion::Continue)
}
}
Expand Down
Loading
Loading