ZoneMap index becomes stale after update and causes a later update to match zero rows

[mmatczuk]

Description

With stable row IDs enabled, an ordinary update can leave a ZoneMap scalar index pointing at the pre-update row addresses. An index-enabled query then returns no rows while a flat scan returns the live row. More seriously, a later UpdateBuilder::update_where using the same indexed predicate returns success with rows_updated: 0, so the write is not applied and no error explains why.

No compaction or deferred index remap is involved, and the dataset uses the default storage version. The same sequence without an index updates both times, and a BTree index over the same column also updates both times.

The update implementation executes its predicate through a normal dataset scan, then commits the rewritten rows as an Operation::Update. For stable-row-ID row rewrites, the transaction registers the new fragments in indices covering unchanged columns by adding their IDs to the index fragment bitmap. However, ZoneMap returns physical row addresses and does not support remapping. The result is an index whose metadata covers the rewritten fragment but whose stored zones still address the old fragment layout.

This is related to #8221, where deferred-remap compaction leaves a BloomFilter index stale. This reproducer shows that a plain update can trigger the same class of false negative, and that the stale index can also cause a later write to match no rows.

Steps to reproduce

Create an empty directory with these two files.

Cargo.toml:

[package]
name = "repro-scalar-index-stale-after-update"
version = "0.1.0"
edition = "2021"

[dependencies]
arrow = "58"
futures = "0.3"
lance = { git = "https://github.com/lance-format/lance", tag = "v10.1.0-beta.2", default-features = false }
lance-index = { git = "https://github.com/lance-format/lance", tag = "v10.1.0-beta.2", default-features = false }
tokio = { version = "1.23", features = ["rt"] }

src/main.rs:

use std::sync::Arc;

use arrow::array::{Int64Array, RecordBatch, RecordBatchIterator, UInt64Array};
use arrow::datatypes::{DataType, Field, Schema};
use futures::TryStreamExt as _;
use lance::dataset::{UpdateBuilder, WriteMode, WriteParams};
use lance::index::DatasetIndexExt as _;
use lance::Dataset;
use lance_index::scalar::{BuiltinIndexType, ScalarIndexParams};
use lance_index::IndexType;

fn schema() -> Arc<Schema> {
    Arc::new(Schema::new(vec![
        Field::new("id", DataType::UInt64, false),
        Field::new("v0", DataType::Int64, true),
    ]))
}

fn batch(ids: Vec<u64>, vals: Vec<i64>) -> RecordBatch {
    RecordBatch::try_new(
        schema(),
        vec![
            Arc::new(UInt64Array::from(ids)),
            Arc::new(Int64Array::from(vals)),
        ],
    )
    .expect("build batch")
}

async fn ids_matching(dataset: &Dataset, value: u64, use_scalar_index: bool) -> Vec<u64> {
    let mut scan = dataset.scan();
    scan.project(&["id"]).expect("project");
    scan.filter(&format!("id = {value}")).expect("filter");
    scan.use_scalar_index(use_scalar_index);
    let batches: Vec<RecordBatch> = scan
        .try_into_stream()
        .await
        .expect("plan")
        .try_collect()
        .await
        .expect("collect");
    let mut ids: Vec<u64> = batches
        .iter()
        .flat_map(|batch| {
            batch
                .column_by_name("id")
                .expect("id column")
                .as_any()
                .downcast_ref::<UInt64Array>()
                .expect("u64 ids")
                .values()
                .to_vec()
        })
        .collect();
    ids.sort_unstable();
    ids
}

async fn v0_matching(dataset: &Dataset, id: u64) -> Vec<i64> {
    let mut scan = dataset.scan();
    scan.project(&["v0"]).expect("project");
    scan.filter(&format!("id = {id}")).expect("filter");
    scan.use_scalar_index(false);
    let batches: Vec<RecordBatch> = scan
        .try_into_stream()
        .await
        .expect("plan")
        .try_collect()
        .await
        .expect("collect");
    batches
        .iter()
        .flat_map(|batch| {
            batch
                .column_by_name("v0")
                .expect("v0 column")
                .as_any()
                .downcast_ref::<Int64Array>()
                .expect("i64 values")
                .iter()
                .flatten()
        })
        .collect()
}

async fn update_v0(dataset: &mut Dataset, value: i64) -> u64 {
    let result = UpdateBuilder::new(Arc::new(dataset.clone()))
        .update_where("id = 0")
        .expect("update_where")
        .set("v0", &value.to_string())
        .expect("set")
        .build()
        .expect("build")
        .execute()
        .await
        .expect("execute update");
    let rows = result.rows_updated;
    *dataset = (*result.new_dataset).clone();
    rows
}

struct Case {
    first: u64,
    second: u64,
    indexed: Vec<u64>,
    flat: Vec<u64>,
    v0: Vec<i64>,
}

async fn run_case(uri: &str, index: Option<BuiltinIndexType>) -> Case {
    let mut dataset = Dataset::write(
        RecordBatchIterator::new([Ok(batch(vec![0, 168, 255], vec![1, 2, 3]))], schema()),
        uri,
        Some(WriteParams {
            mode: WriteMode::Overwrite,
            enable_stable_row_ids: true,
            max_rows_per_file: 2,
            max_rows_per_group: 4,
            ..Default::default()
        }),
    )
    .await
    .expect("create dataset");

    if let Some(kind) = index {
        let index_type = match kind {
            BuiltinIndexType::ZoneMap => IndexType::ZoneMap,
            BuiltinIndexType::BloomFilter => IndexType::BloomFilter,
            _ => IndexType::Scalar,
        };
        dataset
            .create_index(
                &["id"],
                index_type,
                None,
                &ScalarIndexParams::for_builtin(kind),
                true,
            )
            .await
            .expect("create index");
    }

    let first = update_v0(&mut dataset, 10).await;
    let second = update_v0(&mut dataset, 20).await;

    Case {
        first,
        second,
        indexed: ids_matching(&dataset, 0, true).await,
        flat: ids_matching(&dataset, 0, false).await,
        v0: v0_matching(&dataset, 0).await,
    }
}

async fn run() {
    let control = run_case("memory://repro-noindex", None).await;
    let indexed = run_case("memory://repro-zonemap", Some(BuiltinIndexType::ZoneMap)).await;
    let btree = run_case("memory://repro-btree", Some(BuiltinIndexType::BTree)).await;

    for (label, case) in [
        ("no index (control)", &control),
        ("ZoneMap on `id`", &indexed),
        ("BTree on `id`", &btree),
    ] {
        println!("{label}:");
        println!(
            "  rows_updated: first={} second={}",
            case.first, case.second
        );
        println!("  id = 0 through the index -> {:?}", case.indexed);
        println!("  id = 0 flat scan         -> {:?}", case.flat);
        println!("  v0 after both updates    -> {:?}", case.v0);
    }

    assert_eq!((control.first, control.second), (1, 1));
    assert_eq!(control.flat, vec![0]);
    assert_eq!(control.v0, vec![20]);
    assert_eq!(indexed.flat, vec![0]);
    assert_eq!(
        (
            btree.first,
            btree.second,
            btree.indexed.clone(),
            btree.v0.clone()
        ),
        (1, 1, vec![0], vec![20])
    );
    assert_eq!(
        (indexed.first, indexed.second, indexed.v0.clone()),
        (1, 1, vec![20]),
        "the second update matched zero rows and was not applied; an unindexed read still sees \
         the first update's value"
    );
    assert_eq!(indexed.indexed, indexed.flat);
}

fn main() {
    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("Tokio runtime")
        .block_on(run());
}

Run:

cargo run

Expected behavior

An update that rewrites rows should remap or invalidate scalar indices that store physical row addresses. Index-enabled scans must return the same live rows as flat scans, and a later update must not match zero rows because a prior update left its predicate index stale.

Lance version

v10.1.0-beta.2 (94cca93d13b1c731afd6660d46ad2f8d4b4345c4)

Language binding

Rust

Environment

Fedora Linux 44, x86_64, local in-memory object store, rustc 1.97.1

Logs / traceback

no index (control):
  rows_updated: first=1 second=1
  id = 0 through the index -> [0]
  id = 0 flat scan         -> [0]
  v0 after both updates    -> [20]
ZoneMap on `id`:
  rows_updated: first=1 second=0
  id = 0 through the index -> []
  id = 0 flat scan         -> [0]
  v0 after both updates    -> [10]
BTree on `id`:
  rows_updated: first=1 second=1
  id = 0 through the index -> [0]
  id = 0 flat scan         -> [0]
  v0 after both updates    -> [20]

thread 'main' panicked at src/main.rs:
assertion `left == right` failed: the second update matched zero rows and was not applied; an unindexed read still sees the first update's value
  left: (1, 0, [10])
 right: (1, 1, [20])