feat: implement update operation #1390

Blajda · 2023-05-24T02:51:49Z

Description

Users can now update data that matches a predicate.

This operation should be encouraged over the replace write operation since update determines which values require rewriting based on the supplied predicate.

Related Issue(s)

closes Support Update operation #1126

Blajda · 2023-06-06T01:29:55Z

This implementation follows the same structure as the delete command. find_files is used to determine which files contains records the satisfy the predicate. Once the files are identified 3 Datafusion projections are performed to first mark if the record needs to be updated, second add internal columns calculated based on if the column is to be updated, and finally renames the internal columns to their original names.

Something different from previous operations is that a new ExecutionPlan implementation is created to expose a count of how many records were updated. I wanted to avoid creating this count in a scan/loop and was able to take advantage of null counts.

wjones127

This is impressive. I appreciate the variety of test cases :)

Had various questions and corrections throughout. I'll be excited to release this.

rust/src/delta_datafusion.rs

wjones127 · 2023-06-06T02:41:56Z

rust/src/delta_datafusion.rs

+    let partitions = limit.output_partitioning().partition_count();
+    let mut tasks = Vec::with_capacity(partitions);
+
+    for i in 0..partitions {


How do we limit the maximum concurrency? If I have 10,000 partitions, will this try to process all of them at the same time?

One way to limit the concurrency might be something like:

let partition_tasks = futures::iter(0..partitions) .map(|part_i| futures::future::ready(Ok(limit.execute(i, task_ctx.clone())) )) .try_flatten_unordered(max_concurrent_tasks);

wjones127 · 2023-06-06T02:47:05Z

rust/src/delta_datafusion.rs

+    let mut tasks = Vec::with_capacity(partitions);
+
+    for i in 0..partitions {
+        let stream = limit.execute(i, task_ctx.clone())?;


Where does this plan execute? in the current thread? If so, we might want to wrap these in spawn_blocking() instead, so they can be sent to execute across multiple threads.

I changed this area to push parallelism concerns onto Datafusion. I instead call collect and then process each batch for the path that was discovered. It cleans it up quite a bit

rust/src/delta_datafusion.rs

rust/src/operations/update.rs

wjones127 · 2023-06-06T03:16:29Z

rust/src/operations/update.rs

+    // Do not make a commit when there are zero updates to the state
+    if !actions.is_empty() {


Shouldn't we have eliminating this possibility when finding files?

Yes good point. I made a change to return early when find files determines zero candidates. Added a test to ensure the metrics are correct too.

wjones127 · 2023-06-06T03:18:06Z

rust/src/operations/update.rs

+                let array = batch.column_by_name("__delta_rs_update_predicate").unwrap();
+                let copied_rows = array.null_count();
+                let num_updated = array.len() - copied_rows;
+                let c1 = MetricBuilder::new(&self.metrics).global_counter("num_updated_rows");
+                c1.add(num_updated);
+
+                let c2 = MetricBuilder::new(&self.metrics).global_counter("num_copied_rows");
+                c2.add(copied_rows);
+                Some(Ok(batch))


This is cool!

Indeed it is!

Thanks it took me a bit to arrive to this solution but I'm glad ended up this simple.

wjones127 · 2023-06-06T03:19:12Z

rust/src/operations/update.rs

+    // Take advantage of how null counts are tracked in arrow arrays use the
+    // null count to track how many records do NOT statisfy the predicate.  The
+    // count is then exposed through the metrics through the `UpdateCountExec`
+    // execution plan


wjones127 · 2023-06-06T03:23:35Z

rust/src/operations/update.rs

+    #[tokio::test]
+    async fn test_str_expressions() {}


This a TODO?

Yeah my bad. I wanted to add a test to demonstrate that a str can be used for the predicate and update expression but it seemed like a test with little value.

I've updated the null tests to that.
FYI using strings for expressions will require additional work in future PRs.
If you have a predicate like value < 2 or value > 2 DataFusion will return an error about being unable to compare int32 to int64

If you have a predicate like value < 2 or value > 2 DataFusion will return an error about being unable to compare int32 to int64

I stumbled across this one as well, not sure if this is maybe even a datafusion bug, since the expression parser seems to ignore the information in the schema passed to it.

Co-authored-by: Will Jones <[email protected]>

roeap

Impressive work!

Left some minor comments, and there are some unwraps floating around that we maybe can have a look at if we can avoid them...

roeap · 2023-06-07T05:41:15Z

rust/src/delta_datafusion.rs

+        }
+        let array = batch
+            .column_by_name(PATH_COLUMN)
+            .unwrap()


nit: can we get rid of this unwrap, or add a comment why its safe?

I think now we can just use ? :)

😞 Sorry about that.
I factored that entire section into function since two places had the logic should be cleaner now.

rust/src/delta_datafusion.rs

rust/src/operations/update.rs

roeap · 2023-06-07T06:09:29Z

rust/src/operations/update.rs

+                let array = batch.column_by_name("__delta_rs_update_predicate").unwrap();
+                let copied_rows = array.null_count();
+                let num_updated = array.len() - copied_rows;
+                let c1 = MetricBuilder::new(&self.metrics).global_counter("num_updated_rows");
+                c1.add(num_updated);
+
+                let c2 = MetricBuilder::new(&self.metrics).global_counter("num_copied_rows");
+                c2.add(copied_rows);
+                Some(Ok(batch))


Indeed it is!

roeap

Impressive work!

Left some minor comments, and there are some unwraps floating around that we maybe can have a look at if we can avoid them...

Co-authored-by: Robert Pack <[email protected]>

roeap

Looking good! - I'll leave it open for @wjones127 to look at, since he did the bulk of the review.

roeap · 2023-06-08T07:32:51Z

rust/src/delta_datafusion.rs

+        }
+        let array = batch
+            .column_by_name(PATH_COLUMN)
+            .unwrap()


I think now we can just use ? :)

wjones127

Almost there! I have a few performance-related suggestions, and one more test case I think we want. After that, I think this is good to go :)

wjones127 · 2023-06-09T02:48:07Z

rust/src/delta_datafusion.rs

+    // Given RecordBatches that contains `__delta_rs_path` perform a hash join
+    // with actions to obtain original add actions
+
+    let mut files = Vec::new();


We should know the size ahead of time:

Suggested change

let mut files = Vec::new();

let mut files = Vec::with_capacity(batches.iter().map(|batch| batch.num_rows()).sum());

wjones127 · 2023-06-09T02:49:16Z

rust/src/delta_datafusion.rs

+            match actions.remove(path) {
+                Some(action) => files.push(action),


Why mutate the hashmap? isn't the path already guaranteed to be unique?

Yes the current implementation does guarantee that paths are unique but I want to make be defensive against any unexpected changes from Datafusion or future refactoring. My assumption is that removal from the hashmap is O(1) and that rust does not realloc the underlying array when active items go below a certain threshold.

wjones127 · 2023-06-09T03:18:45Z

rust/src/operations/update.rs

+    /// Time taken to execute the entire operation.
+    pub execution_time_ms: u128,
+    /// Time taken to scan the files for matches.
+    pub scan_time_ms: u128,


As before, micro (us) or milliseconds (ms)?

Also, u128 seems a little excessive. Even with microseconds, I think u64 gets you at least 10,000 years.

Suggested change

/// Time taken to execute the entire operation.

pub execution_time_ms: u128,

/// Time taken to scan the files for matches.

pub scan_time_ms: u128,

/// Time taken to execute the entire operation.

pub execution_time_ms: u64,

/// Time taken to scan the files for matches.

pub scan_time_ms: u64,

So the expected unit is actually ms since I'm just pulling the metrics here
I've changed the call sites to use as_millis() which returns a u128. Should I really explicitly downcast that to a u64?

Yes that will always be safe on the timescales we care about.

wjones127 · 2023-06-09T03:29:43Z

rust/src/operations/update.rs

+        }))
+    }
+
+    metrics.execution_time_ms = Instant::now().duration_since(exec_start).as_micros();


Did you mean milliseconds or microseconds? IMO milliseconds is plenty, but if you do microseconds, we should use the us abbreviation:

Suggested change

metrics.execution_time_ms = Instant::now().duration_since(exec_start).as_micros();

metrics.execution_time_us = Instant::now().duration_since(exec_start).as_micros();

rust/src/delta_datafusion.rs

wjones127 · 2023-06-09T04:07:12Z

rust/src/operations/update.rs

+    #[tokio::test]
+    async fn test_update_partitions() {


Since a partition-only predicate is a separate code path from one with a mix of partition and normal columns, I think we are missing some coverage on partition column handling in the case where we need to scan in find files.

Could you add either a separate test or just another part in this test where you update a partitioned table with a predicate that is on the partition column and a normal column?

Co-authored-by: Will Jones <[email protected]>

Blajda added 2 commits May 23, 2023 21:36

mvp update operation

c2e8324

rebase on main and fix expression schemas

0bac94b

github-actions bot added binding/rust Issues for the Rust crate rust labels May 24, 2023

Blajda added 5 commits May 27, 2023 23:29

error update on non-deterministic predicate

392e050

Factor out find files

1c48c32

remove duplicated test

be4cbb9

Obtain update and copy metrics for operation

cd00c6e

Add additional tests

ac6bf57

Blajda changed the title ~~feat: :WIP: implement update operation~~ feat: implement update operation Jun 2, 2023

Blajda added 3 commits June 5, 2023 19:25

Merge branch 'main' into update-op

68dffe0

Allow datafusion expressions or string expressions

c52aa09

Update tests

1188985

Blajda marked this pull request as ready for review June 6, 2023 01:29

Blajda requested review from houqp, xianwill, wjones127, fvaleye, roeap, rtyler and mosyp as code owners June 6, 2023 01:30

wjones127 requested changes Jun 6, 2023

View reviewed changes

Blajda and others added 6 commits June 6, 2023 19:07

Update rust/src/operations/update.rs

a036705

Co-authored-by: Will Jones <[email protected]>

Update rust/src/delta_datafusion.rs

5dc5bdf

Co-authored-by: Will Jones <[email protected]>

Update rust/src/operations/update.rs

8be224a

Co-authored-by: Will Jones <[email protected]>

Update rust/src/delta_datafusion.rs

1e54a71

Co-authored-by: Will Jones <[email protected]>

return early if no files need to be updated. clean up find files impl

3b685e6

Merge branch 'main' into update-op

538848f

roeap reviewed Jun 7, 2023

View reviewed changes

Blajda and others added 7 commits June 7, 2023 19:15

Update rust/src/operations/update.rs

65a26cd

Co-authored-by: Robert Pack <[email protected]>

Update rust/src/operations/update.rs

ffb3631

Co-authored-by: Robert Pack <[email protected]>

cleanup unwraps

540c43b

update comment

a001be0

update iter

eb0d867

Merge branch 'main' into update-op

f603e3d

update cast_options docs

fc30ae4

roeap previously approved these changes Jun 8, 2023

View reviewed changes

cleanup unwraps

0f74bc7

Blajda dismissed roeap’s stale review via 0f74bc7 June 9, 2023 00:41

wjones127 requested changes Jun 9, 2023

View reviewed changes

Update rust/src/delta_datafusion.rs

2f45a06

Co-authored-by: Will Jones <[email protected]>

Blajda marked this pull request as draft June 12, 2023 03:20

Blajda added 4 commits June 11, 2023 23:39

:WIP: resolve schema issues without projection

bc4377a

resolve schema issues with projection

80b27cf

resolve merge conflicts with main

9c7cd73

implement comments

73137cc

Blajda marked this pull request as ready for review June 13, 2023 02:44

store time metrics in u64

111c463

wjones127 approved these changes Jun 14, 2023

View reviewed changes

wjones127 merged commit 9730d59 into delta-io:main Jun 14, 2023

adhish20 mentioned this pull request Jul 28, 2023

Support updates GlareDB/glaredb#1351

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: implement update operation #1390

feat: implement update operation #1390

Blajda commented May 24, 2023

Blajda commented Jun 6, 2023

wjones127 left a comment

wjones127 Jun 6, 2023

wjones127 Jun 6, 2023

Blajda Jun 7, 2023

wjones127 Jun 6, 2023

Blajda Jun 7, 2023

wjones127 Jun 6, 2023

roeap Jun 7, 2023

Blajda Jun 8, 2023

wjones127 Jun 6, 2023

wjones127 Jun 6, 2023

Blajda Jun 7, 2023

roeap Jun 7, 2023

roeap left a comment

roeap Jun 7, 2023

roeap Jun 8, 2023

Blajda Jun 9, 2023

roeap Jun 7, 2023

roeap left a comment

roeap left a comment

roeap Jun 8, 2023

wjones127 left a comment

wjones127 Jun 9, 2023

wjones127 Jun 9, 2023

Blajda Jun 13, 2023

wjones127 Jun 9, 2023

Blajda Jun 13, 2023

wjones127 Jun 13, 2023

wjones127 Jun 9, 2023

wjones127 Jun 9, 2023

Blajda Jun 13, 2023

		// Do not make a commit when there are zero updates to the state
		if !actions.is_empty() {

	let mut files = Vec::new();
	let mut files = Vec::with_capacity(batches.iter().map(\|batch\| batch.num_rows()).sum());

		match actions.remove(path) {
		Some(action) => files.push(action),

	metrics.execution_time_ms = Instant::now().duration_since(exec_start).as_micros();
	metrics.execution_time_us = Instant::now().duration_since(exec_start).as_micros();

feat: implement update operation #1390

feat: implement update operation #1390

Conversation

Blajda commented May 24, 2023

Description

Related Issue(s)

Blajda commented Jun 6, 2023

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

Choose a reason for hiding this comment

roeap 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

roeap left a comment

Choose a reason for hiding this comment

roeap left a 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