feat: transaction files and commit conflict resolution (#1127)

* wip: outline conflict resolution impl

* wip: make transaction serialize

* wip: make transaction a separate file

* docs: write basic format docs for transactions

* get existing tests passing

* cleanup

* minor pr feedback

* fix: handle indices properly

* more fixes and tests

* merge changes

* handle params better

docs/format.rst

@@ -260,3 +260,35 @@ put-if-not-exists, these operations should be used. This is true of local file
 systems and cloud object stores, with the notable except of AWS S3. For ones
 that lack this functionality, an external locking mechanism can be configured
 by the user.
+
+Conflict resolution
+~~~~~~~~~~~~~~~~~~~
+
+If two writers try to commit at the same time, one will succeed and the other
+will fail. The failed writer should attempt to retry the commit, but only if
+it's changes are compatible with the changes made by the successful writer.
+
+The changes for a given commit are recorded as a transaction file, under the
+``_transactions`` prefix in the dataset directory. The transaction file is a
+serialized ``Transaction`` protobuf message. See the ``transaction.proto`` file
+for its definition.
+
+.. image:: _static/conflict_resolution_flow.png
+
+The commit process is as follows:
+
+ 1. The writer finishes writing all data files.
+ 2. The writer creates a transaction file in the ``_transactions`` directory.
+    This files describes the operations that were performed, which is used for two
+    purposes: (1) to detect conflicts, and (2) to re-build the manifest during
+    retries.
+ 3. Look for any new commits since the writer started writing. If there are any,
+    read their transaction files and check for conflicts. If there are any
+    conflicts, abort the commit. Otherwise, continue.
+ 4. Build a manifest and attempt to commit it to the next version. If the commit
+    fails because another writer has already committed, go back to step 3.
+ 5. If the commit succeeds, update the ``_latest.manifest`` file.
+
+When checking whether two transactions conflict, be conservative. If the
+transaction file is missing, assume it conflicts. If the transaction file 
+has an unknown operation, assume it conflicts.

protos/format.proto

@@ -100,6 +100,18 @@ message Manifest {
   // 
   // For a single file, will be zero.
   uint32 max_fragment_id = 11;
+
+  // Path to the transaction file, relative to `{root}/_transactions`
+  //
+  // This contains a serialized Transaction message representing the transaction
+  // that created this version.
+  //
+  // May be empty if no transaction file was written.
+  //
+  // The path format is "{read_version}-{uuid}.txn" where {read_version} is the
+  // version of the table the transaction read from, and {uuid} is a 
+  // hyphen-separated UUID.
+  string transaction_file = 12;
 } // Manifest
 
 // Auxiliary Data attached to a version.

protos/transaction.proto

@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = "proto3";
+
+import "format.proto";
+
+package lance.format.pb;
+
+// A transaction represents the changes to a dataset.
+//
+// This has two purposes:
+// 1. When retrying a commit, the transaction can be used to re-build an updated
+//    manifest.
+// 2. When there's a conflict, this can be used to determine whether the other
+//    transaction is compatible with this one.
+message Transaction {
+  // The version of the dataset this transaction was built from.
+  //
+  // For example, for a delete transaction this means the version of the dataset
+  // that was read from while evaluating the deletion predicate.
+  uint64 read_version = 1;
+
+  // The UUID that unique identifies a transaction.
+  string uuid = 2;
+
+  // Optional version tag.
+  string tag = 3;
+
+  // Add new rows to the dataset.
+  message Append {
+    // The new fragments to append.
+    //
+    // Fragment IDs are not yet assigned.
+    repeated DataFragment fragments = 1;
+  }
+
+  // Mark rows as deleted.
+  message Delete {
+    // The fragments to update
+    //
+    // The fragment IDs will match existing fragments in the dataset.
+    repeated DataFragment updated_fragments = 1;
+    // The fragments to delete entirely.
+    repeated uint64 deleted_fragment_ids = 2;
+    // The predicate that was evaluated
+    //
+    // This may be used to determine whether the delete would have affected 
+    // files written by a concurrent transaction.
+    string predicate = 3;
+  }
+
+  // Create or overwrite the entire dataset.
+  message Overwrite {
+    // The new fragments
+    //
+    // Fragment IDs are not yet assigned.
+    repeated DataFragment fragments = 1;
+    // The new schema
+    repeated Field schema = 2;
+    // Schema metadata.
+    map<string, bytes> schema_metadata = 3;
+  }
+
+  // Add a new secondary index.
+  message CreateIndex {
+    repeated IndexMetadata new_indices = 1;
+  }
+
+  // An operation that rewrites but does not change the data in the table. These
+  // kinds of operations just rearrange data.
+  message Rewrite {
+    // The old fragments that are being replaced
+    //
+    // These should all have existing fragment IDs.
+    repeated DataFragment old_fragments = 1;
+    // The new fragments
+    //
+    // These fragments IDs are not yet assigned.
+    repeated DataFragment new_fragments = 2;
+  }
+
+  // An operation that merges in a new column, altering the schema.
+  message Merge {
+    // The updated fragments
+    //
+    // These should all have existing fragment IDs.
+    repeated DataFragment fragments = 1;
+    // The new schema
+    repeated Field schema = 2;
+    // Schema metadata.
+    map<string, bytes> schema_metadata = 3;
+  }
+
+  // The operation of this transaction.
+  oneof operation {
+    Append append = 100;
+    Delete delete = 101;
+    Overwrite overwrite = 102;
+    CreateIndex create_index = 103;
+    Rewrite rewrite = 104;
+    Merge merge = 105;
+  }
+}

rust/build.rs

@@ -6,7 +6,11 @@ fn main() -> Result<()> {
     let mut prost_build = prost_build::Config::new();
     prost_build.protoc_arg("--experimental_allow_proto3_optional");
     prost_build.compile_protos(
-        &["./protos/format.proto", "./protos/index.proto"],
+        &[
+            "./protos/format.proto",
+            "./protos/index.proto",
+            "./protos/transaction.proto",
+        ],
         &["./protos"],
     )?;