feat: add CompositeOperation for multi-statement transactions

docs/src/format/table/transaction.md

@@ -498,6 +498,36 @@ An UpdateBases operation only modifies the base paths. As a result, it only conf
 UpdateBases operations and even then only conflicts if the two operations have base paths with the
 same id, name, or path.
 
+### Composite
+
+An ordered list of operations applied sequentially as one atomic commit. Each sub-operation is
+applied to the manifest produced by the previous one, and the final manifest is committed as a
+single new version. This enables multi-statement transactions — for example, an append and an
+index update over the appended fragments (with fragment IDs reserved via ReserveFragments ahead
+of time) that become visible atomically.
+
+Each sub-transaction carries only its `operation`; the outer transaction's `read_version`, `uuid`,
+`tag`, and `transaction_properties` govern the commit, and the corresponding inner fields are left
+unset by writers and ignored by readers. The sub-operations must not contain Restore, Clone, or a
+nested Composite, and the list must not be empty.
+
+<details>
+<summary>CompositeOperation protobuf message</summary>
+
+```protobuf
+%%% proto.message.CompositeOperation %%%
+```
+
+</details>
+
+#### Composite Compatibility
+
+A Composite operation conflicts with a concurrent transaction if and only if at least one of its
+sub-operations conflicts with it, applying each sub-operation's own compatibility rules. When the
+concurrent transaction is itself a Composite, it is decomposed the same way, so two Composite
+operations conflict if any pair of their sub-operations conflicts. Rebases are likewise delegated:
+each sub-operation is rebased under its own rules and the results are reassembled in order.
+
 ## Conflict Resolution
 
 ### Terminology

java/lance-jni/src/transaction.rs

@@ -575,7 +575,12 @@ fn convert_to_java_operation_inner<'local>(
             "(I)V",
             &[JValue::Int(num_fragments as i32)],
         )?),
-        _ => unimplemented!(),
+        // A panic here unwinds across the JNI boundary and aborts the JVM,
+        // so unmapped operations must surface as errors instead.
+        other => Err(Error::unsupported_error(format!(
+            "operation {} is not supported by the Java binding yet",
+            other.name()
+        ))),
     }
 }
 
@@ -1303,7 +1308,14 @@ fn convert_to_rust_operation(
                 removed_indices,
             });
         }
-        _ => unimplemented!(),
+        // A panic here unwinds across the JNI boundary and aborts the JVM,
+        // so unmapped operations must surface as errors instead.
+        other => {
+            return Err(Error::unsupported_error(format!(
+                "operation {} is not supported by the Java binding yet",
+                other
+            )));
+        }
     };
     Ok(op)
 }

protos/transaction.proto

@@ -329,6 +329,23 @@ message Transaction {
     repeated BasePath new_bases = 1;
   }
 
+  // An ordered list of operations applied sequentially as one atomic commit.
+  //
+  // Each sub-transaction's `operation` is applied to the manifest produced by
+  // the previous one; the resulting manifest is committed as a single new
+  // version. Only the `operation` field of each sub-transaction is meaningful:
+  // the outer transaction's `read_version`, `uuid`, `tag` and
+  // `transaction_properties` govern the commit, and the corresponding inner
+  // fields are left unset by writers and ignored by readers. Inner fields are
+  // reserved for future per-statement metadata (e.g. merging independently
+  // created transactions while preserving their identity).
+  //
+  // Sub-transactions must not contain `Restore`, `Clone`, or a nested
+  // `CompositeOperation`, and the list must not be empty.
+  message CompositeOperation {
+    repeated Transaction transactions = 1;
+  }
+
   // The operation of this transaction.
   oneof operation {
     Append append = 100;
@@ -346,6 +363,9 @@ message Transaction {
     UpdateMemWalState update_mem_wal_state = 112;
     Clone clone = 113;
     UpdateBases update_bases = 114;
+    // Field 115 is intentionally skipped: it is used by an in-flight proposal
+    // (https://github.com/lance-format/lance/pull/7674).
+    CompositeOperation composite = 116;
   }
 
   // Fields 200/202 (`blob_append` / `blob_overwrite`) previously represented blob dataset ops.

python/python/lance/dataset.py

@@ -4580,10 +4580,15 @@ def commit(
             and not isinstance(
                 operation, (LanceOperation.Overwrite, LanceOperation.Restore)
             )
+            and not (
+                isinstance(operation, LanceOperation.Composite)
+                and len(operation.operations) > 0
+                and isinstance(operation.operations[0], LanceOperation.Overwrite)
+            )
         ):
             raise ValueError(
                 "read_version is required for all operations except "
-                "Overwrite and Restore"
+                "Overwrite, Restore, and Composite starting with Overwrite"
             )
 
         # Storage options provider is automatically created in Rust when
@@ -6073,6 +6078,67 @@ class UpdateConfig(BaseOperation):
         schema_metadata_updates: Optional[LanceOperation.UpdateMap] = None
         field_metadata_updates: Optional[Dict[int, LanceOperation.UpdateMap]] = None
 
+    @dataclass
+    class Composite(BaseOperation):
+        """
+        An ordered list of operations applied sequentially as one atomic commit.
+
+        Each operation is applied to the result of the previous one, and the
+        final state is committed as a single new version. This allows
+        multi-statement transactions such as an append and an index update
+        that become visible atomically.
+
+        The operations must not contain ``Restore``, ``Clone``, or a nested
+        ``Composite``. Sub-operations must not reference fragment IDs that
+        are assigned to fragments added earlier in the same composite —
+        such IDs are not stable across commit retries. To reference new
+        fragments (e.g. an index over appended data), reserve fragment IDs
+        first and set them on the fragments explicitly.
+
+        Attributes
+        ----------
+        operations: list[LanceOperation.BaseOperation]
+            The operations to apply in order.
+
+        Examples
+        --------
+
+        >>> import lance
+        >>> import pyarrow as pa
+        >>> tab1 = pa.table({"a": [1, 2], "b": ["a", "b"]})
+        >>> dataset = lance.write_dataset(tab1, "example")
+        >>> tab2 = pa.table({"a": [3, 4], "b": ["c", "d"]})
+        >>> fragment = lance.fragment.LanceFragment.create("example", tab2)
+        >>> operation = lance.LanceOperation.Composite(
+        ...     [
+        ...         lance.LanceOperation.Append([fragment]),
+        ...         lance.LanceOperation.Delete([], [0], "a < 3"),
+        ...     ]
+        ... )
+        >>> dataset = lance.LanceDataset.commit("example", operation,
+        ...                                     read_version=dataset.version)
+        >>> dataset.to_table().to_pandas()
+           a  b
+        0  3  c
+        1  4  d
+        """
+
+        operations: Iterable[LanceOperation.BaseOperation]
+
+        def __post_init__(self):
+            operations = list(self.operations)
+            invalid = [
+                op
+                for op in operations
+                if not isinstance(op, LanceOperation.BaseOperation)
+            ]
+            if invalid:
+                raise TypeError(
+                    "operations must all be LanceOperation.BaseOperation, "
+                    f"got {[type(op) for op in invalid]}"
+                )
+            self.operations = operations
+
 
 @dataclass
 class ColumnOrdering:

python/python/tests/test_dataset.py

@@ -1838,6 +1838,58 @@ def test_append_with_commit(tmp_path: Path):
     assert tbl == expected
 
 
+def test_composite_with_commit(tmp_path: Path):
+    table = pa.Table.from_pydict({"a": range(100), "b": range(100)})
+    base_dir = tmp_path / "test"
+    dataset = lance.write_dataset(table, base_dir)
+
+    fragment = lance.fragment.LanceFragment.create(base_dir, table)
+    composite = lance.LanceOperation.Composite(
+        [
+            lance.LanceOperation.Append([fragment]),
+            lance.LanceOperation.UpdateConfig(
+                config_updates=lance.LanceOperation.UpdateMap(
+                    updates={"composite.key": "value"}, replace=False
+                )
+            ),
+        ]
+    )
+
+    with pytest.raises(ValueError):
+        # Must specify read version
+        lance.LanceDataset.commit(dataset, composite)
+
+    dataset = lance.LanceDataset.commit(dataset, composite, read_version=1)
+
+    # Both effects are visible in a single new version.
+    assert dataset.version == 2
+    assert dataset.count_rows() == 200
+    assert dataset.config()["composite.key"] == "value"
+
+    # The committed transaction reads back as a composite.
+    transaction = dataset.read_transaction(2)
+    assert isinstance(transaction.operation, lance.LanceOperation.Composite)
+    assert len(transaction.operation.operations) == 2
+    assert isinstance(transaction.operation.operations[0], lance.LanceOperation.Append)
+
+
+def test_composite_validation():
+    with pytest.raises(TypeError):
+        lance.LanceOperation.Composite(operations="not-a-list")
+
+    data = pa.Table.from_pydict({"a": range(10), "b": range(10)})
+    dataset = lance.write_dataset(data, "memory://composite-validation")
+    empty = lance.LanceOperation.Composite(operations=[])
+    with pytest.raises(Exception, match="at least one"):
+        lance.LanceDataset.commit(dataset, empty, read_version=1)
+
+    nested = lance.LanceOperation.Composite(
+        operations=[lance.LanceOperation.Composite(operations=[])]
+    )
+    with pytest.raises(Exception, match="cannot contain"):
+        lance.LanceDataset.commit(dataset, nested, read_version=1)
+
+
 def test_commit_batch_append():
     data1 = pa.Table.from_pydict({"a": range(100), "b": range(100)})
     dataset = lance.write_dataset(data1, "memory://test")

python/src/transaction.rs

@@ -390,6 +390,11 @@ impl FromPyObject<'_, '_> for PyLance<Operation> {
                 };
                 Ok(Self(op))
             }
+            "Composite" => {
+                let operations = extract_vec(&ob.getattr("operations")?)?;
+                let op = Operation::Composite { operations };
+                Ok(Self(op))
+            }
             unsupported => Err(PyValueError::new_err(format!(
                 "Unsupported operation: {unsupported}",
             ))),
@@ -595,6 +600,13 @@ impl<'py> IntoPyObject<'py> for PyLance<&Operation> {
                     base_op.call0()
                 }
             }
+            Operation::Composite { operations } => {
+                let operations = export_vec(py, operations.as_slice())?;
+                let cls = namespace
+                    .getattr("Composite")
+                    .expect("Failed to get Composite class");
+                cls.call1((operations,))
+            }
             _ => todo!(),
         }
     }

rust/lance/src/dataset.rs

@@ -738,16 +738,17 @@ impl Dataset {
             let message_len =
                 LittleEndian::read_u32(&last_block[offset_in_block..offset_in_block + 4]) as usize;
             let message_data = &last_block[offset_in_block + 4..offset_in_block + 4 + message_len];
-            let transaction: Transaction =
-                lance_table::format::pb::Transaction::decode(message_data)?.try_into()?;
-
-            let metadata_cache = session.metadata_cache.for_dataset(uri);
-            let metadata_key = TransactionKey {
-                version: manifest_location.version,
-            };
-            metadata_cache
-                .insert_with_key(&metadata_key, Arc::new(transaction))
-                .await;
+            if let Some(transaction) =
+                decode_inline_transaction(message_data, manifest_location.version)
+            {
+                let metadata_cache = session.metadata_cache.for_dataset(uri);
+                let metadata_key = TransactionKey {
+                    version: manifest_location.version,
+                };
+                metadata_cache
+                    .insert_with_key(&metadata_key, Arc::new(transaction))
+                    .await;
+            }
         }
 
         if manifest.should_use_legacy_format() {
@@ -1368,11 +1369,15 @@ impl Dataset {
         detached: bool,
     ) -> Result<Self> {
         let read_version = read_version.map_or_else(
-            || match operation {
-                Operation::Overwrite { .. } | Operation::Restore { .. } => Ok(0),
-                _ => Err(Error::invalid_input(
-                    "read_version must be specified for this operation",
-                )),
+            || {
+                if operation.overwrites_dataset() || matches!(operation, Operation::Restore { .. })
+                {
+                    Ok(0)
+                } else {
+                    Err(Error::invalid_input(
+                        "read_version must be specified for this operation",
+                    ))
+                }
             },
             Ok,
         )?;
@@ -3662,6 +3667,31 @@ impl ManifestWriteConfig {
     }
 }
 
+/// Decode an inline transaction section for opportunistic caching.
+///
+/// Returns `None` instead of failing when the transaction cannot be decoded:
+/// the section may have been written by a newer version of Lance with an
+/// operation type this version does not know, and that must not prevent
+/// opening the dataset. Paths that need the transaction contents surface the
+/// error at their call sites instead.
+fn decode_inline_transaction(message_data: &[u8], version: u64) -> Option<Transaction> {
+    match lance_table::format::pb::Transaction::decode(message_data)
+        .map_err(Error::from)
+        .and_then(Transaction::try_from)
+    {
+        Ok(transaction) => Some(transaction),
+        Err(err) => {
+            log::warn!(
+                "Failed to decode the inline transaction of version {}; \
+                 it may have been written by a newer version of Lance: {}",
+                version,
+                err
+            );
+            None
+        }
+    }
+}
+
 /// Commit a manifest file and create a copy at the latest manifest path.
 #[allow(clippy::too_many_arguments)]
 pub(crate) async fn write_manifest_file(

rust/lance/src/dataset/tests/dataset_transactions.rs

@@ -208,6 +208,35 @@ async fn test_session_store_registry() {
     assert_eq!(registry.active_stores().len(), 0);
 }
 
+#[test]
+fn test_decode_inline_transaction_tolerates_unknown_operations() {
+    use crate::dataset::decode_inline_transaction;
+    use lance_table::format::pb;
+    use prost::Message;
+
+    // A transaction written by a newer version of Lance may carry an operation
+    // this version cannot decode; prost surfaces it as a missing oneof. This
+    // must not fail (it would prevent opening the dataset), only skip caching.
+    let unknown_operation = pb::Transaction {
+        read_version: 1,
+        uuid: "test".to_string(),
+        ..Default::default()
+    };
+    assert!(decode_inline_transaction(&unknown_operation.encode_to_vec(), 42).is_none());
+
+    // Corrupt bytes are likewise tolerated.
+    assert!(decode_inline_transaction(&[0xff, 0xff, 0xff], 42).is_none());
+
+    // A decodable transaction is returned.
+    let known = pb::Transaction::from(&Transaction::new(
+        1,
+        Operation::Append { fragments: vec![] },
+        None,
+    ));
+    let decoded = decode_inline_transaction(&known.encode_to_vec(), 42).unwrap();
+    assert!(matches!(decoded.operation, Operation::Append { .. }));
+}
+
 #[tokio::test]
 async fn test_migrate_v2_manifest_paths() {
     let test_uri = TempStrDir::default();