Parquet 56: encounter `error: item_reader def levels are None` when reading nested field with row filter

[lewiszlw]

Describe the bug

After upgrading parquet to 56, we encountered Parquet error: item_reader def levels are None when reading nested field with row filter.

To Reproduce

[dependencies]
arrow = "56"
parquet = { version = "56", features = ["async"] }
tokio = { version = "1", features = ["full"] }
futures = "0.3"

use arrow::array::{BooleanArray, Float32Builder, Int32Array, ListBuilder};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
use futures::stream::StreamExt;
use parquet::arrow::arrow_reader::{ArrowPredicateFn, RowFilter};
use parquet::arrow::{AsyncArrowWriter, ParquetRecordBatchStreamBuilder, ProjectionMask};
use std::sync::Arc;
use tokio::fs::File;

#[tokio::main]
async fn main() {
    write_parquet_file().await.unwrap();
    read_parquet_file().await.unwrap();
}

async fn read_parquet_file() -> Result<(), Box<dyn std::error::Error>> {
    let file = File::open("output.parquet").await?;
    let builder = ParquetRecordBatchStreamBuilder::new(file).await?;

    let predicate = ArrowPredicateFn::new(ProjectionMask::all(), |batch| {
        Ok(BooleanArray::from(vec![true; batch.num_rows()]))
    });

    let projection_mask = ProjectionMask::all();

    let mut stream = builder
        .with_row_filter(RowFilter::new(vec![Box::new(predicate)]))
        .with_projection(projection_mask)
        .build()?;

    while let Some(batch) = stream.next().await {
        let batch = batch?;
        println!("Read batch {batch:?}");
    }

    Ok(())
}

async fn write_parquet_file() -> Result<(), Box<dyn std::error::Error>> {
    let list_inner_field = Arc::new(Field::new("item", DataType::Float32, true));
    let table_schema = Arc::new(Schema::new(vec![
        Field::new("id", DataType::Int32, false),
        Field::new("vector", DataType::List(list_inner_field.clone()), true),
    ]));

    let mut list_builder =
        ListBuilder::new(Float32Builder::new()).with_field(list_inner_field.clone());
    list_builder.values().append_slice(&[10.0, 10.0, 10.0]);
    list_builder.append(true);
    list_builder.values().append_slice(&[20.0, 20.0, 20.0]);
    list_builder.append(true);
    list_builder.values().append_slice(&[30.0, 30.0, 30.0]);
    list_builder.append(true);
    list_builder.values().append_slice(&[40.0, 40.0, 40.0]);
    list_builder.append(true);
    let list_array = list_builder.finish();

    let data = vec![RecordBatch::try_new(
        table_schema.clone(),
        vec![
            Arc::new(Int32Array::from(vec![1, 2, 3, 4])),
            Arc::new(list_array),
        ],
    )?];

    let file = File::create("output.parquet").await?;
    let mut writer = AsyncArrowWriter::try_new(file, table_schema, None)?;

    for batch in data {
        writer.write(&batch).await?;
    }

    writer.close().await?;

    println!("Parquet file written successfully!");

    Ok(())
}

thread 'main' panicked at src\main.rs:13:31:
called `Result::unwrap()` on an `Err` value: External(ParquetError("Parquet error: item_reader def levels are None."))
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

Above code works in parquet 55.

Expected behavior

Additional context

[alamb]

I wonder if this is related to this issue (fixed on main/arrow 57):

• Column with List(Struct) causes failed to decode level data for struct array (regression in 56) #8404

[lewiszlw]

Seems no. I tried latest main branch and still encountered the error.

[alamb]

@lewiszlw is this a regression? (aka did it work on 55 but started failing on 56?)

[lewiszlw]

@lewiszlw is this a regression? (aka did it work on 55 but started failing on 56?)

Yes.

[alamb]

I tracked this down to something related to predicate cacheing:

• The test passes at 4a21443

• The test fails at 04f217b - Speed up Parquet filter pushdown v4 (Predicate evaluation cache for async_reader) #7850

[XiangpengHao]

hmmm, I thought nested field will bypass the cache:

arrow-rs/parquet/src/arrow/async_reader/mod.rs

Lines 731 to 759 in 8bbcaef

	fn exclude_nested_columns_from_cache(&self, mask: &ProjectionMask) -> Option<ProjectionMask> {
	let schema = self.metadata.file_metadata().schema_descr();
	let num_leaves = schema.num_columns();
	// Count how many leaves each root column has
	let num_roots = schema.root_schema().get_fields().len();
	let mut root_leaf_counts = vec![0usize; num_roots];
	for leaf_idx in 0..num_leaves {
	let root_idx = schema.get_column_root_idx(leaf_idx);
	root_leaf_counts[root_idx] += 1;
	}
	// Keep only leaves whose root has exactly one leaf (non-nested)
	let mut included_leaves = Vec::new();
	for leaf_idx in 0..num_leaves {
	if mask.leaf_included(leaf_idx) {
	let root_idx = schema.get_column_root_idx(leaf_idx);
	if root_leaf_counts[root_idx] == 1 {
	included_leaves.push(leaf_idx);
	}
	}
	}
	if included_leaves.is_empty() {
	None
	} else {
	Some(ProjectionMask::leaves(schema, included_leaves))
	}
	}

(This is probably the place to fix this)

[alamb]

I have confirmed this is related to the predicate cache. A workaround is turn off the cache

    let mut stream = builder
        .with_row_filter(RowFilter::new(vec![Box::new(predicate)]))
        .with_projection(projection_mask)
        .with_max_predicate_cache_size(0) // <-- turning off the predicate cache works around the issue
        .build()?;

[alamb]

Here is the schema

Schema: SchemaDescriptor {
    schema: GroupType {
        basic_info: BasicTypeInfo {
            name: "arrow_schema",
            repetition: None,
            converted_type: NONE,
            logical_type: None,
            id: None,
        },
        fields: [
            PrimitiveType {
                basic_info: BasicTypeInfo {
                    name: "id",
                    repetition: Some(
                        REQUIRED,
                    ),
                    converted_type: NONE,
                    logical_type: None,
                    id: None,
                },
                physical_type: INT32,
                type_length: -1,
                scale: -1,
                precision: -1,
            },
            GroupType {
                basic_info: BasicTypeInfo {
                    name: "vector",
                    repetition: Some(
                        OPTIONAL,
                    ),
                    converted_type: LIST,
                    logical_type: Some(
                        List,
                    ),
                    id: None,
                },
                fields: [
                    GroupType {
                        basic_info: BasicTypeInfo {
                            name: "list",
                            repetition: Some(
                                REPEATED,
                            ),
                            converted_type: NONE,
                            logical_type: None,
                            id: None,
                        },
                        fields: [
                            PrimitiveType {
                                basic_info: BasicTypeInfo {
                                    name: "item",
                                    repetition: Some(
                                        OPTIONAL,
                                    ),
                                    converted_type: NONE,
                                    logical_type: None,
                                    id: None,
                                },
                                physical_type: FLOAT,
                                type_length: -1,
                                scale: -1,
                                precision: -1,
                            },
                        ],
                    },
                ],
            },
        ],
    },
}

here is the mask (no mask == all columns)

Projection mask: ProjectionMask {
    mask: None,
}

[lewiszlw]

This workround works on latest main branch, but not works on parquet 56.

[alamb]

This workround works on latest main branch, but not works on parquet 56.

I suspect this is due to the fix from @nuno-faria

• [Parquet] Avoid fetching multiple pages when the predicate cache is disabled #8554