feat: rewrite operations #852

roeap · 2022-09-27T21:32:57Z

Description

This PR incorporates some of the learnings with regards to how datafusion "should" be used (or what I think I understood so far how it should be used) and how this applies to our operations module. It also embraces the IntoFuture trait stabilized in rust 1.64.

More specifically:

Model operations as builders that implement IntoFuture
While we figure out how to deal with state sharing - just consume the table as it needs to be updated anyhow (right?)
push datafusion dependencies into command implementations, so we can more easily extend to commands that do not require datafuison.
have command-spcecific errors that map to DeltaTableError (the top level error variants likely have to be refined, for now, many command errors map to a GenericError, which at least shows meaningful error messages.)
an updated PartitionWriter implementation that allows more fine graunlar control on how data is written

Some of choices or trials here mainly (hopefully) make sense when viewed as preparation for what's to come.

For the conflict resolution (feat: optimistic transaction protocol #632), we want to leverage some information on especially the scanned files during an operation. The plan is to leverage the execution metrics from the datafusion plans for this. We already do this in some of the datafusion tests, to make sure we properly use the file metrics in scans.
The new partition writer implementation is kept separate from the existing one for now, to be able to more easily iterate.
I replicated a minimal implementation of the transactions, but the main update for this will follow with conflict resolution.

To keep it somewhat simpler to review I tried to keep the major changes contained in the operations module. If we adopt this the idea is to migrate all existing operations (create, vacuum, optimize) to the builder pattern and into the operations module, and have the methods on DeltaTable just return a pre-populated builder. I think this is where IntoFuture shines, as we can await them just like before.

Implementing this, one of the things I found most convenient, is that we now have the memory:// store available. Coupled with the sync_stores helper, it makes setting up and validating test cases that mutate an existing table very convenient - yay object_store 😆.

There are some things I want to clean up, but it would be great to get some feedback on if this is where we want to go - @houqp @wjones127 @fvaleye.

Related Issue(s)

Documentation

wjones127 · 2022-09-28T04:01:20Z

rust/src/operations/create.rs

+            )
+        };
+
+        // TODO configure more permissive versions based on configuration. Also how should this ideally be handled?


We should write a function that looks at a list of actions and detects which features are used, then determine the protocol versions. It will be much easier if that logic is the responsibility of a single function than spread out.

I think for updates we only need to run that function on new actions and merge the result with the existing protocol versions on the table. There's probably also some concurrency resolution that needs to happen too, but I haven't yet thought about that part much..

I might even say we should let clients set the protocol; we should consider that the responsibility of the library. What do you think? Are there use cases where they should set it?

I think you are right - having the user choose the protocol and defaulting to "what spark uses" seems like the way to go. WE may in that case have to check option compatibility though?

wjones127 · 2022-09-28T04:06:36Z

rust/src/operations/create.rs

+            if table.object_store().is_delta_table_location().await? {
+                match mode {
+                    SaveMode::ErrorIfExists => return Err(CreateError::TableAlreadyExists.into()),
+                    SaveMode::Append => return Err(CreateError::AppendNotAllowed.into()),


Why is append not allowed?

it's what spark does :)

wjones127 · 2022-09-28T04:06:49Z

rust/src/operations/create.rs

+                    SaveMode::Overwrite => {
+                        let curr_files =
+                            flatten_list_stream(table.object_store().as_ref(), None).await?;
+                        table.object_store().delete_batch(&curr_files).await?;


Why delete the current files?

Well .. in this case I was not sure, but conceptually I felt we are not overwriting the data in a table and creating a new version, but creating an entirely new table at version 0. If we were to support also updating table metadata, schema etc via this route, I guess there is more work to be done here validating all changes.

But this one I was definitely unsure about.

Had a look at the spark implementation - seems they agree with you and updating the metadata and evoling the table is the way to go

https://github.com/delta-io/delta/blob/1f6ab824e14794c17202b5e4e5df6a95357a799c/core/src/main/scala/org/apache/spark/sql/delta/commands/CreateDeltaTableCommand.scala#L196-L208

Since the metadata update is a larger operation, I opted for raising not implemented for now in this PR. I opened #917 to track this.

wjones127

I really like these new Builder APIs so far.

I do still like the idea of a having a three-tier API: (action-based, engine-agnostic, DataFusion-based). I think for clarity, it would be best to have those in separate modules of the crate. So, for example, I don't think the Create command and the Load/Write commands belong in the same module, since the first you pass in actions (low-level) while the others are more high-level and deal with data. I'll think about this some more though, since I'm not 100% sure this makes sense.

wjones127 · 2022-09-30T02:38:11Z

So, for example, I don't think the Create command and the Load/Write commands belong in the same module, since the first you pass in actions (low-level) while the others are more high-level and deal with data. I'll think about this some more though, since I'm not 100% sure this makes sense.

Okay I think I mis-read the purpose of Create: it's not a low-level API, it's just to create a table, so there's isn't any data writing interaction involved. So we should just think of operations module as the DataFusion-based high-level API. 👍

wjones127

I'm only halfway through and will try to wrap up reviewing tomorrow. Have some initial comments / suggestions.

rust/src/operations/create.rs

rust/src/operations/load.rs

wjones127 · 2022-10-26T04:29:51Z

rust/src/operations/mod.rs

+    /// Create a new [`DeltaOps`] instance, backed by an un-initialized in memory table
+    ///
+    /// Using this will not persist any changes beyond the lifetime of the table object.
+    /// THe main purpose of in-memory tables is for use in testing.


Suggested change

/// THe main purpose of in-memory tables is for use in testing.

/// The main purpose of in-memory tables is for use in testing.

wjones127 · 2022-10-26T04:30:11Z

rust/src/operations/mod.rs

+    /// let ops = DeltaOps::new_in_memory();
+    /// ```
+    #[must_use]
+    pub fn new_in_memory() -> Self {


This is very cool!

wjones127 · 2022-10-26T04:31:29Z

rust/src/operations/mod.rs

+    ///
+    /// let ops = DeltaOps::new_in_memory();
+    /// ```
+    #[must_use]


This is new to me. Why #[must_use]?

I also only fairly recently learned about must use. The idea is, if you do not consume the result of this call, clippy (or even the compiler) will complain. which to me makes sense, since if not using the return it literally does nothing and should be removed.

I think futures for instance may also be must_use since if not consumed they also so nothing ...

wjones127 · 2022-10-26T04:34:27Z

rust/src/operations/transaction.rs

+    #[error("Tried committing existing table version: {0}")]
+    VersionAlreadyExists(DeltaDataTypeVersion),
+
+    /// Error returned when reading the delta log object failed.


Is this description accurate? It's duplicated below, and reading and serializing seem like opposites.

wjones127

Finished looking through, and few more comments mostly around clean up and follow up issues.

wjones127 · 2022-10-27T02:45:51Z

rust/src/operations/transaction.rs

+/// Low-level transaction API. Creates a temporary commit file. Once created,
+/// the transaction object could be dropped and the actual commit could be executed
+/// with `DeltaTable.try_commit_transaction`.
+async fn prepare_commit(


Is this meant to replace DeltaTransaction::PrepareCommit? Or is it different?

this is correct. I wanted to iterate a bit more on commits (as well as the writer) before adopting it in the main code paths. Next I wanted to finally look into the conflict resolution again, where I expect more changes to the commit behavior.

rust/src/operations/write.rs

wjones127 · 2022-10-27T03:01:14Z

rust/src/operations/write.rs

    }
 }

-impl ExecutionPlan for WriteCommand {
-    fn as_any(&self) -> &dyn Any {
+impl WriteBuilder {


Could we also allow passing the Parquet WriterProperties into this builder? I think users would want to be able to control the max_row_group_size and other options from there.

I went back and forth a bit on this. If we eventually want to have support for both parquet implementations, maybe we should not expose the specific option structs directly, on the other hand there is more options to consider...

rust/src/operations/write.rs

wjones127 · 2022-10-27T03:29:33Z

rust/src/operations/write.rs

+                if batches.is_empty() {
+                    Err(WriteError::MissingData)


Could we instead early return if there is no data?

I guess ... :). if we do we need to either make it a no-op or add some logic to handle the case when we do not have an explicitly defined table schema, in case we need to create the table.

rust/src/operations/writer.rs

wjones127 · 2022-10-27T03:58:39Z

rust/src/operations/write.rs

+            }?;
+
+            let plan = if let Some(plan) = this.input {
+                Ok(plan)


is the plan guaranteed to be partitioned correctly?

no, it is now ... or rather I did think about wrapping that, but this moved to the writer now ... I'll check more explicitly that we handle this correctly.

rust/src/operations/writer.rs

wjones127 · 2022-10-27T04:10:45Z

rust/src/storage/mod.rs

-            Ok(_) => Ok(true),
-            Err(ObjectStoreError::NotFound { .. }) => Ok(false),
-            Err(err) => Err(err),
+        // TODO We should really be using HEAD here, but this fails in windows tests


Have you created a ticket for this in object-store? I can look into this.

No I have not yet - Windows gives a permission denied error, and I wanted to investigate if that is a general / expected behavior. At least in principle there should not be any permission issues, since the tests run in a generated temp folder...

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

roeap · 2022-11-06T20:10:07Z

Cargo.toml

-[profile.dev]
-split-debuginfo = "unpacked"


Not entirely sure why why had this option. however it seems it is only stable on macOS, and was causing build issues for Windows after 1,65 was released, so I removed it.

Signed-off-by: Robert Pack <robstar.pack@gmail.com>

roeap · 2022-11-15T23:23:51Z

@wjones127 - sorry for letting this sit for so long. Did some minor tweaks, mainly related to so deprecations in latest chrono and resolved some conflicts with main.

could you re-approve? :)

wjones127

No worries. Thanks for coming back to it. I'm excited to get this merged :)

roeap force-pushed the operations branch from 504bfd5 to bf71dc2 Compare September 27, 2022 21:35

feat: rewrite operations

0d62767

roeap force-pushed the operations branch from bf71dc2 to 0d62767 Compare September 27, 2022 21:37

chore: cleanup

032d9a7

wjones127 reviewed Sep 28, 2022

View reviewed changes

roeap added 3 commits September 28, 2022 08:01

Merge branch 'main' into operations

edc9711

docs: add basic docstrings to some DeltaOps methods

f9a9d49

try fixing windows tests

7160697

feat: writer updates

fa775b6

roeap mentioned this pull request Oct 12, 2022

Datafusion table provider: issues with timestamp types #441

Closed

This was referenced Oct 17, 2022

build(deps): bump datafusion-common from 12.0.0 to 13.0.0 #888

Closed

build(deps): bump datafusion from 12.0.0 to 13.0.0 #887

Closed

roeap added 4 commits October 19, 2022 21:42

fix: migrate arrow-pyarrow conversions

57d643b

test: fix parquet2 build

e42743d

try using tempfile to fix failing windows tests

7cfcf5b

fix: use list to determine if delta location

cb6e214

roeap force-pushed the operations branch from 53449e4 to cb6e214 Compare October 19, 2022 22:16

roeap marked this pull request as ready for review October 25, 2022 07:24

roeap requested review from fvaleye, rtyler, houqp, xianwill and mosyp as code owners October 25, 2022 07:24

roeap mentioned this pull request Oct 25, 2022

Add TableProviderFactory and test for SQL to register tables dynamically at runtime #892

Merged

roeap requested a review from wjones127 October 25, 2022 07:57

wjones127 requested changes Oct 26, 2022

View reviewed changes

wjones127 requested changes Oct 27, 2022

View reviewed changes

roeap and others added 2 commits October 27, 2022 11:20

Apply suggestions from code review

47ac52c

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

revert option partition columns

a3361c1

wjones127 mentioned this pull request Oct 31, 2022

changes for bump to datafusion13/arrow24 #902

Closed

roeap added 5 commits November 3, 2022 19:28

PR comments

9f3727e

Merge branch 'main' into operations

bfc2b9b

chore: fix some 1.65 clippy errors

1294fee

chore: more clippy

ce37e4f

chore: remove unstable option

04d7584

roeap commented Nov 6, 2022

View reviewed changes

roeap mentioned this pull request Nov 6, 2022

Allow for overwriting in create operation. #917

Closed

roeap requested a review from wjones127 November 6, 2022 20:38

roeap added 2 commits November 6, 2022 21:42

fix: fail on create with overwrite

2181460

fix: unused import

6b30e97

roeap mentioned this pull request Nov 10, 2022

Implement restore command #837

Closed

wjones127 previously approved these changes Nov 11, 2022

View reviewed changes

fix: omit create overwrite test

200e1d7

Signed-off-by: Robert Pack <robstar.pack@gmail.com>

roeap dismissed wjones127’s stale review via 200e1d7 November 15, 2022 22:59

roeap added 2 commits November 16, 2022 00:09

Merge branch 'main' into operations

b88c42d

fix: don't use newly deprecated chrono functions in parquet2

6ebc8a8

wjones127 approved these changes Nov 16, 2022

View reviewed changes

roeap merged commit e72cdfe into delta-io:main Nov 16, 2022

roeap deleted the operations branch November 16, 2022 05:07

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: rewrite operations #852

feat: rewrite operations #852

roeap commented Sep 27, 2022

wjones127 Sep 28, 2022

wjones127 Sep 28, 2022

roeap Nov 6, 2022

wjones127 Sep 28, 2022

roeap Nov 6, 2022

wjones127 Sep 28, 2022

roeap Sep 28, 2022

roeap Sep 28, 2022

roeap Nov 6, 2022

wjones127 left a comment

wjones127 commented Sep 30, 2022

wjones127 left a comment

wjones127 Oct 26, 2022

wjones127 Oct 26, 2022

wjones127 Oct 26, 2022

roeap Oct 27, 2022

wjones127 Oct 26, 2022

wjones127 left a comment

wjones127 Oct 27, 2022

roeap Oct 27, 2022

wjones127 Oct 27, 2022

roeap Nov 3, 2022

wjones127 Oct 27, 2022

roeap Nov 3, 2022

wjones127 Oct 27, 2022

roeap Oct 27, 2022

wjones127 Oct 27, 2022

roeap Oct 27, 2022

roeap Nov 6, 2022

roeap commented Nov 15, 2022

wjones127 left a comment

	/// THe main purpose of in-memory tables is for use in testing.
	/// The main purpose of in-memory tables is for use in testing.

		[profile.dev]
		split-debuginfo = "unpacked"

feat: rewrite operations #852

feat: rewrite operations #852

Conversation

roeap commented Sep 27, 2022

Description

Related Issue(s)

Documentation

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

wjones127 left a comment

Choose a reason for hiding this comment

wjones127 commented Sep 30, 2022

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

roeap commented Nov 15, 2022

wjones127 left a comment

Choose a reason for hiding this comment