feat: allow setting transaction properties in various operations (#6246)

Allow callers to be able to set transaction properties in various
operations including index optimization, creation and data compaction.

---------

Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Happy <yesreply@happy.engineering>

Author: 3 people

rust/lance-index/src/optimize.rs

@@ -1,6 +1,9 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The Lance Authors
 
+use std::collections::HashMap;
+use std::sync::Arc;
+
 /// Options for optimizing all indices.
 #[non_exhaustive]
 #[derive(Debug, Clone, Default)]
@@ -33,6 +36,13 @@ pub struct OptimizeOptions {
     ///
     /// NOTE: this option is only supported for v3 vector indices.
     pub retrain: bool,
+
+    /// Transaction properties to store with this commit.
+    ///
+    /// These key-value pairs are stored in the transaction file
+    /// and can be read later to identify the source of the commit
+    /// (e.g., job_id for tracking completed index jobs).
+    pub transaction_properties: Option<Arc<HashMap<String, String>>>,
 }
 
 impl OptimizeOptions {
@@ -61,6 +71,7 @@ impl OptimizeOptions {
             num_indices_to_merge: None,
             index_names: None,
             retrain: true,
+            ..Default::default()
         }
     }
 
@@ -73,4 +84,10 @@ impl OptimizeOptions {
         self.index_names = Some(names);
         self
     }
+
+    /// Set transaction properties to store in the commit manifest.
+    pub fn transaction_properties(mut self, properties: HashMap<String, String>) -> Self {
+        self.transaction_properties = Some(Arc::new(properties));
+        self
+    }
 }

rust/lance/src/dataset/optimize.rs

@@ -89,7 +89,9 @@ use std::sync::Arc;
 use super::fragment::FileFragment;
 use super::index::DatasetIndexRemapperOptions;
 use super::rowids::load_row_id_sequences;
-use super::transaction::{Operation, RewriteGroup, RewrittenIndex, Transaction};
+use super::transaction::{
+    Operation, RewriteGroup, RewrittenIndex, Transaction, TransactionBuilder,
+};
 use super::utils::make_rowid_capture_stream;
 use super::{WriteMode, WriteParams, write_fragments_internal};
 use crate::Dataset;
@@ -207,6 +209,13 @@ pub struct CompactionOptions {
     /// fragments at a time).
     /// Defaults to `None` (no limit, all eligible fragments are compacted).
     pub max_source_fragments: Option<usize>,
+    /// Transaction properties to store with this commit.
+    ///
+    /// These key-value pairs are stored in the transaction file
+    /// and can be read later to identify the source of the commit
+    /// (e.g., job_id for tracking completed compaction jobs).
+    #[serde(skip)]
+    pub transaction_properties: Option<Arc<HashMap<String, String>>>,
 }
 
 #[allow(deprecated)]
@@ -227,6 +236,7 @@ impl Default for CompactionOptions {
             enable_binary_copy_force: false,
             binary_copy_read_batch_bytes: Some(16 * 1024 * 1024),
             max_source_fragments: None,
+            transaction_properties: None,
         }
     }
 }
@@ -378,6 +388,12 @@ impl CompactionOptions {
             _ => CompactionMode::Reencode,
         }
     }
+
+    /// Set transaction properties to store in the commit manifest.
+    pub fn transaction_properties(mut self, properties: HashMap<String, String>) -> Self {
+        self.transaction_properties = Some(Arc::new(properties));
+        self
+    }
 }
 
 /// Determine if page-level binary copy can safely merge the provided fragments.
@@ -1569,15 +1585,16 @@ pub async fn commit_compaction(
         None
     };
 
-    let transaction = Transaction::new(
+    let transaction = TransactionBuilder::new(
         dataset.manifest.version,
         Operation::Rewrite {
             groups: rewrite_groups,
             rewritten_indices,
             frag_reuse_index,
         },
-        None,
-    );
+    )
+    .transaction_properties(options.transaction_properties.clone())
+    .build();
 
     dataset
         .apply_commit(transaction, &Default::default(), &Default::default())

rust/lance/src/index.rs

@@ -78,7 +78,7 @@ use self::vector::remap_vector_index;
 use crate::dataset::index::LanceIndexStoreExt;
 use crate::dataset::optimize::RemappedIndex;
 use crate::dataset::optimize::remapping::RemapResult;
-use crate::dataset::transaction::{Operation, Transaction};
+use crate::dataset::transaction::{Operation, Transaction, TransactionBuilder};
 use crate::index::frag_reuse::{load_frag_reuse_index_details, open_frag_reuse_index};
 use crate::index::mem_wal::open_mem_wal_index;
 pub use crate::index::prefilter::{FilterLoader, PreFilter};
@@ -950,14 +950,15 @@ impl DatasetIndexExt for Dataset {
             return Ok(());
         }
 
-        let transaction = Transaction::new(
+        let transaction = TransactionBuilder::new(
             self.manifest.version,
             Operation::CreateIndex {
                 new_indices,
                 removed_indices,
             },
-            None,
-        );
+        )
+        .transaction_properties(options.transaction_properties.clone())
+        .build();
 
         self.apply_commit(transaction, &Default::default(), &Default::default())
             .await?;

rust/lance/src/index/create.rs

@@ -5,7 +5,7 @@ use crate::{
     Error, Result,
     dataset::{
         Dataset,
-        transaction::{Operation, Transaction},
+        transaction::{Operation, TransactionBuilder},
     },
     index::{
         DatasetIndexExt, DatasetIndexInternalExt,
@@ -26,7 +26,7 @@ use lance_index::{
     scalar::{LANCE_SCALAR_INDEX, ScalarIndexParams, inverted::tokenizer::InvertedIndexParams},
 };
 use lance_table::format::{IndexMetadata, list_index_files_with_sizes};
-use std::{future::IntoFuture, sync::Arc};
+use std::{collections::HashMap, future::IntoFuture, sync::Arc};
 use tracing::instrument;
 use uuid::Uuid;
 
@@ -56,6 +56,8 @@ pub struct CreateIndexBuilder<'a> {
     index_uuid: Option<String>,
     preprocessed_data: Option<Box<dyn RecordBatchReader + Send + 'static>>,
     progress: Arc<dyn IndexBuildProgress>,
+    /// Transaction properties to store with this commit.
+    transaction_properties: Option<Arc<HashMap<String, String>>>,
 }
 
 impl<'a> CreateIndexBuilder<'a> {
@@ -77,6 +79,7 @@ impl<'a> CreateIndexBuilder<'a> {
             index_uuid: None,
             preprocessed_data: None,
             progress: Arc::new(NoopIndexBuildProgress),
+            transaction_properties: None,
         }
     }
 
@@ -118,6 +121,16 @@ impl<'a> CreateIndexBuilder<'a> {
         self
     }
 
+    /// Set transaction properties to store with this commit.
+    ///
+    /// These key-value pairs are stored in the transaction file
+    /// and can be read later to identify the source of the commit
+    /// (e.g., job_id for tracking completed index jobs).
+    pub fn transaction_properties(mut self, properties: HashMap<String, String>) -> Self {
+        self.transaction_properties = Some(Arc::new(properties));
+        self
+    }
+
     #[instrument(skip_all)]
     pub async fn execute_uncommitted(&mut self) -> Result<IndexMetadata> {
         if self.columns.len() != 1 {
@@ -456,14 +469,15 @@ impl<'a> CreateIndexBuilder<'a> {
         } else {
             vec![]
         };
-        let transaction = Transaction::new(
+        let transaction = TransactionBuilder::new(
             new_idx.dataset_version,
             Operation::CreateIndex {
                 new_indices: vec![new_idx],
                 removed_indices,
             },
-            None,
-        );
+        )
+        .transaction_properties(self.transaction_properties.clone())
+        .build();
 
         self.dataset
             .apply_commit(transaction, &Default::default(), &Default::default())