feat: optimistic transaction protocol #632

roeap · 2022-06-09T18:18:15Z

Description

This PR adds a ConflictChecker struct for conflict resolution in cases of concurrent commit failures. The implementation is heavily inspired by the reference implementation. So far we have most tests from spark that specifically target conflict resolution covered.

Working on this I thought a bit about what we may consider going forward, as we move through the protocol versions :). In the end we could end up with three main structs that are involved in validating a commit.

The existing DataChecker, which validates and potentially mutates data when writing data files to disk. (Currently supports invariants)
The upcoming ConflictChecker, which checks if a commit can be re-tried in case of commit conflicts.
A new CommitChecker, which does a-priory validation of the commit itself (e.g. append only and other rules covered by tests in spark)

My hope is to get this PR merged right after we release 0.8.0, so there is some time to fill some holes and fully leverage the new feature for 0.9.0.

If folks agree, I would open some issues and start work on some follow-ups..

Follow-ups

Extend ConflictChecker support conflict resolution for streaming transactions
Implement CommitChecker
Deprecate old commit function.
Extend DataChecker.
Consolidate record batch writer implementations.

Related Issue(s)

part of #593

Documentation

houqp · 2022-06-13T03:59:31Z

Thank you @roeap for picking up this work! I will take a closer look at it this weekend. 👍 from me for including datafusion-expr if needed.

houqp · 2022-06-19T00:20:49Z

The structure looks good to me 👍 When we introduce a proper expression handling abstraction, we should upgrade what we have in partition handling with it as well:

delta-rs/rust/src/partitions.rs

Line 12 in cc62c49

pub enum PartitionValue<T> {

rust/src/conflict_checker.rs

rust/src/delta.rs

wjones127 · 2023-03-16T23:15:12Z

FYI I'm planning on reviewing this weekend. :)

@rtyler

# Description This PR refactors and extends the table configuration handling. The approach is analogous to how we and object_store handle configuration via storage properties. The idea is to provide a somewhat typed layer over the untyped configuration keys. There was one surprising thing along the way. From what I can tell, we may have been omitting the `delta.` prefix on the config keys we parse. So this would definitely be breaking behaviour, since we no longer recognize keys we were parsing before. We can in principle handle aliases for keys quite easily, but I was not sure what the desired behaviour is. cc @rtyler @xianwill - This change would probably affect `kafka-delta-ingest`, so especially interested in your opinions! # Related Issue(s) part of delta-io#632 # Documentation

# Description Another PR on the road to delta-io#632 - ~~keeping it a draft, as it is based on delta-io#1206~~ While the `commitInfo` action is defined as completely optional, spark and delta-rs write at the very least interesting, but often also quite helpful information into the commit info. To make it easier to work with and centralize some conventions, we introduce a `CommitInfo` struct, that exposes some of the fields at the top level. Additionally we harmonize a bit between spark and delta-rs conventions. # Related Issue(s) part of delta-io#632 # Documentation  --------- Co-authored-by: Will Jones <willjones127@gmail.com>

roeap · 2023-03-17T22:30:47Z

rust/src/action/mod.rs

+    pub fn read_whole_table(&self) -> bool {
+        match self {
+            // TODO just adding one operation example, as currently none of the
+            // implemented operations scan the entire table.
+            Self::Write { predicate, .. } if predicate.is_none() => false,
+            _ => false,
+        }
+    }


The reference implementaition allows for txn to "taint" the entire table, in which case we disregrad analysing the specific actions. The conflict checker covers this behavior in tests, but I haven't investigated yet, when this is actually set. Mainly leaving this fn, as a "reminder" for subsequent PRs soon to come.

Does overwrite read the entire table?

Or could you provide examples of which operations will?

Our operations should actually never use this directly right now. I went through the spark code base, and it seems this is invoked mostly in cases when a plan querying multiple tables with spark has a delta table set as a sink.. This usually uses generic plans and does not invoke the delta operations directly, or rather when we are not too sure what actually happened.

Did not get too deep though. Personally I am a bit conflicted on how to proceed with this. on one hand it does nothing useful right now. On the other hand, we can keep it as a reminder to mindful of this. While right now I think datafusion does not yet have the concept of a sink in their plans, I believe they might be added fairly soon. At this point we may encounter situations, where a plan wants to write to delta, but we have no way of knowing what was scanned.

That said, I do hope that we will be able to inspect the plan metrics and figure out what was read anyhow. Especially since we report the scanned files in out scan operator.

roeap · 2023-03-17T23:37:35Z

rust/src/operations/filesystem_check.rs

            &actions,
            DeltaOperation::FileSystemCheck {},
+            snapshot,
+            // TODO pass through metadata


I plan to do a pass through all our operations in the next Pr, where also I'll be focussing on testing the commit function, and addressing these kinds of TODOs.

roeap · 2023-03-17T23:42:27Z

rust/src/operations/transaction/state.rs

+    pub fn parse_predicate_expression(&self, expr: impl AsRef<str>) -> DeltaResult<Expr> {
+        let dialect = &GenericDialect {};
+        let mut tokenizer = Tokenizer::new(dialect, expr.as_ref());
+        let tokens = tokenizer
+            .tokenize()
+            .map_err(|err| DeltaTableError::GenericError {
+                source: Box::new(err),
+            })?;
+        let sql = Parser::new(dialect)
+            .with_tokens(tokens)
+            .parse_expr()
+            .map_err(|err| DeltaTableError::GenericError {
+                source: Box::new(err),
+            })?;
+
+        // TODO should we add the table name as qualifier when available?
+        let df_schema = DFSchema::try_from_qualified_schema("", self.arrow_schema()?.as_ref())?;
+        let context_provider = DummyContextProvider::default();
+        let sql_to_rel = SqlToRel::new(&context_provider);
+
+        Ok(sql_to_rel.sql_to_expr(sql, &df_schema, &mut Default::default())?)
+    }
+}


Parsing predicate expressions is actually kind of broken. Regardless of the schema we pass in, int fields are always recognized as i64. I hope, that the abstractions in the core project will eventually help us out. Until then I plan to either do a follow-up in re-writing these expressions, or, if this is a bug, addressing this upstream in datafusion.

roeap · 2023-03-17T23:43:37Z

rust/src/operations/transaction/test_utils.rs

Yet another set of test utilities, and yet another TODO for follow up PR, to consolidate our test helpers...

wjones127

Overall looks like a great improvement. I think we want to be more conservative in cases where there are multiple concurrent transactions that beat the candidate one. Also the error messages could be improved.

wjones127 · 2023-03-19T00:20:19Z

rust/src/action/mod.rs

            _ => None,
        }
    }
+
+    /// Denotes if the operation reads the entire table
+    pub fn read_whole_table(&self) -> bool {


This method and changes_data makes me wonder whether DeltaOperation should instead be a trait so that we require each instance to implement.

Agreed! If OK with you, I would defer exploring that to a follow up PR. To really get a feel for how that API should look like, I would like to work a bit more with operations that actually need the conflict resolution :).

That sounds good to me!

wjones127 · 2023-03-19T00:20:27Z

rust/src/action/mod.rs

+    pub fn read_whole_table(&self) -> bool {
+        match self {
+            // TODO just adding one operation example, as currently none of the
+            // implemented operations scan the entire table.
+            Self::Write { predicate, .. } if predicate.is_none() => false,
+            _ => false,
+        }
+    }


Does overwrite read the entire table?

rust/src/operations/transaction/state.rs

wjones127 · 2023-03-19T01:01:09Z

rust/src/operations/transaction/conflict_checker.rs

+        // add / read + no write
+        // transaction would have read file added by concurrent txn, but does not write data,
+        // so no real conflicting change even though data was added


Does read + no write mean no data change? like an optimize?

rust/src/operations/transaction/conflict_checker.rs

wjones127 · 2023-03-19T01:29:26Z

rust/src/operations/transaction/conflict_checker.rs

+            version_to_read += 1;
+        }
+
+        // TODO how to handle commit info for multiple read commits?


Yeah, if there are multiple versions, should we just error for now? If there are multiple commits that beat our candidate, it seems like we should generate a WinningCommitSummary for each transaction after read_version, right? And then checking our candidate commit against each of them?

Good point. Changed the logic, that out commit loop will always only check against the currently conflicting commit and as such the WinningCommitSummary will now summarize exactly one commit.

rust/src/operations/transaction/conflict_checker.rs

wjones127 · 2023-03-19T01:42:45Z

rust/src/action/mod.rs

+    pub fn read_whole_table(&self) -> bool {
+        match self {
+            // TODO just adding one operation example, as currently none of the
+            // implemented operations scan the entire table.
+            Self::Write { predicate, .. } if predicate.is_none() => false,
+            _ => false,
+        }
+    }


Or could you provide examples of which operations will?

Co-authored-by: Will Jones <willjones127@gmail.com>

roeap · 2023-03-28T03:59:37Z

@wjones127 - sorry for taking a long time to incorporate your feedback. I think it's mostly done now.

Mainly we now always check against the next version of commits. The questions around read_whole_table etc. I hope to address right after this.

As I plus I re-enabled an existing test for optimise, which we can now support again.

wjones127 · 2023-04-07T20:31:06Z

I am realizing we have another transaction implementation called DeltaTransaction, which is what we are currently using in Python. This improved implementation doesn't seem to have a low-level API where the user can provide the added and removed files, which is what we need for the PyArrow-based writer.

So I think we need to consolidate those into operations/transaction/mod.rs, make them public, and switch the Python implementation. That sounds right to you?

wjones127

This is looking good now.

The PySpark integration tests will be fixed soon. I should take a look at the sporadic failures in the Azure tests.

roeap · 2023-04-07T20:59:37Z

So I think we need to consolidate those into operations/transaction/mod.rs, make them public, and switch the Python implementation. That sounds right to you?

Yes it does. So far I have deliberately kept the implementation of commits used in the operations module separate from the existing one on the table. Where I am not sure is if we need the DeltaTransaction to be a struct at all, I think this is somehow an artifact from the reference implementation, where the Traction is conceptually used kind of like a context in python (it's been a long time since I wrote scala :D). In the new trasaction module we just have the commit function, which we should make public and which I think will give us what we need in python.

This was referenced Jun 19, 2022

[WIP] Move version field into DeltaTableState struct #553

Closed

refactor: move version field to DeltaTableState #649

Merged

roeap dismissed a stale review via cd8609b June 21, 2022 15:24

roeap force-pushed the conflict-checker branch from e745600 to cd8609b Compare June 21, 2022 15:24

feat: initial conflict checker methods

aedecde

roeap dismissed a stale review via aedecde June 27, 2022 22:20

roeap force-pushed the conflict-checker branch from cd8609b to aedecde Compare June 27, 2022 22:20

wjones127 reviewed Jun 27, 2022

View reviewed changes

rust/src/conflict_checker.rs Outdated Show resolved Hide resolved

roeap added 2 commits June 28, 2022 21:22

Merge branch 'main' into conflict-checker

224b826

initialize conflict checker

3243ee5

roeap mentioned this pull request Jun 29, 2022

[discussion] move delta log handling to new struct DeltaLog and harmonize operations #661

Closed

roeap added 6 commits June 30, 2022 00:02

feat: collect more transaction infos

bf55226

docs: add some comments

595e91b

Merge branch 'main' into conflict-checker

9c4bb02

Merge branch 'main' into conflict-checker

139c1a4

chore: small fixes

fcc7c90

Merge branch 'main' into conflict-checker

bb4cfeb

This was referenced Jul 17, 2022

feat: sharable reference to storage backend #697

Merged

Support get_backend_for_uri_with_options in Read Path #689

Closed

roeap mentioned this pull request Aug 6, 2022

Prune scanned files on column stats #724

Merged

roeap mentioned this pull request Sep 6, 2022

cleanup errors and make DeltaWriterError internal #784

Merged

roeap mentioned this pull request Sep 16, 2022

chore: break up datafusion imports #821

Closed

roeap mentioned this pull request Sep 27, 2022

feat: rewrite operations #852

Merged

MrPowers mentioned this pull request Oct 4, 2022

Roadmap 2022 H2 (discussion) delta-io/delta#1307

Open

Merge branch 'main' into conflict-checker

1460d91

github-actions bot added delta-rs-crate rust labels Dec 30, 2022

andrei-ionescu reviewed Dec 31, 2022

View reviewed changes

roeap requested review from rtyler and mosyp as code owners March 11, 2023 21:11

roeap requested a review from wjones127 March 11, 2023 21:12

roeap changed the title ~~Add ConflictChecker for concurrency control~~ feat: optimistic transaction protocol Mar 11, 2023

roeap added 2 commits March 17, 2023 18:33

chore: cleanup

54b1a85

Merge branch 'main' into conflict-checker

3d801e0

roeap dismissed a stale review via 3d801e0 March 17, 2023 17:36

roeap added 2 commits March 17, 2023 18:42

chore: clippy

3eb23b3

chore: parquet2 clippy

d390884

roeap commented Mar 17, 2023

View reviewed changes

wjones127 requested changes Mar 19, 2023

View reviewed changes

roeap and others added 3 commits March 19, 2023 08:56

Apply suggestions from code review

68f1b6e

Co-authored-by: Will Jones <willjones127@gmail.com>

fix: chack conflicts against next commit only

740eeff

test: re-enable ignored optimize test

bfcb393

roeap requested a review from wjones127 March 28, 2023 03:59

wjones127 mentioned this pull request Mar 29, 2023

Data is replicated when overwriting data concurrently #1254

Closed

roeap added 2 commits April 4, 2023 07:35

Merge branch 'main' into conflict-checker

1b1e4ad

chore: rename PrefixObjectStore imports

4cbd158

wjones127 approved these changes Apr 7, 2023

View reviewed changes

roeap merged commit 490122c into delta-io:main Apr 7, 2023

roeap deleted the conflict-checker branch April 7, 2023 21:01

This was referenced Apr 30, 2023

feat: add optimize command in python binding #1313

Merged

Support Optimize on non-append-only tables #1125

Closed

docs: update message since we fixed the commit logic #1318

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: optimistic transaction protocol #632

feat: optimistic transaction protocol #632

roeap commented Jun 9, 2022 •

edited

Loading

houqp commented Jun 13, 2022

houqp commented Jun 19, 2022

wjones127 commented Mar 16, 2023

roeap Mar 17, 2023

wjones127 Mar 19, 2023

wjones127 Mar 19, 2023

roeap Mar 28, 2023

roeap Mar 17, 2023

roeap Mar 17, 2023

roeap Mar 17, 2023

wjones127 left a comment

wjones127 Mar 19, 2023

roeap Mar 28, 2023

wjones127 Mar 28, 2023

wjones127 Mar 19, 2023

wjones127 Mar 19, 2023

wjones127 Mar 19, 2023

roeap Mar 28, 2023

wjones127 Mar 19, 2023

roeap commented Mar 28, 2023

wjones127 commented Apr 7, 2023

wjones127 left a comment

roeap commented Apr 7, 2023

feat: optimistic transaction protocol #632

feat: optimistic transaction protocol #632

Conversation

roeap commented Jun 9, 2022 • edited Loading

Description

Follow-ups

Related Issue(s)

Documentation

houqp commented Jun 13, 2022

houqp commented Jun 19, 2022

wjones127 commented Mar 16, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wjones127 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

roeap commented Mar 28, 2023

wjones127 commented Apr 7, 2023

wjones127 left a comment

Choose a reason for hiding this comment

roeap commented Apr 7, 2023

roeap commented Jun 9, 2022 •

edited

Loading