Implement Log Replay for Change Data Feed #540

OussamaSaoudi-db · 2024-11-25T22:44:16Z

What changes are proposed in this pull request?

This PR introduces the path for replaying the log for TableChanges and resolving cdc, add, and remove actions.

At the top level, we introduce TableChangesScan::scan_data, which gets TableChangesScanData used to read CDF. The stream of scan data requires a log replay.

To perform log replay, the table_changes::LogReplayScanner is introduced, which processes a single commit. It is responsible for two things:

Producing TableChangesScanData, which is made up of transformed EngineData, a selection vector, and a map remove_dvs: HashMap<String, DvInfo>. remove_dvs maps from a remove action's path to its deletion vector.
The replay scanner also performs schema, protocol, and table property validation to ensure that the Change Data Feed can be processed.

The LogReplayScanner performs two passes over the actions for each commit in try_new and into_scan_batches respectively.

To perform the operations above, two new visitors are added: PreparePhaseVisitor, and FileActionSelectionVisitor.

To test the changes, a new LocalMockTable struct is created for testing. This struct is used to write batches of actions into commits. This is used to verify that LogReplay produces correct output.

The physical schema is added to TableChangesScan.

How was this change tested?

The following cases are tested:

Valid metadata and protocol processing
Failure due to delta.enableChangeDataFeed not being enabled.
Failure due to incompatible schema
Simple add and remove case where there are no shared paths among the actions
A cdc action is present and all other actions must be filtered.
A remove and add action with the same path are resolved: The remove action is not selected, but it's registered in the remove_dv map. The add action must be selected.
Failure due to incompatible protocol update.
Correctly using default timestamp from the file modification time.
Data skipping works during log replay.
The following schema validation cases are tested:
Adding non-nullable column
adding nullable column
commit has wider type than cdf schema
type widening (will eventually be supported)
cdf column is nullable while the commit schema is non-nullable (will eventually be supported)
cdf schema and commit schema have completely incompatible types
cdf schema has an extra nullable column.

codecov · 2024-11-25T22:48:37Z

Codecov Report

Attention: Patch coverage is 93.50797% with 57 lines in your changes missing coverage. Please review.

Project coverage is 82.33%. Comparing base (eb95c5b) to head (c1d6103).
Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
kernel/src/table_changes/log_replay.rs	86.41%	9 Missing and 24 partials ⚠️
kernel/src/table_changes/scan.rs	7.69%	12 Missing ⚠️
kernel/src/table_changes/log_replay/tests.rs	98.92%	0 Missing and 6 partials ⚠️
kernel/src/table_changes/mod.rs	82.14%	4 Missing and 1 partial ⚠️
ffi/src/error.rs	0.00%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #540      +/-   ##
==========================================
+ Coverage   81.63%   82.33%   +0.69%     
==========================================
  Files          68       71       +3     
  Lines       14874    15734     +860     
  Branches    14874    15734     +860     
==========================================
+ Hits        12142    12954     +812     
- Misses       2145     2162      +17     
- Partials      587      618      +31

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

nicklan

quick initial pass, will look more soon

kernel/src/table_changes/log_replay.rs

nicklan · 2024-11-25T23:38:28Z

kernel/src/table_changes/log_replay.rs

+                dir,
+            }
+        }
+        pub(crate) fn commit(&mut self, actions: &[Action]) {


nice, this is cool. Can we maybe move it into the test_utils crate? We have some similar but more limited functionality in there. If we put it in the same place we could probably replace a lot of what's there with some of this.

One issue is that I have Serialize for actions as cfg_attr(test, ...). Should I try to make it dev-visible?

Tentatively moved to dev-visibility.

I'm also punting on replacing TestAction since that changes a lot of stuff in read.rs.

https://users.rust-lang.org/t/problems-sharing-test-code-traits-via-a-testutils-crate/53305/2

Test-utils may not be able to support this fully 🤔

kernel/src/table_changes/log_replay.rs

scovich

Good start -- I don't see anything worrisome

kernel/src/table_changes/log_replay.rs

scovich · 2024-11-26T01:26:13Z

kernel/src/table_changes/log_replay.rs

+            // we start our selection vector based on what was filtered. we will add to this vector
+            // below if a file has been removed
+            let selection_vector = match filter_vector {
+                Some(ref filter_vector) => filter_vector.clone(),


Why do we need to clone it? I don't see any later uses...

let selection_vector = self .filter .as_ref() .map(|filter| filter.apply(actions.as_ref())) .transpose()? .unwrap_or_else(|| vec![true, actions.len()]);

(again below -- maybe we should find a way to consolidate that very similar code?)

Yeah, consolidating makes a lot of sense. This feels like it should be part of the DataSkippingFilter, since it's similar code to scan. What do you think?

Are you suggesting to add an associated method that encapsulates the above logic, like this?

impl DataSkippingFilter { fn compute_selection_vector(filter: &Option<Self>, data: &dyn EngineData) -> DeltaResult<Vec<bool>> { .. }

If so, that seems like a good idea to me!

Yep! Shoulda made myself clearer lol. I can take that on in a followup PR.

Update: We only do data skipping once during CDF replay now... but this still seems useful since CDF and normal replay both would use it?

Yep both of them use it. It'd certainly be nice to have

follow-up issue?

kernel/src/table_changes/log_replay.rs

OussamaSaoudi-db · 2024-11-26T21:41:10Z

kernel/src/table_changes/scan.rs

+        &self,
+        engine: &dyn Engine,
+    ) -> DeltaResult<impl Iterator<Item = DeltaResult<TableChangesScanData>>> {
+        let commits = self


Just wanted to make sure that copying ParsedLogPath is okay. Passing a &[&ParsedLogPath] makes it so the LogReplayScanner holds it, and the iterator now has a lifetime tied to table_changes.log_segment.

Copying seems fine? The main worry is if we ever might need to catch up on thousands/millions of versions because streaming fell behind, in which case we'd be materializing them twice instead of "only" once?

It's fairly small, so doing it once per query is likely fine

kernel/src/table_changes/log_replay.rs

nicklan

looking pretty good. one structural comment and a few nits

kernel/src/scan/data_skipping.rs

kernel/src/table_changes/log_replay.rs

nicklan · 2024-12-02T22:21:18Z

kernel/src/table_changes/scan.rs

+        &self,
+        engine: &dyn Engine,
+    ) -> DeltaResult<impl Iterator<Item = DeltaResult<TableChangesScanData>>> {
+        let commits = self


It's fairly small, so doing it once per query is likely fine

kernel/src/table_changes/log_replay.rs

scovich · 2024-12-02T19:12:19Z

kernel/src/error.rs

+        expected: &StructType,
+        actual: &StructType,
+    ) -> Self {
+        Self::ChangeDataFeedIncompatibleSchema(format!("{:?}", expected), format!("{:?}", actual))


If we want those to be readable, we probably need to use alt format?

Suggested change

Self::ChangeDataFeedIncompatibleSchema(format!("{:?}", expected), format!("{:?}", actual))

Self::ChangeDataFeedIncompatibleSchema(format!("{expected:#?}"), format!("{actual:#?}"))

... and we should probably also filter out the compatible bits of the schema (and maybe even keep just one counter-example, so people don't drown in the error message). Nicer formatting can be a follow-up item tho.

scovich · 2024-12-02T19:21:12Z

kernel/src/table_changes/scan.rs

+        &self,
+        engine: &dyn Engine,
+    ) -> DeltaResult<impl Iterator<Item = DeltaResult<TableChangesScanData>>> {
+        let commits = self


Copying seems fine? The main worry is if we ever might need to catch up on thousands/millions of versions because streaming fell behind, in which case we'd be materializing them twice instead of "only" once?

kernel/src/utils.rs

kernel/src/table_changes/log_replay.rs

scovich · 2024-12-02T22:32:03Z

kernel/src/table_changes/log_replay.rs

+///     - Find the timestamp from a `CommitInfo` action if it exists. These are generated when
+///       In-commit timestamps is enabled.


Do we actually support ICT yet in kernel? It changes how time travel resolution works, for example?

Doesn't seem like we have it enabled, I don't see it in TableProperties.

It changes how time travel resolution works, for example?

Yeah anywhere there are timestamps, we need to use the ICT if it's enabled. This could be for Snapshot timetravel, or specifying a CDF with a timestamp. For CDF _commit_timestamp field, the behaviour is to use ICT if it's there.

That means CDF will anyway fail protocol checks if the table has ICT, no?
If we anyway don't support ICT yet, maybe we don't need to worry about timestamps here quite yet?

Now updated to ignore ICT

Note that we can't error out on ICT being present because it's not in TableProperties yet. @zachschuermann this will probably go into the TableProperties/TableConfiguration update

I have a PR up right now to add ICT as table property. Let's separate the two. I'd prefer to have a follow-up issue here that's specifically for 'enable ICT for CDF reads'

zachschuermann

looking great - generally advocating for keeping more things private PTAL at all those pub spots carefully. test review TODO

kernel/src/error.rs

kernel/src/scan/log_replay.rs

kernel/src/scan/mod.rs

kernel/src/utils.rs

kernel/src/table_changes/log_replay.rs

kernel/src/table_changes/scan.rs

kernel/src/table_changes/log_replay.rs

zachschuermann · 2024-12-03T18:00:25Z

kernel/src/table_changes/log_replay.rs

+        if let Some(protocol) = visitor.protocol {
+            protocol.ensure_read_supported()?;
+        }


this is saying that if a protocol action is present during iteration we check it's supported (I read the comment above - just trying to think through this)

do we need to check for table features at all? I guess CDF doesn't need to? (and i see the table properties check below)

CDF is absolutely impacted by table features -- DV and ICT at a minimum (and soon: CC, Variant, Geo, etc). Even CDF itself is actually a table feature, and we could absolutely have new table features relating to it (DML team has been dreaming of a "zero-cost CDF" table feature based on row tracking, for example)

this is saying that if a protocol action is present during iteration we check it's supported

Correct

do we need to check for table features at all?

No specific reader or writer features.

CDF is absolutely impacted by table features -- DV and ICT at a minimum

Is there any specific behaviour that I should be checking for this? I don't see anything explicit in the delta-spark code for this

It's already implemented... just without checking for the table feature:

ICT - get timestamp from commit info

DV - use deletion vectors

ICT table property checking is currently unsupported, and @zachschuermann is working on introducing it

We actually always have to handle deletion vectors, even when it's not specified in TableProperties. From the protocol:

Readers must read the table considering the existence of DVs, even when the delta.enableDeletionVectors table property is not set.

Edit:
wrong link. Here's the right one

That's only true if the table feature is present in protocol. If the feature is not present in protocol then there should not be any DV and any that seem to be present are probably garbage.

scovich

Getting close. Main outstanding issues IMO are:

How do table features interact with CDF code?
Don't we need to pay attention to dataChange flag on file actions?

kernel/src/table_changes/log_replay.rs

scovich · 2024-12-03T18:50:14Z

kernel/src/table_changes/log_replay.rs

+        if let Some(protocol) = visitor.protocol {
+            protocol.ensure_read_supported()?;
+        }


CDF is absolutely impacted by table features -- DV and ICT at a minimum (and soon: CC, Variant, Geo, etc). Even CDF itself is actually a table feature, and we could absolutely have new table features relating to it (DML team has been dreaming of a "zero-cost CDF" table feature based on row tracking, for example)

kernel/src/table_changes/log_replay.rs

kernel/src/table_changes/scan.rs

scovich · 2024-12-03T20:47:28Z

kernel/src/table_changes/log_replay.rs

+            );
+            let table_properties = TableProperties::from(configuration);
+            require!(
+                table_properties.enable_change_data_feed.unwrap_or(false),


@zachschuermann here's another case of cross-validation we need to add -- is it legal for the table property to exist if the table doesn't even include CDF in its protocol?

yea ideally all of this will move to a new TableConfiguration like we talked about - I'm happy to tackle that soon as i get some time in the next few days

scovich · 2024-12-03T20:58:15Z

kernel/src/table_changes/log_replay.rs

+///     - Otherwise, select `add` and `remove` actions. Note that only `remove` actions that do not
+///       share a path with an `add` action are selected.


Don't we also need to ignore non-dataChange actions? I don't see any code for that?

ooohhh this is a good question! I had forgotten about this, thank you! also note we do have to be careful not to ignore all non-dataChange actions because I think cdc actions (which we need) always (usually?) have dataChange=false.

Right... I meant add and remove (forgot that cdc actions are also considered "file actions")

Hmm correct me if I'm wrong, but this should be covered already in the FileActionSelectionVisitor

if getters[0].get_str(i, "add.path")?.is_some() { self.selection_vector[i] = !self.has_cdc_action; } else if let Some(path) = getters[1].get_str(i, "remove.path")? { self.selection_vector[i] = !self.has_cdc_action && !self.remove_dvs.contains_key(path) } else { self.selection_vector[i] = getters[2].get_str(i, "cdc.path")?.is_some() };

the final case sets to true if it is a cdc action. Otherwise all other actions are set to false.

From the spec for dataChange field:

When false the logical file must already be present in the table or the records in the added file must be contained in one or more remove actions in the same version

The missing bit would be something like this, I think?

if getters[0].get_str(i, "add.path")?.is_some() { let isDataChange = getters[XXX].get("add.dataChange"); self.selection_vector[i] = isDataChange && !self.has_cdc_action;

(similar story for remove actions)

Let's add a test for this as well :)

test got added I believe, right @OussamaSaoudi-db? (I had the same comment in test file I think)

kernel/src/table_changes/log_replay/tests.rs

nicklan

looking good. just re-request me when the remaining comments are addressed

kernel/src/table_changes/log_replay.rs

OussamaSaoudi-db · 2024-12-03T23:02:15Z

@scovich

How do table features interact with CDF code?

That's a good question. At the moment, I'm not sure how certain features should interact with CDF. I suppose a simple approach is to fail on any changes to table features.

Don't we need to pay attention to dataChange flag on file actions?

Also a good point. Looking at delta-spark, they tfilter out actions that have dataChange = false.

            val addActions = actions.collect { case a: AddFile if a.dataChange => a }
            val removeActions = actions.collect { case r: RemoveFile if r.dataChange => r }

I can go ahead and add that filtering in the selection visitor.

OussamaSaoudi-db · 2024-12-04T00:23:12Z

@scovich update regarding table features: I spoke to @zachschuermann And we're deciding to do some extra checks for protocol and table features. We'll start off being extra restrictive with the table features/properties we support.

kernel/src/table_changes/mod.rs

…n prepare phase

…it is true

github-actions bot assigned OussamaSaoudi-db Nov 25, 2024

OussamaSaoudi-db marked this pull request as draft November 25, 2024 22:44

OussamaSaoudi-db changed the title ~~Implement Log Replay for Change Data Feedl~~ [WIP] Implement Log Replay for Change Data Feedl Nov 25, 2024

github-actions bot added the breaking-change Change that will require a version bump label Nov 25, 2024

nicklan reviewed Nov 25, 2024

View reviewed changes

scovich reviewed Nov 26, 2024

View reviewed changes

OussamaSaoudi-db force-pushed the log_replay_2 branch from 2348f7e to c902995 Compare November 26, 2024 02:41

OussamaSaoudi-db requested review from nicklan, scovich and zachschuermann November 26, 2024 04:05

zachschuermann changed the title ~~[WIP] Implement Log Replay for Change Data Feedl~~ [WIP] Implement Log Replay for Change Data Feed Nov 26, 2024

OussamaSaoudi-db changed the title ~~[WIP] Implement Log Replay for Change Data Feed~~ Implement Log Replay for Change Data Feed Nov 26, 2024

OussamaSaoudi-db force-pushed the log_replay_2 branch from 3223a53 to ba76bb1 Compare November 26, 2024 19:05

OussamaSaoudi-db marked this pull request as ready for review November 26, 2024 19:08

OussamaSaoudi-db commented Nov 26, 2024

View reviewed changes

kernel/src/table_changes/log_replay.rs Outdated Show resolved Hide resolved

OussamaSaoudi-db commented Nov 26, 2024

View reviewed changes

kernel/src/table_changes/log_replay.rs Show resolved Hide resolved

OussamaSaoudi-db commented Nov 26, 2024

View reviewed changes

OussamaSaoudi-db commented Nov 29, 2024

View reviewed changes

kernel/src/table_changes/log_replay.rs Outdated Show resolved Hide resolved

nicklan reviewed Dec 2, 2024

View reviewed changes

scovich reviewed Dec 2, 2024

View reviewed changes

OussamaSaoudi-db requested review from scovich and nicklan December 3, 2024 17:57

zachschuermann reviewed Dec 3, 2024

View reviewed changes

scovich reviewed Dec 3, 2024

View reviewed changes

kernel/src/table_changes/log_replay/tests.rs Outdated Show resolved Hide resolved

nicklan reviewed Dec 3, 2024

View reviewed changes

kernel/src/table_changes/log_replay.rs Show resolved Hide resolved

zachschuermann reviewed Dec 4, 2024

View reviewed changes

kernel/src/table_changes/mod.rs Outdated Show resolved Hide resolved

OussamaSaoudi-db added 27 commits December 6, 2024 13:07

address more pr comments

91b9f92

add comments ,revert iterator change

3b2c3d3

Fix doc comment

3cf6955

Share the data skipping filter

bf0d566

PreparePhaseVisitor no longer has a new function

e9423c4

fix clippy error

49921ef

Add in more doc comments explaining data skipping filter is missing i…

549cbdc

…n prepare phase

Add data change filtering

e05cedb

add tests for failing table properties and protocol

6a8eceb

Remove ICT support

5731e63

Fix naming, fix clippy

8628165

Add link to issue for ICT support

f1b6ef9

Add documentation for DVs

28db41d

appease clippy for row visitor

8b4aadb

Add source for dvs

64e0a25

Address more pr comments

7fcc4dc

Add comment to type widening

16a549c

Implement has_cdc_file optimization to skip add and remove checks if …

e1d922b

…it is true

Shorten tests

5533389

remove scan file that snuck in

562d65f

Add doc comments, add an extra schema incompatibility test

3440577

Address pr comments

5fd3d1c

Change tests to use higher level api

d0eb811

Update comments

7108fd8

Add data skipping filter test

1e4e09a

Fix import issue

54382aa

Add unwrap so that test doesn't capture and check the wrong error

c1d6103

OussamaSaoudi-db force-pushed the log_replay_2 branch from 34782e1 to c1d6103 Compare December 6, 2024 21:07

OussamaSaoudi-db merged commit 3b456e4 into delta-io:main Dec 6, 2024
20 checks passed

OussamaSaoudi-db mentioned this pull request Dec 14, 2024

[tracking issue] Change Data Feed read path #440

Open

21 tasks

	Self::ChangeDataFeedIncompatibleSchema(format!("{:?}", expected), format!("{:?}", actual))
	Self::ChangeDataFeedIncompatibleSchema(format!("{expected:#?}"), format!("{actual:#?}"))

		/// - Find the timestamp from a `CommitInfo` action if it exists. These are generated when
		/// In-commit timestamps is enabled.

		/// - Otherwise, select `add` and `remove` actions. Note that only `remove` actions that do not
		/// share a path with an `add` action are selected.

Implement Log Replay for Change Data Feed #540

Implement Log Replay for Change Data Feed #540

Conversation

OussamaSaoudi-db commented Nov 25, 2024 • edited Loading

What changes are proposed in this pull request?

How was this change tested?

codecov bot commented Nov 25, 2024 • edited Loading

Codecov Report

nicklan 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

scovich left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

scovich Dec 2, 2024 • edited Loading

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

nicklan 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

zachschuermann 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

OussamaSaoudi-db Dec 4, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

scovich 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

scovich Dec 3, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nicklan left a comment

Choose a reason for hiding this comment

OussamaSaoudi-db commented Dec 3, 2024

OussamaSaoudi-db commented Dec 4, 2024

OussamaSaoudi-db commented Nov 25, 2024 •

edited

Loading

codecov bot commented Nov 25, 2024 •

edited

Loading

scovich Dec 2, 2024 •

edited

Loading

OussamaSaoudi-db Dec 4, 2024 •

edited

Loading

scovich Dec 3, 2024 •

edited

Loading