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
33 changes: 27 additions & 6 deletions rust/lance-table/src/transaction/manifest_build.rs
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ use crate::io::{
manifest::{read_manifest, read_manifest_indexes},
};
use crate::rowids::version::build_version_meta;
use crate::system_index::frag_reuse::FRAG_REUSE_INDEX_NAME;
use crate::system_index::is_system_index;
use crate::system_index::mem_wal::{
CompactedSsTable, IndexCatchupProgress, MEM_WAL_INDEX_NAME, load_mem_wal_index_details,
Expand Down Expand Up @@ -421,16 +422,24 @@ impl Transaction {
));
}

if config.migration_next_row_id.is_some() && !current_indices.is_empty() {
let names: Vec<&str> = current_indices
.iter()
.map(|idx| idx.name.as_str())
.collect();
// The fragment-reuse index is internal bookkeeping registered by
// compaction with deferred index remap, not something a user can
// meaningfully drop and recreate, so it must not gate the migration.
// It is dropped from the migrated manifest below, which is what makes
// ignoring it here safe. Every other index, including the MemWAL
// index, still blocks: only the fragment-reuse index is known to be
// discardable.
let blocking_indices: Vec<&str> = current_indices
.iter()
.filter(|idx| idx.name != FRAG_REUSE_INDEX_NAME)
.map(|idx| idx.name.as_str())
.collect();
if config.migration_next_row_id.is_some() && !blocking_indices.is_empty() {
return Err(Error::invalid_input(format!(
"Cannot migrate to stable row IDs while indexes exist on the dataset. \
Drop the following indexes first, then re-run the migration, and \
recreate them afterwards: {}",
names.join(", ")
blocking_indices.join(", ")
)));
}
let mut reference_paths = match current_manifest {
Expand Down Expand Up @@ -492,6 +501,18 @@ impl Transaction {
let mut final_fragments = Vec::new();
let mut final_indices = current_indices;

// A fragment-reuse index maps old row *addresses* to new ones, and the
// read path attaches it to every index it opens without checking
// whether the dataset uses stable row ids. Carrying it past the
// migration would therefore rewrite freshly issued row ids as if they
// were addresses, and rows whose new id happens to fall in the old
// address range would disappear from indexed queries. Nothing needs it
// afterwards either, since compaction rejects deferred index remap on
// a stable-row-id dataset.
if config.migration_next_row_id.is_some() {
final_indices.retain(|idx| idx.name != FRAG_REUSE_INDEX_NAME);
}

// Snapshot taken before the operation rewrites the list, so coverage can
// be compared against what each logical index looked like going in. Only
// tables with a MemWAL index maintain coverage, so every other commit --
Expand Down
139 changes: 139 additions & 0 deletions rust/lance/src/dataset/tests/dataset_migrations.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ use crate::dataset::optimize::{CompactionOptions, compact_files};
use crate::index::DatasetIndexExt;
use crate::utils::test::copy_test_data_to_tmp;
use crate::{Dataset, Result};
use lance_index::frag_reuse::FRAG_REUSE_INDEX_NAME;
use lance_index::{IndexCriteria, IndexType, scalar::ScalarIndexParams};
use lance_table::feature_flags::FLAG_STABLE_ROW_IDS;
use lance_table::format::{Fragment, IndexMetadata, RowIdMeta};
Expand Down Expand Up @@ -797,6 +798,144 @@ async fn test_migrate_to_stable_row_ids_basic() {
dataset.validate().await.unwrap();
}

#[tokio::test]
async fn test_migrate_to_stable_row_ids_ignores_system_indices() {
// Compaction with deferred index remap registers the fragment-reuse
// system index (`__lance_frag_reuse`) for rewritten, indexed fragments.
// It is internal bookkeeping, not a user index the migration would
// invalidate, and the user may have already dropped every user index —
// the migration must not be gated on it.
let uri = "memory://migrate_system_idx";
let schema = Arc::new(ArrowSchema::new(vec![ArrowField::new(
"values",
DataType::Int64,
false,
)]));
let batch = RecordBatch::try_new(
schema.clone(),
vec![Arc::new(Int64Array::from_iter_values(0..10))],
)
.unwrap();
let mut dataset = Dataset::write(
RecordBatchIterator::new(vec![Ok(batch)], schema.clone()),
uri,
Some(WriteParams {
max_rows_per_file: 5,
..Default::default()
}),
)
.await
.unwrap();
assert_eq!(dataset.get_fragments().len(), 2);

dataset
.create_index(
&["values"],
IndexType::Scalar,
None,
&ScalarIndexParams::default(),
false,
)
.await
.unwrap();

compact_files(
&mut dataset,
CompactionOptions {
defer_index_remap: true,
..Default::default()
},
None,
)
.await
.unwrap();

// Drop the user index; only the bookkeeping system index remains.
let user_index_name = dataset
.load_indices()
.await
.unwrap()
.iter()
.map(|idx| idx.name.clone())
.find(|name| name != FRAG_REUSE_INDEX_NAME)
.expect("user index should exist after compaction");
dataset.drop_index(&user_index_name).await.unwrap();
let index_names: Vec<String> = dataset
.load_indices()
.await
.unwrap()
.iter()
.map(|idx| idx.name.clone())
.collect();
assert_eq!(
index_names,
vec![FRAG_REUSE_INDEX_NAME.to_string()],
"precondition: only the fragment-reuse system index remains"
);

dataset.migrate_to_stable_row_ids().await.unwrap();
assert!(
dataset.manifest.uses_stable_row_ids(),
"migration must succeed with only the fragment-reuse index present"
);

// The migration must drop the fragment-reuse index rather than carry it
// forward. It maps old row addresses to new ones, and the read path
// attaches it to every index it opens, so a survivor would rewrite the
// freshly issued stable row ids as if they were addresses.
let remaining: Vec<String> = dataset
.load_indices()
.await
.unwrap()
.iter()
.map(|idx| idx.name.clone())
.collect();
assert!(
remaining.is_empty(),
"migration must leave no indices behind, got {remaining:?}"
);

// An index built after the migration must return the row it covers. With
// the fragment-reuse index still attached this lookup finds nothing: the
// remapper rewrites row id 0 into the compacted fragment's address space,
// which no live row carries.
dataset
.create_index(
&["values"],
IndexType::Scalar,
None,
&ScalarIndexParams::default(),
true,
)
.await
.unwrap();
let matched = dataset
.scan()
.filter("values = 0")
.unwrap()
.try_into_batch()
.await
.unwrap()
.num_rows();
assert_eq!(
matched, 1,
"an index built after the migration must find the row it covers"
);

// Deleting enough rows to materialize a compaction task (the default
// materialize-deletions threshold is 10%) exercises the commit path once
// more on the migrated dataset.
dataset.delete("values >= 7").await.unwrap();
let metrics = compact_files(&mut dataset, CompactionOptions::default(), None)
.await
.unwrap();
assert!(
metrics.fragments_removed > 0,
"precondition: the compaction must do real work, got metrics {metrics:?}"
);
dataset.validate().await.unwrap();
}

#[tokio::test]
async fn test_migrate_to_stable_row_ids_already_migrated() {
// Create a dataset that already uses stable row IDs.
Expand Down
Loading