feat: check constraints #1915

hntd187 · 2023-11-26T21:46:20Z

Description

This PR adds CHECK constraints on delta tables. I still have some outstanding work to do on this, but this is my working draft PR for this feature.

let constraint = table.add_constraint().with_constraint("id", "id < 100");
constraint.await?;

Related Issue(s)

#1881

Documentation

github-actions · 2023-11-26T21:46:45Z

ACTION NEEDED

delta-rs follows the Conventional Commits specification for release automation.

The PR title and description are used as the merge commit message. Please update your PR title and description to match the specification.

ion-elgreco · 2023-11-26T21:58:13Z

@hntd187 does the delta protocol allow to add constraints during create operation?

hntd187 · 2023-11-26T22:08:35Z

@hntd187 does the delta protocol allow to add constraints during create operation?

I'm not sure, the documentation only seems to add them post creation.

ion-elgreco · 2023-11-26T22:13:17Z

Hmm, perhaps we can have that then during create actions. It just adds the constraints as additional in the config, right?

hntd187 · 2023-11-26T23:04:46Z

Yeah, do an initial create, then just do the alter to add the constraint should be easy since the table is empty.

…traint

roeap · 2023-12-07T21:33:28Z

@hntd187 - could we reduce the number of files in the example table to only contain a single add action per commit (unless we need more for the case), to keep entropy down.

roeap

Thanks for looking into this @hntd187 - left a few comments.

Would it make sense to also make sure we check this on the python side?

roeap · 2023-12-07T21:35:29Z

crates/deltalake-core/src/delta_datafusion/mod.rs

    invariants: Vec<Invariant>,
    ctx: SessionContext,
 }

 impl DeltaDataChecker {
+    /// Create a new DeltaDataChecker with a specified set of invariants
+    pub fn with_invariants(invariants: Vec<Invariant>) -> Self {


we usually reserve the with_* style syntax for builder like objects (i.e. when mutate an existing object) and new_* when having something constructor like.

Here new_with_invariants might work

Done for both

roeap · 2023-12-07T21:39:01Z

crates/deltalake-core/src/delta_datafusion/mod.rs

+        let metadata = snapshot.metadata();
+
+        let invariants = metadata
+            .and_then(|meta| meta.schema.get_invariants().ok())


Just had a quick look, but i think get_invariants only errors when there are invariants defined, but in an illegal format. This is something we should return as an error.

This was copied from the place it originally was prior to invariant enforcement in the data checker. Perhaps since this is not related to check constraints and slightly alters what would happen normally can we have a followup PR on it?

sure. but maybe add a TODO?

roeap · 2023-12-07T21:40:00Z

crates/deltalake-core/src/delta_datafusion/mod.rs

            return Ok(());
        }

+        // if !self.ctx.table_exist("data")? {


remove commented code?

roeap · 2023-12-07T21:43:15Z

crates/deltalake-core/src/operations/transaction/protocol.rs

@@ -170,7 +170,7 @@ pub static INSTANCE: Lazy<ProtocolChecker> = Lazy::new(|| {
    let mut writer_features = HashSet::new();
    writer_features.insert(WriterFeatures::AppendOnly);
    writer_features.insert(WriterFeatures::Invariants);
-    // writer_features.insert(WriterFeatures::CheckConstraints);
+    writer_features.insert(WriterFeatures::CheckConstraints);


we missed that for invariants, but should fix that now :).

The Invariants and check features should only be added if the datafusion features is enabled.

@hntd187 @scarman-db - one last thing, could we hide Invariants and CheckConstraints behind the datafusion features, as they will not work otherwise ...

roeap · 2023-12-07T21:52:06Z

crates/deltalake-core/src/operations/constraints.rs

+            if this.name.is_none() {
+                return Err(DeltaTableError::Generic("No name provided".to_string()));
+            } else if this.expr.is_none() {
+                return Err(DeltaTableError::Generic(
+                    "No expression provided".to_string(),
+                ));
+            }
+            let name = this.name.unwrap();
+            let expr = match this.expr {
+                Some(Expression::String(s)) => s,
+                Some(Expression::DataFusion(e)) => e.to_string(),
+                None => unreachable!(),
+            };


Rather then doing the prior checks and then use unwrap / unreachable, could we just return the error in the match statement?

roeap · 2023-12-07T21:55:48Z

crates/deltalake-core/src/operations/constraints.rs

+            let files_to_check =
+                find_files(&this.snapshot, this.log_store.clone(), &state, None).await?;
+            let scan = DeltaScanBuilder::new(&this.snapshot, this.log_store.clone(), &state)
+                .with_files(&files_to_check.candidates)
+                .build()
+                .await?;


I think we can just create a scan without explicitly setting files, since it will just scan all files, right?

Correct, still learning about the code base. :-) I removed it

roeap · 2023-12-07T21:57:42Z

crates/deltalake-core/src/operations/constraints.rs

+            let records = collect_sendable_stream(record_stream).await?;
+
+            for batch in records {
+                checker.check_batch(&batch).await?;
+            }


I think it would be better to scan each batch as we materialize it, rather then requiring to keep the entire table in memory for a scan. Similar to what we use the checker for writing data.

I don't have an execution plan here from what I can tell I need that to split it all into tasks similar to what update does. Any suggestion on what to do there?

Should look similar to what we do when we write the stream to disk - sans the writing stuff :)

delta-rs/crates/deltalake-core/src/operations/write.rs

Lines 324 to 351 in e54996a

let mut tasks = vec![];

for i in 0..plan.output_partitioning().partition_count() {

let inner_plan = plan.clone();

let inner_schema = schema.clone();

let task_ctx = Arc::new(TaskContext::from(&state));

let config = WriterConfig::new(

inner_schema.clone(),

partition_columns.clone(),

writer_properties.clone(),

target_file_size,

write_batch_size,

);

let mut writer = DeltaWriter::new(object_store.clone(), config);

let checker_stream = checker.clone();

let mut stream = inner_plan.execute(i, task_ctx)?;

let handle: tokio::task::JoinHandle<DeltaResult<Vec<Add>>> =

tokio::task::spawn(async move {

while let Some(maybe_batch) = stream.next().await {

let batch = maybe_batch?;

checker_stream.check_batch(&batch).await?;

let arr = cast_record_batch(&batch, inner_schema.clone(), safe_cast)?;

writer.write(&arr).await?;

}

writer.close().await

});

tasks.push(handle);

}

To avoid having to iterate through all the partitions in a loop, You could also wrap it in a CoalesceExec, where you would always end up with a single partition.

I used exactly this to reflect the ask. :-)

roeap · 2023-12-07T22:00:18Z

crates/deltalake-core/src/operations/constraints.rs

+            let protocol = Protocol {
+                min_reader_version: if old_protocol.min_reader_version > 1 {
+                    old_protocol.min_reader_version
+                } else {
+                    1
+                },
+                min_writer_version: if old_protocol.min_reader_version > 3 {
+                    old_protocol.min_reader_version
+                } else {
+                    3
+                },


We have some mix-ups here when to use reader and writer version?

Also maybe add a comment that this would be the least required versions for the check feature.

Done and done

roeap · 2023-12-07T22:05:03Z

crates/deltalake-core/src/operations/constraints.rs

+            let commit_info = CommitInfo {
+                timestamp: Some(Utc::now().timestamp_millis()),
+                operation: Some("ADD CONSTRAINT".to_string()),
+                operation_parameters: Some(operational_parameters),
+                read_version: Some(this.snapshot.version()),
+                isolation_level: Some(IsolationLevel::Serializable),
+                is_blind_append: Some(false),
+                ..Default::default()
+            };


We usually create the CommitInfo via the assoaciated commit_info method on the DeltaOperation.

This doesn't seem better than what I'm doing here, I'd still have to set most of this myself. I took this info from the spark implementation when I created it. Is that okay?

Yeah its fine .. the creation of the commit info itself is not so critical.

The only really valuable thing over there is that we have a single place where we can define the string representation for an operation, in this case ADD CONSTRAINT. In erarlier version of delta-rs we hade some inconsistencies in that regard.

Come to think of it, most information - like read_version, isolation_level etc. we should probably set in a central place within the commit function anyhow. Especially since there are some cases where we can (and do) relax the isolation level for the actual commit. Should we maybe add an issue for that?

roeap · 2023-12-07T22:08:25Z

crates/deltalake-core/src/operations/constraints.rs

+            )
+            .await?;
+
+            Ok(DeltaTable::new_with_state(this.log_store, this.snapshot))


I don't think, this snapshot will reflect he latest table state including the updates we just did, as the commit function will not update the snapshot.

So the correct thing here @roeap is to do a merge on the states after the commit right?

Exactly, we should add the actions we just committed to the table also to the new state.

…o not need a physical table, addressed most (but not all) PR comments

roeap

Just a minor nit and a small ask, then one more sync with main and we are good to go!

roeap · 2023-12-11T17:18:36Z

crates/deltalake-core/src/operations/constraints.rs

+                Action::Protocol(protocol),
+            ];
+
+            let _version = commit(


just a minor nit, but since we use it these is no more need for the underscore.

roeap · 2023-12-11T17:20:01Z

crates/deltalake-core/src/operations/transaction/protocol.rs

@@ -170,7 +170,7 @@ pub static INSTANCE: Lazy<ProtocolChecker> = Lazy::new(|| {
    let mut writer_features = HashSet::new();
    writer_features.insert(WriterFeatures::AppendOnly);
    writer_features.insert(WriterFeatures::Invariants);
-    // writer_features.insert(WriterFeatures::CheckConstraints);
+    writer_features.insert(WriterFeatures::CheckConstraints);


@hntd187 @scarman-db - one last thing, could we hide Invariants and CheckConstraints behind the datafusion features, as they will not work otherwise ...

…ag because it caused many tests to fail

roeap

Thanks for sticking with it, great work @hntd187!

scarman-db and others added 4 commits November 24, 2023 12:00

feat: Add CHECK constraint functionality

aec1add

feat: check constraints

34a3180

Merge remote-tracking branch 'upstream/main' into check-constraints

0c44f10

feat: CHECK constraints

3192887

hntd187 requested review from wjones127, roeap and rtyler as code owners November 26, 2023 21:46

github-actions bot added binding/rust Issues for the Rust crate crate/core labels Nov 26, 2023

hntd187 marked this pull request as draft November 26, 2023 21:46

feat: CHECK constraints, added check on table for initial add of cons…

d6020e8

…traint

github-actions bot added binding/python Issues for the Python package crate/sql labels Dec 2, 2023

hntd187 and others added 4 commits December 3, 2023 18:01

feat: CHECK constraints, added check on table for initial add of cons…

916e1e7

…traint

Merge branch 'main' into check-constraints

e139f33

feat: check constraints

305bae9

feat: CHECK constraints, added check on table for initial add of cons…

8f913e7

…traint

hntd187 marked this pull request as ready for review December 6, 2023 03:15

hntd187 requested review from fvaleye and ion-elgreco as code owners December 6, 2023 03:15

Merge branch 'main' into check-constraints

c83c7cf

hntd187 changed the title ~~feat: Check constraints~~ feat: check constraints Dec 6, 2023

hntd187 added 2 commits December 5, 2023 22:32

feat: CHECK constraints, added check on table for initial add of cons…

fba0b50

…traint

feat: CHECK constraints addressing a lint

811a6d4

roeap reviewed Dec 7, 2023

View reviewed changes

scarman-db and others added 3 commits December 8, 2023 17:49

feat: check constraints, removed physical test table, rewrote tests t…

e906f3c

…o not need a physical table, addressed most (but not all) PR comments

Merge branch 'main' into check-constraints

b475f29

feat: check constraints, fix a lint

c5e23c1

roeap reviewed Dec 11, 2023

View reviewed changes

hntd187 and others added 4 commits December 13, 2023 14:24

Merge branch 'main' into check-constraints

9d88746

feat: CHECK constraints addressing a lint

1efd1fe

feat: CHECK constraints, back out limiting features behind feature fl…

e4f59a3

…ag because it caused many tests to fail

Merge branch 'main' into check-constraints

62894a0

roeap approved these changes Dec 14, 2023

View reviewed changes

roeap merged commit 2d65e72 into delta-io:main Dec 14, 2023
24 checks passed

	let mut tasks = vec![];
	for i in 0..plan.output_partitioning().partition_count() {
	let inner_plan = plan.clone();
	let inner_schema = schema.clone();
	let task_ctx = Arc::new(TaskContext::from(&state));
	let config = WriterConfig::new(
	inner_schema.clone(),
	partition_columns.clone(),
	writer_properties.clone(),
	target_file_size,
	write_batch_size,
	);
	let mut writer = DeltaWriter::new(object_store.clone(), config);
	let checker_stream = checker.clone();
	let mut stream = inner_plan.execute(i, task_ctx)?;
	let handle: tokio::task::JoinHandle<DeltaResult<Vec<Add>>> =
	tokio::task::spawn(async move {
	while let Some(maybe_batch) = stream.next().await {
	let batch = maybe_batch?;
	checker_stream.check_batch(&batch).await?;
	let arr = cast_record_batch(&batch, inner_schema.clone(), safe_cast)?;
	writer.write(&arr).await?;
	}
	writer.close().await
	});

	tasks.push(handle);
	}

feat: check constraints #1915

feat: check constraints #1915

Conversation

hntd187 commented Nov 26, 2023

Description

Related Issue(s)

Documentation

github-actions bot commented Nov 26, 2023

ion-elgreco commented Nov 26, 2023

hntd187 commented Nov 26, 2023

ion-elgreco commented Nov 26, 2023

hntd187 commented Nov 26, 2023

roeap commented Dec 7, 2023

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

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

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

roeap left a comment

Choose a reason for hiding this comment