Core: Add Catalog Transactions API #6948

nastra · 2023-02-27T12:25:38Z

I have written up a design doc, which is available here

I think eventually we'd want to split this into 2 PRs, so that the introduced APIs can be reviewed independently from the REST-specific things.

jackye1995 · 2023-06-27T16:24:55Z

core/src/main/java/org/apache/iceberg/catalog/CatalogTransaction.java

+  enum IsolationLevel {
+
+    /**
+     * All reads that are being made will see the last committed values that existed when the table


I have doubt about this definition:

All reads that are being made will see the last committed values that existed when the table was loaded first inside the catalog transaction.

This is different from the definition in other places like here:

SNAPSHOT isolation specifies that data read within a transaction will never reflect changes made by other simultaneous transactions.

The key difference is when the table was loaded first inside the catalog transaction, which means there will be a time difference between when 2 tables are loaded in the transaction.

Consider the following case of 2 processes:

process 1:

t0: CatalogTransaction catalogTransaction = catalog().createTransaction(isolationLevel); Catalog txCatalog = catalogTransaction.asCatalog(); t1: Table table1 = txCatalog.load(TableIdentifier.of("db", "table1")) t2: Table table2 = txCatalog.load(TableIdentifier.of("db", "table2"))

process 2:

t1.5: table2.newAppend().addFiles(...).commit()

This means the state of tables in the same transaction in process1 is different.

When translated to a real-life SQL use case, it means:

process 1:

SELECT * FROM table1 JOIN table2 ON ...

process 2:

INSERT INTO table2 ...

has a chance to cause a phantom read in process 1, and that clearly violates the isolation guarantee

@jackye1995, I think the behavior for the situation you're describing is an open question for the snapshot isolation level. First, though, I want to cover some background on how we think about serializable isolation to make sure we are thinking about "simultaneous" the same way.

For serializable, the requirement is that there exists some ordering of transactions that produces the correct result. Say transaction 1 starts and reads from a table, then transaction 2 commits to that table, and finally transaction 1 attempts to commit. Iceberg will allow the commit as long as the result of the commit is not affected by the changes made by transaction 2. That is, Iceberg reorders the transactions if the reads would have produced the same result. This relies on checking conflicts in tables that have been updated. And the idea of when a transaction happens or starts is flexible in this model.

Another complication is the idea of a point in time in a catalog. There's no guarantee that catalogs have an order of changes that applies across tables. The only requirement imposed by Iceberg is that there is a linear history for any single table and there isn't a general happens-before relationship between commits across tables. There is, however, a limited relationship between tables involved in transactions.

I think there's a good argument that we can't just use that as a reason not to do anything. After all, I can have situations where completely ignoring the point-in-time concern is clearly wrong:

CREATE TABLE a (id bigint); CREATE TABLE b (id bigint); INSERT INTO a VALUES (1); -- Time T1 BEGIN TRANSACTION; INSERT INTO b SELECT id FROM a; TRUNCATE TABLE a; COMMIT; -- Time T2

If I were to read from a at T1 and read from b at T2, then I could see row(id=1) twice, even though there was never a "time" when it was in both tables because the transaction was atomic. I don't think I'd call this a "dirty read" because that means uncommitted data was read, but it's still clearly a problem.

I think there are 3 options:

Coordinate table loading with the catalog

Use validations that data read has not changed

Define snapshot isolation differently, or change the name to something else

Right now, the difference between 2 and 3 is basically adding checks so we don't have to decide right away. That's why it's like this today.

1. Coordinate table loading with the catalog

To coordinate table loading, we'd add something like startTransaction() to get a transaction state. Then we'd pass that state back to the catalog in order to load at the start time. You can imagine the catalog passing back basically a transaction ID and loading tables using the state of that TID.

I don't think this is a viable option. It's a lot of work to define the APIs and there are still catalogs that can't implement it. The result is that we'd end up with inconsistent behavior across catalogs, where some catalogs just load the latest copy of the table because that's all they can do.

2. Use validations that data read has not changed

This option is similar to how serializable is handled. The transaction "start" time floats -- you can use data that was read as long as when the transaction commits, the outcome would be the same.

The main drawback of this approach is that it isn't clear whether this is actually distinct from serializable, which basically does the same validation.

3. Define snapshot isolation differently or rename it

This option may seem crazy, but there's a good argument for it. If there's little difference between serializable and snapshot isolation with option 2, then serializable should be used for cases like the SQL above. However, we clearly want to be able to relax the constraints for certain cases:

Simultaneous DELETE FROM and INSERT INTO -- it makes little sense to fail a delete because a matching row was just inserted, when a slightly different order (delete commits first) would have been perfectly fine

External coordination -- running a scheduled MERGE INTO that selects from a source table that's constantly updated should generally succeed, even if the source table is modified. Most of the time, consumption will be incremental and coordinated externally. The important thing is knowing what version of the source table was used, not failing if it is updated during the job.

These two cases align with the existing behavior of snapshot isolation in single-table commits -- but it makes sense there because "snapshot" applies to a single table. If the table is simultaneously modified, the original version is used and new data is ignored. However, deletions are validated.

Last comment

I think that captures the background for the decision that we need to make on this. Right now, I think the best approach is to collect use cases and think about how they should be handled, rather than making a decision.

Sorry for my late reply on this topic.

The current implementation does perform option 2, where it detects that read data hasn't been changed. This is being done for snapshot as well as serializable isolation.

The only difference between snapshot and serializable isolation in the current implementation is that serializable also detects and avoids write skew.

@jackye1995 did you have a chance to look at the implementation and the tests or is the definition of snapshot isolation confusing?

rdblue · 2023-07-03T21:00:05Z

core/src/main/java/org/apache/iceberg/BaseScan.java

@@ -99,7 +99,7 @@ protected Schema tableSchema() {
    return schema;
  }

-  protected TableScanContext context() {
+  public TableScanContext context() {


This should not be made public. I think we'll need to find a different way to report context.

rdblue · 2023-07-03T21:04:42Z

core/src/main/java/org/apache/iceberg/catalog/SupportsCatalogTransactions.java

+  /**
+   * Create a new {@link CatalogTransaction} with the given {@link IsolationLevel}.
+   *
+   * @param isolationLevel The isolation level to use.


We usually omit . from situations like this.

jackye1995 · 2023-07-03T21:12:49Z

Discussed offline with Ryan, Daniel and Eduard. I think we should be good here.

The requirement to have all table snapshots consistent at the starting time of transaction is not a hard requirement in the definition of snapshot or serializable isolation.

For the non-repeatable or phantom read issue raised, the issue has to happen in a repeated read of the same table in the same transaction. So in the example I gave, table2 is only read once, so whatever state is at the catalog load time could be considered the valid status and not a non-repeatable or phantom read.

There are 2 cases which could cause phantom read:

a self join query
multiple reads to the same table in a multi-statement transaction

In the first case, although not strictly enforced, but most query engines cache the metadata when fetching table from catalog, so it is not possible to read new data in the second scan of the table. This is something that ideally engines should enforce, but definitely not something that catalog should enforce.

In the second case, a multi-statement transaction must implement the proposed transaction API, so we should be good there.

I will continue to think about any other edge cases, but at least we are not blocked by this.

rdblue · 2023-07-04T21:10:37Z