Binary-copy compaction corrupts files whose physical column order differs from the schema

[mmatczuk]

Binary-copy compaction corrupts files whose physical column order differs from the schema

Description

compact_files with CompactionMode::TryBinaryCopy copies pages in physical-column order, then records the compacted file with the schema-order mapping. If all source files have the same non-schema physical order, the eligibility check accepts them and the compacted file maps each logical column to another column's pages. A debug scan panics when Decimal128(38, 10) decodes a UInt64 inline constant; a release scan can instead return wrong values.

Steps to reproduce

The complete program below uses only the public Rust API:

1. Create an empty v_dec: Decimal128(38, 10), v_u64: UInt64 dataset at file version 2.3 with one row per file.
2. Append two batches with the fields ordered (v_u64, v_dec), so both files have fields=[1, 0] and column_indices=[0, 1].
3. Compact with CompactionMode::TryBinaryCopy.
4. Run an unfiltered scan.

From this checkout:

cd bugs
cargo run --bin decimal128-scalar-inline-length-mismatch

The repro asserts that compaction changed the recorded mapping to fields=[0, 1], then scans the crossed pages.

Create Cargo.toml:

[package]
name = "lance-binary-copy-repro"
version = "0.1.0"
edition = "2024"

[dependencies]
arrow = "58"
futures = "0.3"
tempfile = "3"
tokio = { version = "1.23", features = ["macros", "rt"] }
lance = { git = "https://github.com/lance-format/lance", tag = "v11.0.0-beta.4", default-features = false }

Create src/main.rs:

//! Binary-copy compaction loses a data file's physical column order, so a scan decodes one
//! column's pages with another column's data type.
//!
//! Minimized from `crashes/dataset_ops/scalar-inline-constant-length-mismatch-decimaln-expecte-d65eb3bf.bin`
//! (panic inside Lance at `lance-arrow/src/scalar.rs:156`, reached from a scan).
//!
//! A data file does not have to store its columns in dataset-schema order: when the batch handed
//! to a write has its columns in a different order, the writer lays the file out in *batch* order
//! and records the real mapping in `DataFile::fields` / `DataFile::column_indices`. Readers use
//! that mapping, so such a file reads back correctly.
//!
//! `rewrite_files_binary_copy` (`lance/src/dataset/optimize/binary_copy.rs`) copies pages
//! source-column-`i` -> output-column-`i`, but then *recomputes* the output mapping from the
//! dataset schema (`compute_field_column_indices`, same file). Any source file whose physical
//! order was not the schema order therefore gets a mapping that no longer describes it, and every
//! column of the compacted file is crossed. `can_use_binary_copy` only rejects groups whose files
//! *disagree with each other* (`optimize.rs:590`), so a group of uniformly rotated files sails
//! through.
//!
//! Here the compacted file's decimal column is really the UInt64 column, whose one-row page is a
//! constant page carrying an 8-byte inline value. Decoding it as `Decimal128(38, 10)` trips the
//! debug assert in `lance_arrow::scalar::decode_scalar_from_inline_value`:
//!
//! ```text
//! assertion `left == right` failed: Inline constant length mismatch for Decimal128(38, 10): expected 16 bytes but got 8
//! ```
//!
//! In a release build (no debug assertions) the same file decodes 8 bytes as a 16-byte decimal
//! instead, i.e. silently wrong data, so the panic is the visible face of a data-corruption bug.
//!
//! Constant pages only exist in file format 2.2 and newer (2.1 has no constant page encoding), so
//! the assert needs `data_storage_version` >= 2.2; verified on both 2.2 and 2.3.
//!
//! Run with: cargo run --bin decimal128-scalar-inline-length-mismatch

use std::sync::Arc;

use arrow::array::{ArrayRef, Decimal128Builder, RecordBatch, RecordBatchIterator, UInt64Array};
use arrow::datatypes::{DataType, Field, Schema};
use futures::TryStreamExt as _;
use lance::Dataset;
use lance::dataset::optimize::{CompactionMode, CompactionOptions, compact_files};
use lance::dataset::{WriteMode, WriteParams};

const DEC: DataType = DataType::Decimal128(38, 10);

/// Dataset schema. The decimal column comes first so that its crossed page is the first one
/// decoded: that way the panic names `Decimal128`, exactly like the fuzz artifact.
fn schema() -> Arc<Schema> {
    Arc::new(Schema::new(vec![
        Field::new("v_dec", DEC, true),
        Field::new("v_u64", DataType::UInt64, true),
    ]))
}

/// One row, with the batch's columns in the *opposite* order to the dataset schema. This is all it
/// takes to get a data file whose physical column order is not the schema order.
fn swapped_row(id: u64) -> RecordBatch {
    let mut dec = Decimal128Builder::new()
        .with_precision_and_scale(38, 10)
        .expect("decimal builder");
    dec.append_value(i128::from(id) * 1_000_000_000_000_000_000_000);
    let fields = vec![
        Field::new("v_u64", DataType::UInt64, true),
        Field::new("v_dec", DEC, true),
    ];
    let columns: Vec<ArrayRef> = vec![
        Arc::new(UInt64Array::from(vec![id])),
        Arc::new(dec.finish()),
    ];
    RecordBatch::try_new(Arc::new(Schema::new(fields)), columns).expect("swapped batch")
}

fn reader(batches: Vec<RecordBatch>) -> Box<dyn arrow::array::RecordBatchReader + Send> {
    let schema = batches[0].schema();
    Box::new(RecordBatchIterator::new(batches.into_iter().map(Ok), schema))
}

/// `(field_ids, column_indices)` of every data file, plus a printout.
fn layouts(dataset: &Dataset, when: &str) -> Vec<(Vec<i32>, Vec<i32>)> {
    let mut out = Vec::new();
    println!("{when}: {} fragment(s)", dataset.get_fragments().len());
    for frag in dataset.get_fragments() {
        for file in &frag.metadata().files {
            println!(
                "  fragment {} rows={:?} fields={:?} column_indices={:?}",
                frag.id(),
                frag.metadata().physical_rows,
                file.fields.as_ref(),
                file.column_indices.as_ref()
            );
            out.push((file.fields.to_vec(), file.column_indices.to_vec()));
        }
    }
    out
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let dir = tempfile::tempdir().expect("temp dir");
    let uri = dir.path().join("ds").to_str().expect("utf8").to_owned();

    let params = WriteParams {
        mode: WriteMode::Overwrite,
        // One row per file: two fragments for compaction to merge, and a single-value page per
        // column, which the writer stores as a constant page with an inline value.
        max_rows_per_file: 1,
        data_storage_version: Some("2.3".parse().expect("file version")),
        ..Default::default()
    };

    // Empty first write: it fixes the dataset schema without creating a canonically laid out
    // fragment. A mix of layouts would make `can_use_binary_copy` bail out and hide the bug.
    let mut dataset = Dataset::write(
        reader(vec![RecordBatch::new_empty(schema())]),
        &uri,
        Some(params.clone()),
    )
    .await
    .expect("create dataset");

    dataset
        .append(
            reader(vec![swapped_row(201), swapped_row(202)]),
            Some(WriteParams {
                mode: WriteMode::Append,
                ..params
            }),
        )
        .await
        .expect("append");

    // Precondition: both files really are laid out in batch order (field 1 = v_u64 in column 0,
    // field 0 = v_dec in column 1), and they agree with each other.
    let before = layouts(&dataset, "after appending column-swapped rows");
    assert_eq!(
        before,
        vec![(vec![1, 0], vec![0, 1]), (vec![1, 0], vec![0, 1])],
        "precondition failed: the appended files are not laid out in batch order"
    );

    compact_files(
        &mut dataset,
        CompactionOptions {
            target_rows_per_fragment: 8,
            materialize_deletions: false,
            compaction_mode: Some(CompactionMode::TryBinaryCopy),
            ..Default::default()
        },
        None,
    )
    .await
    .expect("compact");

    // The bug: pages were copied column-for-column, but the new file claims the canonical
    // schema-order mapping, so column 1 (really v_u64) is now described as the decimal column.
    let after = layouts(&dataset, "after binary-copy compaction");
    assert_eq!(
        after,
        vec![(vec![0, 1], vec![0, 1])],
        "the compacted file did not get the canonical mapping (was binary copy skipped?)"
    );

    // Panics: `Inline constant length mismatch for Decimal128(38, 10): expected 16 bytes but got 8`
    let batches: Vec<RecordBatch> = dataset
        .scan()
        .try_into_stream()
        .await
        .expect("plan scan")
        .try_collect()
        .await
        .expect("run scan");

    for batch in &batches {
        println!("scanned: {:?}", batch.columns());
    }
    println!("no panic: the bug did not reproduce (data above may still be crossed)");
}

Run:

cargo run

Expected behavior

Binary-copy compaction must preserve the source files' physical-column mapping, or decline binary copy for non-schema-order files. A scan after compaction must return the same decimal and integer values written before compaction.

Lance version

v11.0.0-beta.4

Language binding

Rust

Environment

Linux x86_64, local filesystem

Logs / traceback

after appending column-swapped rows: 2 fragment(s)
  fragment 0 rows=Some(1) fields=[1, 0] column_indices=[0, 1]
  fragment 1 rows=Some(1) fields=[1, 0] column_indices=[0, 1]
after binary-copy compaction: 1 fragment(s)
  fragment 2 rows=Some(2) fields=[0, 1] column_indices=[0, 1]

thread 'main' panicked at .../lance-arrow/src/scalar.rs:156:9:
assertion `left == right` failed: Inline constant length mismatch for Decimal128(38, 10): expected 16 bytes but got 8
  left: 8
 right: 16