feat(transaction): compound append + add-index via action transactions

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.

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;
 }

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"

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):
         """

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"

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)?;