Updated Transaction Model

[wjones127]

TLDR

Introduce CompositeOperation to allow a Transaction with multiple operations in one.

The new operation type could also be an opportunity to solve several other long-standing issues with transactions.

Design goals

Primary goal: Allow merging multiple transactions into one.

Part of merging will need to be conflict resolution. To make this possible, transactions need to serve as true diffs between manifests. In some respects already they do: an Append operation shows the new fragments added. But other operations are not helpful: in Update operation, we just provide the new fragments, but it's unclear which data files were modified. If we want to detect conflicts thoroughly, we need to be able to tell from just the transaction what changed.

Secondary goals we can accomplish while refactoring transactions:

• Provide a user-friendly view of dataset operation history (something like DESCRIBE HISTORY from Delta Lake or history metadata table from Iceberg)
• Allow formulating new transaction types without having to create new protobuf messages and new conflict resolution implementations.

Use cases

If Transaction can be merged, this unlocks three different use cases:

Use case 1: Higher write TPS

Currently, each transaction requires committing a new manifest to storage. This makes the sequential writing of this file a bottleneck. If you can merge transactions, then you could batch many operations, and commit several at once.

Use case 2: Simplified distributed commits

Right now, when users do distributed writes, they have to construct transactions themselves. This means handling the read_version and blob_operation themselves, which can sometimes be confusing. If instead we made each partial change return a transaction, we could merge these transactions later before committing the overall operation.

This is already implemented for append, where you can use InsertBuilder::execute_uncommitted() to write new data files, and it returns a Transaction. Meanwhile CommitBuilder::execute_batch() can take an array of Transaction.

Use case 3: Multi-statement transactions

Combining successive transactions could also be used to do multi-statement transactions. To make successive operations see the one before, this would have to be combined with detached commits. We could make several detached commits, then take the transaction files from those commits, merge them, and use that to commit the multi-statement transaction.

Proposed design

An early prototype exists here: #3204

Provide a new CompositeOperation type that is meant to be a generic operation type. It will be three levels:

• CompositeOperation represents the full operation that is applied to create the next version of the manifest
• Has multiple UserOperation, which represent a logical user-facing operation. (For example, delete multiple rows.)
• Each UserOperation contains multiple Actions, which represent a single change to the manifest.

To apply a CompositeOperation to create a new manifest, simply apply the flattened list of Actions in order.

Example

Consider the SQL query:

BEGIN TRANSACTION;
INSERT INTO t VALUES (1);
DELETE FROM t WHERE id = 32;
UPDATE t SET a = 2 WHERE id = 1;
COMMIT;

Could be represented as a single CompositeOperation:

- description: "INSERT INTO t VALUES (1);"
  uuid: 123e4567-e89b-12d3-a456-426655440001
  read_version: 0
  actions:
    - type: add_fragments
      fragments:
        - id: 10
          files:
            - path: data/123e4567-e89b-12d3-a456-426655440000.lance
              fields: [0]
- description: "DELETE FROM t WHERE id = 32;"
  uuid: 123e4567-e89b-12d3-a456-426655440002
  read_version: 0
  actions:
    - type: updated_fragments
      fragments:
        - id: 0
          files:
            - path: data/dfdsfdsd-e89b-12d3-a456-426655440000.lance
              fields: [0]
              deletion_file:
                - file_type: ARROW_ARRAY
                  read_version: 0
                  id: 10
                  num_deleted_rows: 1
- description: "UPDATE t SET a = 2 WHERE id = 1;"
  uuid: 123e4567-e89b-12d3-a456-426655440003
  read_version: 0
  actions:
    - type: updated_fragments
      fragments:
        - id: 0
          files:
            - path: data/123e4567-sdfs-12d3-sdfs-426655440000.lance
              fields: [0]
              deletion_file:
                - file_type: ARROW_ARRAY
                  read_version: 0
                  id: 10
                  num_deleted_rows: 1

lance/protos/transaction.proto

Lines 93 to 124 in 5ea6589

	message CompositeOperation {
	repeated UserOperation user_operations = 1;
	}
	// A logical grouping of actions that correspond to a single user action.
	message UserOperation {
	uint64 read_version = 1;
	string uuid = 2;
	string description = 4;
	repeated Action actions = 5;
	}
	// An action to apply to the manifest.
	message Action {
	message AddFragments {
	repeated DataFragment fragments = 1;
	}
	message DeleteFragments {
	repeated uint64 deleted_fragment_ids = 1;
	}
	message UpdateFragments {
	repeated DataFragment updated_fragments = 1;
	}
	oneof action {
	AddFragments add_fragments = 1;
	DeleteFragments delete_fragments = 2;
	UpdateFragments update_fragments = 3;
	}
	}

[majin1102]

Right now, when users do distributed writes, they have to construct transactions themselves. This means handling the read_version and blob_operation themselves, which can sometimes be confusing. If instead we made each partial change return a transaction, we could merge these transactions later before committing the overall operation.

If constructing transactions is somehow confusing, why make every write() a single transaction, then merge could help. I mean there will be more read_version and blob operation need be handled? Or is there something missed here

[wjones127]

You wouldn't have to directly handle read_version or blob_operation, because each individual write would return the Transaction object.

[majin1102]

Thanks for explaining, really look forward to this feature

[majin1102]

Can we put a DDL like drop_column and merge_column into one transaction? cause we have met the demand to implement a update_column operation(update the whole column by a stream or udf)

[wjones127]

Yeah, I see no reason why that wouldn't be possible.

[majin1102]

Excellent! As schema changes aren't documented, Is there any work to be done here?

[wjones127]

I think the main blocker is having a transaction type that can express that change.

[jackye1995]

@wjones127 I discussed with @majin1102 a bit on the transaction topic in #4219, want to revive this thread a bit:

could you clarify a bit about the data model we envision for transactions?

I see in the prototype, we are designing this 4 layer transaction layout in Transaction -> CompositeOperation -> UserOperation -> Action, but in the graphs and description of use case 2 and 3, we are leveraging CommitBuilder::execute_batch() can take an array of Transaction to be committed as a single one. This means we don't really need the 4 layers to achieve multi-statement transaction? It feels like we just need to expand the supported list of operations in execute_batch beyond append, or am I missing anything important here?

Given right now we have the conflict resolver, the logic for committing a list of transactions could basically be:

• for each transaction:
  • rebase and commit transaction against the latest version to produce new manifest and transaction file
  • manifest and transaction file not written to disk, to create an in-memory latest version
• commit the final manifest and transaction file to disk

One question becomes now that the committed transaction file need to include all the transaction information. Maybe this is why you are saying why we need a composite operation?

[jackye1995]

I see, so the correct way to understand Action is that it is basically "OperationV2" that is more fine-grained and describing changes only. It is intended to replace the current Operations.

With Action, the goal would be to describe everything in in terms of a change.

+100000 on that. When I was trying to add new features in the operations I always go back and forth between describing the change or the post-image because currently it's already a mixture of both. I think it is always better to describe changes so that it can be easily rebased and apply to the final manifest.

[jackye1995]

Provide a user-friendly view of dataset operation history (something like DESCRIBE HISTORY from Delta Lake or history metadata table from Iceberg)

I don't think Iceberg history metadata table does what you describe since it is always just describing the post image, and it also mix together all the changes in a multi-statement transaction. But I get what you mean now.

The backend could merge these into txn_merged

to confirm, this txn_merged would only exist in memory right? When persisted it will be in the form of separated user operations.

[jackye1995]

Okay I think I am getting what this means:

So a user can concurrently do txn1, txn2, tx3

So when user runs execute_batch([txn1, txn2, txn3]), it is translated to a single transaction with 3 user operations, so each user operation maps to its original transaction.

And this assumes that each original transaction also uses the CompositeOperation with a single UserOperation? Otherwise if the user commits a Operation then we cannot always derive what are the equivalent actions.

[wjones127]

and it also mix together all the changes in a multi-statement transaction

We should be clear about the different use cases here:

1. Multi-statement transaction (Use case 3): The end user runs multiple operations in a single transaction. Think BEGIN TRANSACTION; ...; COMMIT;. We might want this to show up as a single entry in SHOW HISTORY
2. Merged transactions for higher write TPS (Use case 1): As an optimization, a server might merge multiple user transactions (potentially from different users even) into one Lance transaction. These should still show up as separate in SHOW HISTORY.

[wjones127]

And this assumes that each original transaction also uses the CompositeOperation with a single UserOperation? Otherwise if the user commits a Operation then we cannot always derive what are the equivalent actions.

I see no reason it has to have a single UserOperation. Depends on the use case, whether it's (1) or (3). See comment above.

You are right that we need to always use CompositeOperation. I guess I'm hoping we can transition all transactions to use that, and deprecate the old Operation variants over time. At the very least, the ones passed to CommitBuilder::execute_batch() might only support CompositeOperation and a few straightforward ones like Append and CreateIndex.

[majin1102]

Why not just use a nested transaction list like:

message CompositeOperation {
  repeated Trasanction transactions = 1
}

We could have whatever we want layers transactions. Typical the leaf transactions under the same parent should share the same read_version(as well as uuid I think). none-leaf transactions should be equavilent to the 'UserOperation' you metiond(merged by detached mode?). From format specification, this seems easier to be explained as 'nested transactions'. WDYT @wjones127 @jackye1995

[jackye1995]

Yeah this actually is what I am also thinking mentally. I don't know if we should really support more than 1 level of recursion, that feels like an overkill. But this definition will basically allow execute_batch to just apply transactions one by one in memory and eventually commit, and also have a transaction file recording this batch execution.

In my mental model, having different read version is in theory okay, because we apply each transaction individually and as long as the transaction itself can be validly applied from its read version, it is okay to batch commit transaction starting at different points in time. In practice, a database system should enforce a consistent read version, but that is out of the concern at the storage commit layer.

[majin1102]

it is okay to batch commit transaction starting at different points in time.

I agree with this. But I'm a little confused for this point:

a database system should enforce a consistent read version, but that is out of the concern at the storage commit layer

For multi-statement transaction scenario, I think the leaf transactions under the same parent basicly means statements(because the transaction format is one statement one transaction/action, the leaf would be a statement wrapped by transaction struct). I think whether statements shared the same read version in one transaction is somehow determined by the isolation level of lance format.

Let's say we have a transaction contains:

statment1 -> read_version 10
statment2 -> read_version 11

This means inconsistent read happened. We've learned that Iceberg, Paimon, Delta and Hudi provide at least snapshot isolation which would not allow inconsistent read. I don't know if lance format should break that since it has provided the basic semantics of snapshot(I think inconsistent read is not unacceptable, just no benifits if we could provide higher isolation levels with no cost).

I think maybe I understand wrong about this: We can provide a higher tool like MergeCommit which could merge transactions into one commit with a larger scope of ACID, which brings efficience. This doen't break the snapshot isolation level because this is not a transaction model thing to engines or users. I don't know if we are talking about this or truely mean multi-statement could contain multiple read version? If so I think it may be necessary to make sure the reasonable isolation is provided.

The basic transaction model in my mental mind is: