Error during vacuum: the length + offset of the sliced StructArray cannot exceed the existing length

[Menziess]

Environment

deltalake==0.25.0:

Binding: Python3.11?

Environment:

• Azure:
• Mac OS:

---

Bug

What happened:

During vacuuming the delta table, this error was raised:

PanicException: the length + offset of the sliced StructArray cannot exceed the existing length.

delta.vacuum(
    retention_hours=1,
    dry_run=False,
    enforce_retention_duration=False
)

What you expected to happen:

The program has worked using deltalake==0.23.2 before. I expected the delta table to be vacuum'd after being optimized without raising exceptions.

How to reproduce it:

Not quite sure yet.

More details:

INFO:main.py:Optimizing delta table.
INFO:main.py:Vacuuming delta table.
thread '<unnamed>' panicked at /Users/runner/.cargo/registry/src/index.crates.io-6f17d22bba15001f/arrow-array-54.2.0/src/array/struct_array.rs:275:9:
the length + offset of the sliced StructArray cannot exceed the existing length
stack backtrace:
   0: _rust_begin_unwind
   1: core::panicking::panic_fmt
   2: arrow_array::array::struct_array::StructArray::slice
   3: <arrow_array::array::struct_array::StructArray as arrow_array::array::Array>::slice
   4: <alloc::vec::Vec<T> as alloc::vec::spec_from_iter::SpecFromIter<T,I>>::from_iter
   5: arrow_array::array::struct_array::StructArray::slice
   6: <arrow_array::array::struct_array::StructArray as arrow_array::array::Array>::slice
   7: <alloc::vec::Vec<T> as alloc::vec::spec_from_iter::SpecFromIter<T,I>>::from_iter
   8: arrow_array::array::struct_array::StructArray::slice
   9: <arrow_array::array::struct_array::StructArray as arrow_array::array::Array>::slice
  10: <alloc::vec::Vec<T> as alloc::vec::spec_from_iter::SpecFromIter<T,I>>::from_iter
  11: arrow_array::array::struct_array::StructArray::slice
  12: <arrow_array::array::struct_array::StructArray as arrow_array::array::Array>::slice
  13: arrow_select::filter::filter_array
  14: <alloc::vec::Vec<T> as alloc::vec::spec_from_iter::SpecFromIter<T,I>>::from_iter
  15: core::iter::adapters::try_process
  16: arrow_select::filter::filter_record_batch
  17: deltalake_core::kernel::snapshot::replay::LogReplayScanner::process_files_batch
  18: <core::iter::adapters::map::Map<I,F> as core::iter::traits::iterator::Iterator>::try_fold
  19: <core::iter::adapters::chain::Chain<A,B> as core::iter::traits::iterator::Iterator>::try_fold
  20: <alloc::vec::Vec<T> as alloc::vec::spec_from_iter::SpecFromIter<T,I>>::from_iter
  21: core::iter::adapters::try_process
  22: deltalake_core::kernel::snapshot::EagerSnapshot::advance
  23: deltalake_core::operations::transaction::PostCommit::run_post_commit_hook::{{closure}}
  24: <deltalake_core::operations::transaction::PostCommit as core::future::into_future::IntoFuture>::into_future::{{closure}}
  25: <deltalake_core::operations::transaction::PreCommit as core::future::into_future::IntoFuture>::into_future::{{closure}}
  26: <deltalake_core::operations::vacuum::VacuumBuilder as core::future::into_future::IntoFuture>::into_future::{{closure}}
  27: tokio::runtime::park::CachedParkThread::block_on
  28: tokio::runtime::context::runtime::enter_runtime
  29: tokio::runtime::runtime::Runtime::block_on
  30: deltalake::RawDeltaTable::vacuum
  31: deltalake::RawDeltaTable::__pymethod_vacuum__
  32: pyo3::impl_::trampoline::trampoline
  33: deltalake::<impl pyo3::impl_::pyclass::PyMethods<deltalake::RawDeltaTable> for pyo3::impl_::pyclass::PyClassImplCollector<deltalake::RawDeltaTable>>::py_methods::ITEMS::trampoline
  34: _cfunction_call
  35: __PyObject_MakeTpCall
  36: __PyEval_EvalFrameDefault
  37: __PyEval_Vector
  38: _context_run
  39: _cfunction_vectorcall_FASTCALL_KEYWORDS
  40: _partial_vectorcall
  41: __PyEval_EvalFrameDefault
  42: __PyEval_Vector
  43: __PyEval_EvalFrameDefault
  44: __PyEval_Vector
  45: _method_vectorcall
  46: __PyEval_EvalFrameDefault
  47: __PyEval_Vector
  48: __PyObject_FastCallDictTstate
  49: __PyObject_Call_Prepend
  50: _slot_tp_call
  51: __PyObject_Call
  52: _thread_run
  53: _pythread_wrapper
  54: __pthread_deallocate

[ion-elgreco]

Please create an MRE

[Menziess]

I don't know where to start yet. It's raised in StructArray within the rust library: link.

/// Returns a zero-copy slice of this array with the indicated offset and length.
pub fn slice(&self, offset: usize, len: usize) -> Self {
    assert!(
        offset.saturating_add(len) <= self.len,
        "the length + offset of the sliced StructArray cannot exceed the existing length"
    );

    let fields = self.fields.iter().map(|a| a.slice(offset, len)).collect();

    Self {
        len,
        data_type: self.data_type.clone(),
        nulls: self.nulls.as_ref().map(|n| n.slice(offset, len)),
        fields,
    }
}

[thorfour]

We're currently running into this quite frequently. I've noticed that it only happens when record batch has a single add row Nevermind recreated with a record with two rows. Trying to create an MRE to replicate.

For us the slice is panicing because the stats_parsed struct contains columns of length 0. Trying to determine how this happens.

[thorfour]

#[test]
    fn test_filter_empty_struct() {
        /*
            "a": {
                "b": int64,
                "c": {}
            },
        */
        let fields = arrow_schema::Field::new(
            "a",
            arrow_schema::DataType::Struct(arrow_schema::Fields::from(vec![
                arrow_schema::Field::new("b", arrow_schema::DataType::Int64, true),
                arrow_schema::Field::new(
                    "c",
                    arrow_schema::DataType::Struct(arrow_schema::Fields::empty()),
                    true,
                ),
            ])),
            true,
        );

        /* Test record
            {"a":{"c": {}}}
            {"a":{"c": {}}}
            {"a":{"c": {}}}
        */

        // Create the record batch with the nested struct array
        let schema = Arc::new(Schema::new(vec![fields]));

        let b = Arc::new(Int64Array::from(vec![None, None, None]));
        let c = Arc::new(StructArray::new_empty_fields(
            3,
            Some(NullBuffer::from(vec![true, true, true])),
        ));
        let a = StructArray::new(
            vec![
                Field::new("b", DataType::Int64, true),
                Field::new("c", DataType::Struct(Fields::empty()), true),
            ]
            .into(),
            vec![b.clone(), c.clone()],
            Some(NullBuffer::from(vec![true, true, true])),
        );
        let record_batch = RecordBatch::try_new(schema, vec![Arc::new(a)]).unwrap();
        println!("{record_batch:?}");

        // Apply the filter
        let predicate = BooleanArray::from(vec![true, false, true]);
        let filtered_batch = filter_record_batch(&record_batch, &predicate).unwrap();

        // The filtered batch should have 2 rows (the 1st and 3rd)
        assert_eq!(filtered_batch.num_rows(), 2);

        // Get the filtered array
        let filtered_array = filtered_batch.column(0);
        let filtered_struct = filtered_array
            .as_any()
            .downcast_ref::<StructArray>()
            .unwrap();

        // Verify the null buffer was properly filtered
        let expected_nulls = NullBuffer::from(vec![true, true]);
        assert_eq!(filtered_struct.nulls().unwrap(), &expected_nulls);
    }

This recreates the bug, which was "fixed" where it will no longer set an array of len 0. However I think this will probably cause delta-rs to panic earlier. Confirmed, this now panics directly in process_files_batch when filtering a batch, instead of creating a new batch with some len 0 arrays.

Delta was calling filter_record_batch in the log segment update code, and the stats_parsed field contained an empty struct.