From 4fd0d1929343c6e46a92180c106941b30f0333b1 Mon Sep 17 00:00:00 2001 From: Will Jones Date: Tue, 7 Jul 2026 10:59:37 -0700 Subject: [PATCH 1/6] feat(transaction): add NewFragment placeholder + AddIndex action Extend the experimental action-transaction types for the first compound-commit slice. `AddFragments` now carries `NewFragment`s, each with an operation-local `local_id` placeholder so a later action can reference a fragment before its real id is assigned. Adds the `AddIndex` action, which registers an index segment and expresses coverage as the union of already-committed fragment ids and same-op placeholders. All behind the non-default `unstable-action-transactions` feature. Co-Authored-By: Claude Opus 4.8 (1M context) --- protos/transaction_experimental.proto | 52 +++++++-- rust/lance-table/src/transaction.rs | 149 +++++++++++++++++++++++--- 2 files changed, 179 insertions(+), 22 deletions(-) diff --git a/protos/transaction_experimental.proto b/protos/transaction_experimental.proto index f4558da5996..b8fe4b4dff8 100644 --- a/protos/transaction_experimental.proto +++ b/protos/transaction_experimental.proto @@ -60,21 +60,61 @@ message UserAction { // A granular change to the manifest. Traditional operations decompose into an // ordered list of these. // -// EXPERIMENTAL: only `AddFragments` is defined so far, as a proof of shape. -// The remaining actions (RemoveFragments, UpdateDeletionVector, AddIndex, ...) -// land in follow-up vertical slices. +// EXPERIMENTAL: only `AddFragments` and `AddIndex` are defined so far. The +// remaining actions (RemoveFragments, UpdateDeletionVector, ...) land in +// follow-up vertical slices. message Action { oneof action { AddFragments add_fragments = 1; + AddIndex add_index = 2; } } +// A fragment to be appended by this operation, before its final id is assigned. +// +// This is deliberately distinct from `DataFragment`: a committed fragment has an +// assigned `id`, but a fragment being added does not yet, and instead carries an +// operation-local placeholder so later actions can reference it. +message NewFragment { + // Operation-local placeholder token identifying this fragment within the + // enclosing UserOperation, unique among that operation's new fragments. Later + // actions (e.g. AddIndex.covers_local) reference this token; at apply time it + // is resolved to the real fragment id assigned from the manifest counter. + uint32 local_id = 1; + // The fragment payload to append. Its `id` field is IGNORED: the real id is + // assigned at apply time from the manifest's fragment counter (late binding), + // and re-assigned on every commit retry. + DataFragment fragment = 2; +} + // Append new fragments to the table. // // Covers: Append, the "add new rows" part of Overwrite, and new fragments // produced by compaction. message AddFragments { - // The new fragments to append. Fragment IDs are not carried here; they are - // assigned at apply time from the manifest's counters (late binding). - repeated DataFragment fragments = 1; + // The new fragments to append, each with an operation-local placeholder. + repeated NewFragment new_fragments = 1; +} + +// Register a secondary index segment in the manifest. +// +// The index file(s) are expected to have already been written; this action only +// records the index metadata. Coverage is the union of already-committed +// fragment ids (`covers_existing`) and fragments added earlier in this same +// operation (`covers_local`, resolved to their assigned ids at apply time). +message AddIndex { + // Unique identifier of the index across all dataset versions. + string uuid = 1; + // Human-readable index name. Replace-by-uuid semantics apply on the manifest. + string name = 2; + // The field ids the index is built on. + repeated int32 fields = 3; + // Already-committed fragment ids this index covers. + repeated uint64 covers_existing = 4; + // Placeholders (NewFragment.local_id) of same-operation fragments this index + // covers, resolved to real fragment ids at apply time. + repeated uint32 covers_local = 5; + // Optional opaque, type-specific index metadata: a serialized + // google.protobuf.Any. Absent when the index carries no details. + optional bytes index_details = 6; } diff --git a/rust/lance-table/src/transaction.rs b/rust/lance-table/src/transaction.rs index 7bee2ec8146..5a5b0733b70 100644 --- a/rust/lance-table/src/transaction.rs +++ b/rust/lance-table/src/transaction.rs @@ -24,9 +24,9 @@ //! //! # Scope //! -//! Only [`AddFragments`] is implemented, as a proof of shape. The remaining -//! actions and the translation/conflict-resolution machinery land in follow-up -//! vertical slices. +//! [`AddFragments`] and [`AddIndex`] are implemented, enough for the first +//! compound-commit vertical slice (append data + register an index over it in +//! one transaction). The remaining actions land in follow-up slices. use lance_core::deepsize::DeepSizeOf; use lance_core::{Error, Result}; @@ -71,6 +71,27 @@ pub struct UserAction { pub enum Action { /// Append new fragments to the table. AddFragments(AddFragments), + /// Register a secondary index segment in the manifest. + AddIndex(AddIndex), +} + +/// A fragment to be appended, before its final id is assigned. +/// +/// Distinct from [`Fragment`]: a committed fragment has an assigned id, but a +/// fragment being added does not yet, and instead carries an operation-local +/// [`local_id`](Self::local_id) placeholder that later actions can reference. +#[derive(Debug, Clone, PartialEq, DeepSizeOf)] +pub struct NewFragment { + /// Operation-local placeholder token identifying this fragment within the + /// enclosing [`UserOperation`], unique among that operation's new fragments. + /// Later actions (e.g. [`AddIndex::covers_local`]) reference this token; at + /// apply time it is resolved to the real id assigned from the manifest + /// counter. + pub local_id: u32, + /// The fragment payload to append. Its `id` field is ignored: the real id is + /// assigned at apply time from the manifest's fragment counter (late + /// binding), and re-assigned on every commit retry. + pub fragment: Fragment, } /// Append new fragments to the table. @@ -79,18 +100,65 @@ pub enum Action { /// produced by compaction. #[derive(Debug, Clone, PartialEq, DeepSizeOf)] pub struct AddFragments { - /// The new fragments to append. Fragment IDs are assigned at apply time - /// from the manifest's counters (late binding), not carried here. - pub fragments: Vec, + /// The new fragments to append, each with an operation-local placeholder. + pub new_fragments: Vec, +} + +/// Register a secondary index segment in the manifest. +/// +/// The index file(s) are expected to have already been written; this action only +/// records the metadata. Coverage is the union of already-committed fragment ids +/// ([`covers_existing`](Self::covers_existing)) and fragments added earlier in +/// this same operation ([`covers_local`](Self::covers_local), resolved to their +/// assigned ids at apply time). +#[derive(Debug, Clone, PartialEq, DeepSizeOf)] +pub struct AddIndex { + /// Unique identifier of the index across all dataset versions. + pub uuid: String, + /// Human-readable index name. Replace-by-uuid semantics apply on the manifest. + pub name: String, + /// The field ids the index is built on. + pub fields: Vec, + /// Already-committed fragment ids this index covers. + pub covers_existing: Vec, + /// Placeholders ([`NewFragment::local_id`]) of same-operation fragments this + /// index covers, resolved to real fragment ids at apply time. + pub covers_local: Vec, + /// Optional opaque, type-specific index metadata: a serialized + /// `google.protobuf.Any`. `None` when the index carries no details. + pub index_details: Option>, +} + +impl From<&NewFragment> for pb::NewFragment { + fn from(new_fragment: &NewFragment) -> Self { + Self { + local_id: new_fragment.local_id, + fragment: Some(pb::DataFragment::from(&new_fragment.fragment)), + } + } +} + +impl TryFrom for NewFragment { + type Error = Error; + + fn try_from(proto: pb::NewFragment) -> Result { + let fragment = proto.fragment.ok_or_else(|| { + Error::invalid_input("NewFragment is missing its fragment".to_string()) + })?; + Ok(Self { + local_id: proto.local_id, + fragment: Fragment::try_from(fragment)?, + }) + } } impl From<&AddFragments> for pb::AddFragments { fn from(action: &AddFragments) -> Self { Self { - fragments: action - .fragments + new_fragments: action + .new_fragments .iter() - .map(pb::DataFragment::from) + .map(pb::NewFragment::from) .collect(), } } @@ -101,19 +169,48 @@ impl TryFrom for AddFragments { fn try_from(proto: pb::AddFragments) -> Result { Ok(Self { - fragments: proto - .fragments + new_fragments: proto + .new_fragments .into_iter() - .map(Fragment::try_from) + .map(NewFragment::try_from) .collect::>()?, }) } } +impl From<&AddIndex> for pb::AddIndex { + fn from(action: &AddIndex) -> Self { + Self { + uuid: action.uuid.clone(), + name: action.name.clone(), + fields: action.fields.clone(), + covers_existing: action.covers_existing.clone(), + covers_local: action.covers_local.clone(), + index_details: action.index_details.clone(), + } + } +} + +impl TryFrom for AddIndex { + type Error = Error; + + fn try_from(proto: pb::AddIndex) -> Result { + Ok(Self { + uuid: proto.uuid, + name: proto.name, + fields: proto.fields, + covers_existing: proto.covers_existing, + covers_local: proto.covers_local, + index_details: proto.index_details, + }) + } +} + impl From<&Action> for pb::Action { fn from(action: &Action) -> Self { let action = match action { Action::AddFragments(add) => pb::action::Action::AddFragments(add.into()), + Action::AddIndex(add) => pb::action::Action::AddIndex(add.into()), }; Self { action: Some(action), @@ -127,6 +224,7 @@ impl TryFrom for Action { fn try_from(proto: pb::Action) -> Result { match proto.action { Some(pb::action::Action::AddFragments(add)) => Ok(Self::AddFragments(add.try_into()?)), + Some(pb::action::Action::AddIndex(add)) => Ok(Self::AddIndex(add.try_into()?)), None => Err(Error::invalid_input( "Action protobuf has no variant set".to_string(), )), @@ -197,10 +295,29 @@ mod tests { uuid: "test-uuid".to_string(), read_version: 7, actions: vec![UserAction { - description: "append batch".to_string(), - actions: vec![Action::AddFragments(AddFragments { - fragments: vec![Fragment::new(0), Fragment::new(1)], - })], + description: "append batch + index".to_string(), + actions: vec![ + Action::AddFragments(AddFragments { + new_fragments: vec![ + NewFragment { + local_id: 0, + fragment: Fragment::new(0), + }, + NewFragment { + local_id: 1, + fragment: Fragment::new(0), + }, + ], + }), + Action::AddIndex(AddIndex { + uuid: "idx-uuid".to_string(), + name: "my_idx".to_string(), + fields: vec![1], + covers_existing: vec![3, 4], + covers_local: vec![0], + index_details: Some(vec![1, 2, 3]), + }), + ], }], }; From 4141eb568e5b3ac1bc652658793a0bea339d76da Mon Sep 17 00:00:00 2001 From: Will Jones Date: Tue, 7 Jul 2026 11:02:56 -0700 Subject: [PATCH 2/6] feat(transaction): add opaque experimental-operation wire hook Add an ExperimentalUserOperation message + oneof arm (field 115) to the canonical transaction.proto. The action list is carried as an opaque `bytes` payload (a serialized experimental UserOperation) so the PMC-voted schema stays free of the unstable action structure. Builds that cannot interpret it reject the transaction; a real Operation variant is wired up in the next commit. Co-Authored-By: Claude Opus 4.8 (1M context) --- protos/transaction.proto | 14 ++++++++++++++ rust/lance/src/dataset/transaction.rs | 9 +++++++++ 2 files changed, 23 insertions(+) diff --git a/protos/transaction.proto b/protos/transaction.proto index e72e95025a4..caef9a39edf 100644 --- a/protos/transaction.proto +++ b/protos/transaction.proto @@ -329,6 +329,19 @@ message Transaction { repeated BasePath new_bases = 1; } + // EXPERIMENTAL: an action-based (UserOperation) transaction. + // + // The action list is carried as an opaque payload so this stable, PMC-voted + // schema stays free of the unstable action structure: `payload` is a + // serialized `lance.table.UserOperation` from the (non-default, + // feature-gated) transaction_experimental.proto. Builds without the + // `unstable-action-transactions` feature cannot interpret it and reject the + // transaction. Once the format is ratified, this is replaced by the real + // typed message. + message ExperimentalUserOperation { + bytes payload = 1; + } + // The operation of this transaction. oneof operation { Append append = 100; @@ -346,6 +359,7 @@ message Transaction { UpdateMemWalState update_mem_wal_state = 112; Clone clone = 113; UpdateBases update_bases = 114; + ExperimentalUserOperation experimental_user_operation = 115; } // Fields 200/202 (`blob_append` / `blob_overwrite`) previously represented blob dataset ops. diff --git a/rust/lance/src/dataset/transaction.rs b/rust/lance/src/dataset/transaction.rs index 3261f9300c4..ee3a0e485f9 100644 --- a/rust/lance/src/dataset/transaction.rs +++ b/rust/lance/src/dataset/transaction.rs @@ -3348,6 +3348,15 @@ impl TryFrom for Transaction { })) => Operation::UpdateBases { new_bases: new_bases.into_iter().map(BasePath::from).collect(), }, + Some(pb::transaction::Operation::ExperimentalUserOperation(_)) => { + // Wired to a real Operation variant in a later commit; until then + // (and in any build without the feature) it is unreadable. + return Err(Error::not_supported_source( + "action-based (experimental) transactions require a build with the \ + `unstable-action-transactions` feature" + .into(), + )); + } None => { return Err(Error::internal( "Transaction message did not contain an operation".to_string(), From e31cd57a68ba3b72e80a5b3df7364df62b518fba Mon Sep 17 00:00:00 2001 From: Will Jones Date: Tue, 7 Jul 2026 11:26:18 -0700 Subject: [PATCH 3/6] feat(transaction): apply + commit action-based UserOperation Wire the experimental action-based transaction through the existing commit path. Adds a feature-gated `Operation::UserOperation` variant and an `apply_user_operation` step in `build_manifest` that: - mints fragment ids from the manifest counter (late binding) and records each `NewFragment.local_id` -> assigned id, and - registers `AddIndex` metadata whose fragment bitmap is the union of `covers_existing` and the placeholders resolved above. Committing a UserOperation declares the `action-transactions` experimental writer feature (sets FLAG_EXPERIMENTAL). Conflict resolution treats the op like an Append (commutes with concurrent appends); retries re-run `build_manifest` against the latest manifest, so ids and placeholders are recomputed each attempt. The wire payload is encoded into the opaque `experimental_user_operation` transaction field. Co-Authored-By: Claude Opus 4.8 (1M context) --- rust/lance/src/dataset/transaction.rs | 186 +++++++++++++++++- rust/lance/src/io/commit/conflict_resolver.rs | 91 +++++++++ 2 files changed, 275 insertions(+), 2 deletions(-) diff --git a/rust/lance/src/dataset/transaction.rs b/rust/lance/src/dataset/transaction.rs index ee3a0e485f9..86326050b53 100644 --- a/rust/lance/src/dataset/transaction.rs +++ b/rust/lance/src/dataset/transaction.rs @@ -453,6 +453,12 @@ pub enum Operation { /// The new base paths to add to the manifest. new_bases: Vec, }, + + /// EXPERIMENTAL: an action-based transaction. The ordered action list is + /// applied to the manifest atomically, enabling compound commits (e.g. + /// append + add index). Gated behind `unstable-action-transactions`. + #[cfg(feature = "unstable-action-transactions")] + UserOperation(lance_table::transaction::UserOperation), } #[derive(Debug, Clone, PartialEq, DeepSizeOf)] @@ -502,6 +508,8 @@ impl std::fmt::Display for Operation { Self::Clone { .. } => write!(f, "Clone"), Self::UpdateMemWalState { .. } => write!(f, "UpdateMemWalState"), Self::UpdateBases { .. } => write!(f, "UpdateBases"), + #[cfg(feature = "unstable-action-transactions")] + Self::UserOperation(_) => write!(f, "UserOperation"), } } } @@ -1345,6 +1353,14 @@ impl PartialEq for Operation { (Self::Clone { .. }, Self::UpdateBases { .. }) => { std::mem::discriminant(self) == std::mem::discriminant(other) } + // Two action-based operations compare by content; any pairing with a + // non-action operation is unequal (different discriminants). Gated so + // the match stays exhaustive without a catch-all when the feature is + // off. + #[cfg(feature = "unstable-action-transactions")] + (Self::UserOperation(a), Self::UserOperation(b)) => a == b, + #[cfg(feature = "unstable-action-transactions")] + _ => std::mem::discriminant(self) == std::mem::discriminant(other), } } } @@ -1524,6 +1540,8 @@ impl Operation { Self::UpdateMemWalState { .. } => "UpdateMemWalState", Self::Clone { .. } => "Clone", Self::UpdateBases { .. } => "UpdateBases", + #[cfg(feature = "unstable-action-transactions")] + Self::UserOperation(_) => "UserOperation", } } } @@ -2317,6 +2335,19 @@ impl Transaction { // Base paths are handled in the manifest creation section below final_fragments.extend(maybe_existing_fragments?.clone()); } + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(user_op) => { + final_fragments.extend(maybe_existing_fragments?.clone()); + let new_version = current_manifest.map(|m| m.version + 1).unwrap_or(1); + Self::apply_user_operation( + user_op, + &mut fragment_id, + &mut next_row_id, + new_version, + &mut final_fragments, + &mut final_indices, + )?; + } }; // If a fragment was reserved then it may not belong at the end of the fragments list. @@ -2361,6 +2392,16 @@ impl Transaction { manifest.tag.clone_from(&self.tag); + // Declaring the experimental feature makes `apply_feature_flags` set + // FLAG_EXPERIMENTAL, so libraries without the feature refuse to commit. + #[cfg(feature = "unstable-action-transactions")] + if matches!(self.operation, Operation::UserOperation(_)) { + let feature = lance_table::transaction::FEATURE_NAME.to_string(); + if !manifest.experimental_writer_features.contains(&feature) { + manifest.experimental_writer_features.push(feature); + } + } + if config.auto_set_feature_flags { // Internal operations (e.g. CreateIndex) use ManifestWriteConfig::default() // which has use_stable_row_ids = false. Without inheriting from the previous @@ -2561,6 +2602,127 @@ impl Transaction { Ok((manifest, final_indices)) } + /// Apply an action-based [`UserOperation`] to the in-progress fragment and + /// index lists. Fragment ids are minted from `fragment_id` (late binding), + /// and `NewFragment` placeholders are resolved so a later `AddIndex` in the + /// same operation can cover fragments it created. Re-run per commit attempt, + /// so ids and the placeholder map are recomputed against the latest manifest. + #[cfg(feature = "unstable-action-transactions")] + fn apply_user_operation( + user_op: &lance_table::transaction::UserOperation, + fragment_id: &mut u64, + next_row_id: &mut Option, + new_version: u64, + final_fragments: &mut Vec, + final_indices: &mut Vec, + ) -> Result<()> { + use lance_table::transaction::Action; + use prost::Message as _; + + // Placeholders resolve to the fragment ids assigned during this apply. + let mut local_to_id: HashMap = HashMap::new(); + + let to_bitmap_id = |id: u64| -> Result { + u32::try_from(id).map_err(|_| { + Error::invalid_input(format!( + "fragment id {} exceeds the u32 range of an index fragment bitmap", + id + )) + }) + }; + + for user_action in &user_op.actions { + for action in &user_action.actions { + match action { + Action::AddFragments(add) => { + let mut new_fragments = Vec::with_capacity(add.new_fragments.len()); + for new_fragment in &add.new_fragments { + let mut fragment = new_fragment.fragment.clone(); + fragment.id = *fragment_id; + *fragment_id += 1; + if local_to_id + .insert(new_fragment.local_id, fragment.id) + .is_some() + { + return Err(Error::invalid_input(format!( + "duplicate NewFragment local_id {} within the operation", + new_fragment.local_id + ))); + } + new_fragments.push(fragment); + } + if let Some(next_row_id) = next_row_id.as_mut() { + Self::assign_row_ids(next_row_id, new_fragments.as_mut_slice())?; + for fragment in new_fragments.iter_mut() { + let version_meta = build_version_meta(fragment, new_version); + fragment.last_updated_at_version_meta = version_meta.clone(); + fragment.created_at_version_meta = version_meta; + } + } + final_fragments.extend(new_fragments); + } + Action::AddIndex(add) => { + let mut fragment_bitmap = RoaringBitmap::new(); + for existing in &add.covers_existing { + fragment_bitmap.insert(to_bitmap_id(*existing)?); + } + for local in &add.covers_local { + let id = local_to_id.get(local).ok_or_else(|| { + Error::invalid_input(format!( + "AddIndex covers_local placeholder {} was not produced by an \ + earlier AddFragments in this operation", + local + )) + })?; + fragment_bitmap.insert(to_bitmap_id(*id)?); + } + let uuid = Uuid::parse_str(&add.uuid).map_err(|e| { + Error::invalid_input(format!( + "AddIndex has an invalid uuid {:?}: {}", + add.uuid, e + )) + })?; + let index_details = add + .index_details + .as_ref() + .map(|bytes| prost_types::Any::decode(bytes.as_slice())) + .transpose() + .map_err(|e| { + Error::invalid_input(format!( + "AddIndex index_details is not a valid protobuf Any: {}", + e + )) + })? + .map(Arc::new); + let index = IndexMetadata { + uuid, + name: add.name.clone(), + fields: add.fields.clone(), + dataset_version: new_version, + fragment_bitmap: Some(fragment_bitmap), + index_details, + index_version: 0, + created_at: None, + base_id: None, + files: None, + }; + // Replace-by-uuid, mirroring CreateIndex. + final_indices.retain(|existing| existing.uuid != index.uuid); + final_indices.push(index); + } + // `Action` is #[non_exhaustive]; a variant this build predates + // cannot be applied. + _ => { + return Err(Error::not_supported_source( + "unsupported experimental transaction action".into(), + )); + } + } + } + } + Ok(()) + } + fn register_pure_rewrite_rows_update_frags_in_indices( indices: &mut [IndexMetadata], pure_update_frag_ids: &[u64], @@ -3348,9 +3510,21 @@ impl TryFrom for Transaction { })) => Operation::UpdateBases { new_bases: new_bases.into_iter().map(BasePath::from).collect(), }, + #[cfg(feature = "unstable-action-transactions")] + Some(pb::transaction::Operation::ExperimentalUserOperation(inner)) => { + use prost::Message as _; + let user_op = pb::UserOperation::decode(inner.payload.as_slice()).map_err(|e| { + Error::invalid_input(format!( + "invalid experimental UserOperation payload: {}", + e + )) + })?; + Operation::UserOperation(user_op.try_into()?) + } + #[cfg(not(feature = "unstable-action-transactions"))] Some(pb::transaction::Operation::ExperimentalUserOperation(_)) => { - // Wired to a real Operation variant in a later commit; until then - // (and in any build without the feature) it is unreadable. + // Unreadable without the feature: the payload is an experimental + // format this build does not understand. return Err(Error::not_supported_source( "action-based (experimental) transactions require a build with the \ `unstable-action-transactions` feature" @@ -3644,6 +3818,14 @@ impl From<&Transaction> for pb::Transaction { .collect::>(), }) } + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(user_op) => { + use prost::Message as _; + let payload = pb::UserOperation::from(user_op).encode_to_vec(); + pb::transaction::Operation::ExperimentalUserOperation( + pb::transaction::ExperimentalUserOperation { payload }, + ) + } }; let transaction_properties = value diff --git a/rust/lance/src/io/commit/conflict_resolver.rs b/rust/lance/src/io/commit/conflict_resolver.rs index d95821dd130..12ace5d54ac 100644 --- a/rust/lance/src/io/commit/conflict_resolver.rs +++ b/rust/lance/src/io/commit/conflict_resolver.rs @@ -62,6 +62,16 @@ impl<'a> TransactionRebase<'a> { conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_merged_gens: Vec::new(), }), + // Action-based operations currently only add fragments/indices. + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => Ok(Self { + transaction, + affected_rows, + initial_fragments: HashMap::new(), + modified_fragment_ids: HashSet::new(), + conflicting_frag_reuse_indices: Vec::new(), + conflicting_mem_wal_merged_gens: Vec::new(), + }), Operation::Delete { updated_fragments, deleted_fragment_ids, @@ -235,6 +245,30 @@ impl<'a> TransactionRebase<'a> { Operation::UpdateBases { .. } => { self.check_add_bases_txn(other_transaction, other_version) } + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => { + self.check_user_operation_txn(other_transaction, other_version) + } + } + } + + /// Conflict check for an action-based [`Operation::UserOperation`]. It + /// currently only adds fragments and indices, so it commutes with everything + /// except operations that replace the schema/table wholesale — mirroring + /// [`Self::check_append_txn`]. + #[cfg(feature = "unstable-action-transactions")] + fn check_user_operation_txn( + &mut self, + other_transaction: &Transaction, + other_version: u64, + ) -> Result<()> { + match &other_transaction.operation { + Operation::Overwrite { .. } + | Operation::Restore { .. } + | Operation::UpdateMemWalState { .. } => { + Err(self.incompatible_conflict_err(other_transaction, other_version)) + } + _ => Ok(()), } } @@ -252,6 +286,9 @@ impl<'a> TransactionRebase<'a> { | Operation::Append { .. } | Operation::UpdateConfig { .. } | Operation::UpdateBases { .. } => Ok(()), + // Action-based op only adds new fragments/indices (like Append). + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => Ok(()), Operation::Rewrite { groups, .. } => { if groups .iter() @@ -408,6 +445,14 @@ impl<'a> TransactionRebase<'a> { } Ok(()) } + // Action-based op adds new fragments like Append; same reasoning. + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => { + if self_inserted_rows_filter.is_some() { + return Err(self.retryable_conflict_err(other_transaction, other_version)); + } + Ok(()) + } Operation::Rewrite { groups, .. } => { if groups .iter() @@ -515,6 +560,10 @@ impl<'a> TransactionRebase<'a> { Operation::Append { .. } | Operation::Clone { .. } | Operation::UpdateBases { .. } => Ok(()), + // Action-based op adds new fragments/indices; distinct-uuid indices + // don't collide with this CreateIndex (like a concurrent Append). + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => Ok(()), Operation::CreateIndex { new_indices: created_indices, .. @@ -674,6 +723,10 @@ impl<'a> TransactionRebase<'a> { | Operation::UpdateConfig { .. } | Operation::UpdateMemWalState { .. } | Operation::UpdateBases { .. } => Ok(()), + // Action-based op only adds new fragments; a rewrite of existing + // fragments doesn't touch them (like a concurrent Append). + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => Ok(()), Operation::Delete { updated_fragments, deleted_fragment_ids, @@ -880,6 +933,10 @@ impl<'a> TransactionRebase<'a> { | Operation::Update { .. } | Operation::Project { .. } | Operation::UpdateBases { .. } => Ok(()), + // An overwrite replaces everything, so a concurrent action-based op is + // irrelevant here (like a concurrent Append). + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => Ok(()), } } @@ -908,6 +965,8 @@ impl<'a> TransactionRebase<'a> { | Operation::UpdateConfig { .. } | Operation::Clone { .. } | Operation::DataReplacement { .. } => Ok(()), + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => Ok(()), } } @@ -924,6 +983,9 @@ impl<'a> TransactionRebase<'a> { | Operation::ReserveFragments { .. } | Operation::Project { .. } | Operation::UpdateBases { .. } => Ok(()), + // Action-based op only adds new fragments (like a concurrent Append). + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => Ok(()), Operation::Merge { .. } => { // Merge rewrites the whole fragment list; always conflict // (symmetric with check_merge_txn). @@ -1075,6 +1137,12 @@ impl<'a> TransactionRebase<'a> { | Operation::DataReplacement { .. } => { Err(self.retryable_conflict_err(other_transaction, other_version)) } + // Merge rewrites the whole fragment list, so it conflicts with an + // action-based op that added fragments (same as a concurrent Append). + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => { + Err(self.retryable_conflict_err(other_transaction, other_version)) + } Operation::Overwrite { .. } | Operation::Restore { .. } | Operation::Project { .. } @@ -1104,6 +1172,8 @@ impl<'a> TransactionRebase<'a> { | Operation::Project { .. } | Operation::Clone { .. } | Operation::UpdateConfig { .. } => Ok(()), + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => Ok(()), Operation::UpdateMemWalState { .. } => { Err(self.incompatible_conflict_err(other_transaction, other_version)) } @@ -1132,6 +1202,8 @@ impl<'a> TransactionRebase<'a> { | Operation::UpdateConfig { .. } | Operation::UpdateMemWalState { .. } | Operation::UpdateBases { .. } => Ok(()), + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => Ok(()), } } @@ -1152,6 +1224,10 @@ impl<'a> TransactionRebase<'a> { | Operation::Clone { .. } | Operation::ReserveFragments { .. } | Operation::UpdateBases { .. } => Ok(()), + // Action-based op only adds fragments/indices; it doesn't change the + // schema (like a concurrent Append). + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => Ok(()), Operation::Merge { .. } | Operation::Project { .. } => { // Need to recompute the schema Err(self.retryable_conflict_err(other_transaction, other_version)) @@ -1219,6 +1295,9 @@ impl<'a> TransactionRebase<'a> { | Operation::Project { .. } | Operation::UpdateMemWalState { .. } | Operation::UpdateBases { .. } => Ok(()), + // Config updates don't conflict with a fragment/index add. + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => Ok(()), } } else { Err(wrong_operation_err(&self.transaction.operation)) @@ -1290,6 +1369,12 @@ impl<'a> TransactionRebase<'a> { | Operation::Project { .. } => { Err(self.incompatible_conflict_err(other_transaction, other_version)) } + // Mem-WAL state changes are incompatible with a fragment add + // (same as a concurrent Append). + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => { + Err(self.incompatible_conflict_err(other_transaction, other_version)) + } } } else { Err(wrong_operation_err(&self.transaction.operation)) @@ -1385,6 +1470,10 @@ impl<'a> TransactionRebase<'a> { | Operation::UpdateConfig { .. } | Operation::UpdateMemWalState { .. } | Operation::UpdateBases { .. } => Ok(self.transaction), + // No action-level rebase: retries re-run build_manifest against the + // latest manifest, which re-assigns ids and re-resolves placeholders. + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => Ok(self.transaction), } } @@ -3226,6 +3315,8 @@ mod tests { | Operation::UpdateBases { .. } | Operation::Restore { .. } | Operation::UpdateMemWalState { .. } => Box::new(std::iter::empty()), + #[cfg(feature = "unstable-action-transactions")] + Operation::UserOperation(_) => Box::new(std::iter::empty()), Operation::Delete { updated_fragments, deleted_fragment_ids, From b511351ec80e332ebbbcca9398d47336ae219905 Mon Sep 17 00:00:00 2001 From: Will Jones Date: Tue, 7 Jul 2026 11:34:18 -0700 Subject: [PATCH 4/6] test(transaction): cover compound append + add-index UserOperation Feature-gated integration tests: (1) append two fragments and register an index covering only the first in one action-based transaction, asserting fragment ids, the index fragment bitmap (placeholder-resolved), FLAG_EXPERIMENTAL, and row count; (2) a stale UserOperation rebases past a concurrent plain Append and still commits, confirming action-based appends don't conflict with appends. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/dataset/tests/dataset_transactions.rs | 190 ++++++++++++++++++ 1 file changed, 190 insertions(+) diff --git a/rust/lance/src/dataset/tests/dataset_transactions.rs b/rust/lance/src/dataset/tests/dataset_transactions.rs index 3e2a4caa3b3..79f1255edea 100644 --- a/rust/lance/src/dataset/tests/dataset_transactions.rs +++ b/rust/lance/src/dataset/tests/dataset_transactions.rs @@ -502,3 +502,193 @@ async fn test_list_detached_manifests() { assert_eq!(versions.len(), 1); assert_eq!(versions[0].version, 1); } + +#[cfg(feature = "unstable-action-transactions")] +mod action_transactions { + use super::*; + use crate::dataset::transaction::Operation; + use lance_table::feature_flags::FLAG_EXPERIMENTAL; + use lance_table::format::Fragment; + use lance_table::transaction::{ + Action, AddFragments, AddIndex, NewFragment, UserAction, UserOperation, + }; + + fn schema() -> Arc { + Arc::new(ArrowSchema::new(vec![ArrowField::new( + "id", + DataType::Int32, + false, + )])) + } + + fn batch(values: Vec) -> RecordBatch { + RecordBatch::try_new(schema(), vec![Arc::new(Int32Array::from(values))]).unwrap() + } + + /// Write a single uncommitted fragment against `dataset` and return it. + async fn write_one_fragment(dataset: Arc, values: Vec) -> Fragment { + let reader = RecordBatchIterator::new([Ok(batch(values))], schema()); + let params = WriteParams { + mode: WriteMode::Append, + ..Default::default() + }; + let txn = InsertBuilder::new(dataset) + .with_params(¶ms) + .execute_uncommitted_stream(reader) + .await + .unwrap(); + match txn.operation { + Operation::Append { mut fragments } => { + assert_eq!(fragments.len(), 1, "expected exactly one fragment"); + fragments.pop().unwrap() + } + other => panic!("expected Append, got {other}"), + } + } + + // A pylance-equivalent flow: append two new fragments and register an index + // covering only the first, in a single action-based transaction. + #[tokio::test] + async fn append_and_add_index_in_one_transaction() { + let test_uri = TempStrDir::default(); + let dataset = Arc::new( + Dataset::write( + RecordBatchIterator::new([Ok(batch(vec![1, 2, 3]))], schema()), + &test_uri, + None, + ) + .await + .unwrap(), + ); + let read_version = dataset.manifest().version; + // Initial dataset has a single fragment with id 0. + assert_eq!(dataset.get_fragments().len(), 1); + + let frag_a = write_one_fragment(dataset.clone(), vec![4, 5, 6]).await; + let frag_b = write_one_fragment(dataset.clone(), vec![7, 8]).await; + + let index_uuid = "11111111-1111-1111-1111-111111111111"; + let user_op = UserOperation { + description: "append + index".to_string(), + uuid: "test-op".to_string(), + read_version, + actions: vec![UserAction { + description: "append two fragments, index the first".to_string(), + actions: vec![ + Action::AddFragments(AddFragments { + new_fragments: vec![ + NewFragment { + local_id: 0, + fragment: frag_a, + }, + NewFragment { + local_id: 1, + fragment: frag_b, + }, + ], + }), + Action::AddIndex(AddIndex { + uuid: index_uuid.to_string(), + name: "id_idx".to_string(), + fields: vec![0], + covers_existing: vec![], + covers_local: vec![0], + index_details: None, + }), + ], + }], + }; + + let txn = Transaction::new(read_version, Operation::UserOperation(user_op), None); + let committed = CommitBuilder::new(dataset).execute(txn).await.unwrap(); + + // Both fragments landed alongside the original, and all rows are readable. + assert_eq!(committed.manifest().version, read_version + 1); + assert_eq!(committed.get_fragments().len(), 3); + assert_eq!(committed.count_rows(None).await.unwrap(), 8); + + // The dataset now declares the experimental writer feature. + assert_ne!( + committed.manifest().writer_feature_flags & FLAG_EXPERIMENTAL, + 0 + ); + assert!( + committed + .manifest() + .experimental_writer_features + .iter() + .any(|f| f == "action-transactions") + ); + + // The index was registered and covers exactly the first new fragment + // (assigned id 1), not the second (id 2) or the original (id 0). + let indices = committed.load_indices().await.unwrap(); + assert_eq!(indices.len(), 1); + let index = &indices[0]; + assert_eq!(index.name, "id_idx"); + assert_eq!(index.uuid.to_string(), index_uuid); + let bitmap = index.fragment_bitmap.as_ref().unwrap(); + assert_eq!(bitmap.len(), 1); + assert!(bitmap.contains(1)); + assert!(!bitmap.contains(0)); + assert!(!bitmap.contains(2)); + } + + // An action-based append must not conflict with a concurrent plain Append: + // the stale UserOperation rebases against the newer version and still lands. + #[tokio::test] + async fn does_not_conflict_with_concurrent_append() { + let test_uri = TempStrDir::default(); + let dataset = Arc::new( + Dataset::write( + RecordBatchIterator::new([Ok(batch(vec![1, 2, 3]))], schema()), + &test_uri, + None, + ) + .await + .unwrap(), + ); + let read_version = dataset.manifest().version; + + // Prepare the action-based append against the original version. + let frag = write_one_fragment(dataset.clone(), vec![4, 5]).await; + let user_op = UserOperation { + description: "action append".to_string(), + uuid: "action-op".to_string(), + read_version, + actions: vec![UserAction { + description: "append one fragment".to_string(), + actions: vec![Action::AddFragments(AddFragments { + new_fragments: vec![NewFragment { + local_id: 0, + fragment: frag, + }], + })], + }], + }; + let action_txn = Transaction::new(read_version, Operation::UserOperation(user_op), None); + + // A concurrent plain append commits first, advancing the version. + let after_append = CommitBuilder::new(dataset.clone()) + .execute(Transaction::new( + read_version, + Operation::Append { + fragments: vec![write_one_fragment(dataset.clone(), vec![6, 7]).await], + }, + None, + )) + .await + .unwrap(); + assert_eq!(after_append.manifest().version, read_version + 1); + + // Now the stale action-based transaction still commits (rebased). + let committed = CommitBuilder::new(dataset) + .execute(action_txn) + .await + .unwrap(); + assert_eq!(committed.manifest().version, read_version + 2); + // Original fragment + concurrent append + action append. + assert_eq!(committed.get_fragments().len(), 3); + assert_eq!(committed.count_rows(None).await.unwrap(), 7); + } +} From f54b8431f078300fee18bf9e6a56f0ceae1c2b31 Mon Sep 17 00:00:00 2001 From: Will Jones Date: Tue, 7 Jul 2026 11:48:11 -0700 Subject: [PATCH 5/6] feat(python): expose action-based UserOperation commit surface Add gated LanceOperation.UserOperation / UserAction / AddFragments / AddIndex / NewFragment dataclasses, committed through the existing LanceDataset.commit path. The PyO3 layer parses them into the Rust Operation::UserOperation under the non-default `unstable-action-transactions` feature; without it, committing one raises a clear error. A `lance.lance._action_transactions_enabled` capability flag lets callers/tests detect whether the extension was built with the feature. Co-Authored-By: Claude Opus 4.8 (1M context) --- python/Cargo.toml | 3 + python/python/lance/dataset.py | 118 +++++++++++++++++++++++++++++++++ python/src/lib.rs | 6 ++ python/src/transaction.rs | 91 +++++++++++++++++++++++++ 4 files changed, 218 insertions(+) diff --git a/python/Cargo.toml b/python/Cargo.toml index f0c84c01a16..862930d0fe6 100644 --- a/python/Cargo.toml +++ b/python/Cargo.toml @@ -86,6 +86,9 @@ bytes = "1.4" default = [] datagen = ["lance-datagen"] fp16kernels = ["lance/fp16kernels"] +# EXPERIMENTAL: action-based (UserOperation) transactions. Non-default; the wire +# format is unstable. Build with `maturin develop --features ...` to enable. +unstable-action-transactions = ["lance/unstable-action-transactions"] [profile.ci] debug = "line-tables-only" diff --git a/python/python/lance/dataset.py b/python/python/lance/dataset.py index 6be0a78d2e8..89c10f7178c 100644 --- a/python/python/lance/dataset.py +++ b/python/python/lance/dataset.py @@ -5707,6 +5707,124 @@ class Append(BaseOperation): def __post_init__(self): LanceOperation._validate_fragments(self.fragments) + @dataclass + class NewFragment: + """ + EXPERIMENTAL. A fragment to append via an action-based transaction. + + Attributes + ---------- + local_id: int + Operation-local placeholder identifying this fragment, unique among + the operation's new fragments. A later :class:`AddIndex` references it + via ``covers_local``; it is resolved to the real fragment id assigned + at commit time. + fragment: FragmentMetadata + The fragment to append. Its ``id`` is ignored and assigned at commit. + """ + + local_id: int + fragment: FragmentMetadata + + @dataclass + class AddFragments: + """ + EXPERIMENTAL. Append new fragments within a + :class:`LanceOperation.UserOperation`. + + Attributes + ---------- + new_fragments: list[LanceOperation.NewFragment] + The fragments to append, each with an operation-local placeholder. + """ + + new_fragments: List["LanceOperation.NewFragment"] + + @dataclass + class AddIndex: + """ + EXPERIMENTAL. Register an index segment within a + :class:`LanceOperation.UserOperation`. + + The index file(s) must already be written; this only records the index + metadata. Coverage is the union of already-committed fragment ids + (``covers_existing``) and same-operation placeholders (``covers_local``). + + Attributes + ---------- + uuid: str + Unique identifier of the index across all dataset versions. + name: str + Index name. Replace-by-uuid semantics apply on the manifest. + fields: list[int] + The field ids the index is built on. + covers_existing: list[int] + Already-committed fragment ids this index covers. + covers_local: list[int] + Placeholders (:attr:`NewFragment.local_id`) of same-operation + fragments this index covers. + index_details: bytes, optional + Opaque, type-specific index metadata (a serialized + ``google.protobuf.Any``). ``None`` when the index carries no details. + """ + + uuid: str + name: str + fields: List[int] + covers_existing: List[int] + covers_local: List[int] + index_details: Optional[bytes] = None + + @dataclass + class UserAction: + """ + EXPERIMENTAL. A named step within a + :class:`LanceOperation.UserOperation`, expanding to granular actions. + + Attributes + ---------- + description: str + Human-readable description of this step. + actions: list + The granular actions (:class:`AddFragments` / :class:`AddIndex`) this + step expands to. + """ + + description: str + actions: List[Union["LanceOperation.AddFragments", "LanceOperation.AddIndex"]] + + @dataclass + class UserOperation(BaseOperation): + """ + EXPERIMENTAL. An action-based transaction. + + Models a transaction as an ordered list of actions applied atomically, + enabling compound commits such as *append data + register an index over + it* in one commit. Commit it through :meth:`LanceDataset.commit` like any + other operation. + + Requires the ``pylance`` extension to be built with the + ``unstable-action-transactions`` feature; otherwise committing raises a + ``ValueError``. The wire format is unstable and carries no compatibility + guarantee. + + Attributes + ---------- + read_version: int + The dataset version this operation was planned against. + uuid: str + Unique identifier for this operation. + description: str + Human-readable description, e.g. ``"INSERT INTO t VALUES (1)"``. + actions: list[LanceOperation.UserAction] + The ordered user actions to apply. + """ + + read_version: int + uuid: str + description: str + actions: List["LanceOperation.UserAction"] + @dataclass class Delete(BaseOperation): """ diff --git a/python/src/lib.rs b/python/src/lib.rs index 466d4ea90f2..9c92684e9e1 100644 --- a/python/src/lib.rs +++ b/python/src/lib.rs @@ -329,6 +329,12 @@ fn lance(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_wrapped(wrap_pyfunction!(debug::format_fragment))?; m.add_wrapped(wrap_pyfunction!(debug::list_transactions))?; m.add("__version__", env!("CARGO_PKG_VERSION"))?; + // Whether this extension was built with the experimental action-based + // transaction feature. Committing a LanceOperation.UserOperation requires it. + m.add( + "_action_transactions_enabled", + cfg!(feature = "unstable-action-transactions"), + )?; register_datagen(py, m)?; register_indices(py, m)?; diff --git a/python/src/transaction.rs b/python/src/transaction.rs index 1b659395099..4e91fb27658 100644 --- a/python/src/transaction.rs +++ b/python/src/transaction.rs @@ -224,10 +224,101 @@ impl FromPyObject<'_, '_> for PyUpdateMode { } } +// EXPERIMENTAL: parsing for action-based (UserOperation) transactions. Gated by +// the non-default `unstable-action-transactions` feature. +#[cfg(feature = "unstable-action-transactions")] +mod action_transactions { + use super::*; + use lance_table::transaction::{ + Action, AddFragments, AddIndex, NewFragment, UserAction, UserOperation, + }; + + impl FromPyObject<'_, '_> for PyLance { + type Error = PyErr; + fn extract(ob: Borrowed<'_, '_, PyAny>) -> PyResult { + let local_id = ob.getattr("local_id")?.extract()?; + let PyLance(fragment) = ob.getattr("fragment")?.extract()?; + Ok(Self(NewFragment { local_id, fragment })) + } + } + + impl FromPyObject<'_, '_> for PyLance { + type Error = PyErr; + fn extract(ob: Borrowed<'_, '_, PyAny>) -> PyResult { + let new_fragments = extract_vec(&ob.getattr("new_fragments")?)?; + Ok(Self(AddFragments { new_fragments })) + } + } + + impl FromPyObject<'_, '_> for PyLance { + type Error = PyErr; + fn extract(ob: Borrowed<'_, '_, PyAny>) -> PyResult { + Ok(Self(AddIndex { + uuid: ob.getattr("uuid")?.extract()?, + name: ob.getattr("name")?.extract()?, + fields: ob.getattr("fields")?.extract()?, + covers_existing: ob.getattr("covers_existing")?.extract()?, + covers_local: ob.getattr("covers_local")?.extract()?, + index_details: ob.getattr("index_details")?.extract()?, + })) + } + } + + impl FromPyObject<'_, '_> for PyLance { + type Error = PyErr; + fn extract(ob: Borrowed<'_, '_, PyAny>) -> PyResult { + let action = match class_name(&ob)?.as_str() { + "AddFragments" => Action::AddFragments(PyLance::::extract(ob)?.0), + "AddIndex" => Action::AddIndex(PyLance::::extract(ob)?.0), + other => { + return Err(PyValueError::new_err(format!( + "Unsupported action: {other}" + ))); + } + }; + Ok(Self(action)) + } + } + + impl FromPyObject<'_, '_> for PyLance { + type Error = PyErr; + fn extract(ob: Borrowed<'_, '_, PyAny>) -> PyResult { + let description = ob.getattr("description")?.extract()?; + let actions = extract_vec(&ob.getattr("actions")?)?; + Ok(Self(UserAction { + description, + actions, + })) + } + } + + impl FromPyObject<'_, '_> for PyLance { + type Error = PyErr; + fn extract(ob: Borrowed<'_, '_, PyAny>) -> PyResult { + let description = ob.getattr("description")?.extract()?; + let uuid = ob.getattr("uuid")?.extract()?; + let read_version = ob.getattr("read_version")?.extract()?; + let actions = extract_vec(&ob.getattr("actions")?)?; + Ok(Self(UserOperation { + description, + uuid, + read_version, + actions, + })) + } + } +} + impl FromPyObject<'_, '_> for PyLance { type Error = PyErr; fn extract(ob: Borrowed<'_, '_, PyAny>) -> PyResult { match class_name(&ob)?.as_str() { + #[cfg(feature = "unstable-action-transactions")] + "UserOperation" => { + let PyLance(user_op) = + PyLance::::extract(ob)?; + Ok(Self(Operation::UserOperation(user_op))) + } "Overwrite" => { let schema = extract_schema(&ob.getattr("new_schema")?)?; From 42fb3b6913f428b74ecf30ae9fbab8a7d8ec085f Mon Sep 17 00:00:00 2001 From: Will Jones Date: Tue, 7 Jul 2026 11:48:15 -0700 Subject: [PATCH 6/6] test(python): cover compound append + add-index UserOperation Feature-gated (skips unless the extension was built with `unstable-action-transactions`): a pylance user constructs and commits a UserOperation that appends two fragments and registers an index over only the first, in one transaction, then asserts the row count, fragment count, and the index's resolved fragment coverage. Co-Authored-By: Claude Opus 4.8 (1M context) --- python/python/tests/test_dataset.py | 67 +++++++++++++++++++++++++++++ 1 file changed, 67 insertions(+) diff --git a/python/python/tests/test_dataset.py b/python/python/tests/test_dataset.py index e449c6b9865..1d3c6dbb3f1 100644 --- a/python/python/tests/test_dataset.py +++ b/python/python/tests/test_dataset.py @@ -1812,6 +1812,73 @@ def test_commit_timeout(tmp_path: Path): # throttling would be flaky on fast runners. +def test_user_operation_append_and_add_index(tmp_path: Path): + # EXPERIMENTAL: action-based transactions are gated behind a non-default + # Cargo feature, so this only runs when pylance was built with it. + if not getattr(lance.lance, "_action_transactions_enabled", False): + pytest.skip("pylance built without the unstable-action-transactions feature") + + base_dir = tmp_path / "action_txn" + dataset = lance.write_dataset(pa.Table.from_pydict({"id": [1, 2, 3]}), base_dir) + assert len(dataset.get_fragments()) == 1 + + # Two fragments written but not yet committed. + frag_a = lance.fragment.LanceFragment.create( + base_dir, pa.Table.from_pydict({"id": [4, 5, 6]}) + ) + frag_b = lance.fragment.LanceFragment.create( + base_dir, pa.Table.from_pydict({"id": [7, 8]}) + ) + + index_uuid = "11111111-1111-1111-1111-111111111111" + op = lance.LanceOperation.UserOperation( + read_version=dataset.version, + uuid="test-op", + description="append + index", + actions=[ + lance.LanceOperation.UserAction( + description="append two fragments, index the first", + actions=[ + lance.LanceOperation.AddFragments( + new_fragments=[ + lance.LanceOperation.NewFragment( + local_id=0, fragment=frag_a + ), + lance.LanceOperation.NewFragment( + local_id=1, fragment=frag_b + ), + ] + ), + lance.LanceOperation.AddIndex( + uuid=index_uuid, + name="id_idx", + fields=[0], + covers_existing=[], + covers_local=[0], + ), + ], + ) + ], + ) + + committed = lance.LanceDataset.commit(dataset, op, read_version=dataset.version) + + # Both fragments landed alongside the original and all rows are readable. + assert committed.version == dataset.version + 1 + assert committed.count_rows() == 8 + assert len(committed.get_fragments()) == 3 + + # The index was registered and covers exactly the first new fragment + # (assigned id 1), not the second (id 2) or the original (id 0). + indices = committed.describe_indices() + assert len(indices) == 1 + assert indices[0].name == "id_idx" + segments = indices[0].segments + assert len(segments) == 1 + assert segments[0].uuid == index_uuid + assert sorted(segments[0].fragment_ids) == [1] + + def test_append_with_commit(tmp_path: Path): table = pa.Table.from_pydict({"a": range(100), "b": range(100)}) base_dir = tmp_path / "test"