-
Notifications
You must be signed in to change notification settings - Fork 416
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
12 changed files
with
237 additions
and
37 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
138 changes: 138 additions & 0 deletions
138
crates/deltalake-core/src/operations/transaction/protocol.rs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,138 @@ | ||
| use std::collections::HashSet; | ||
|
|
||
| use once_cell::sync::Lazy; | ||
|
|
||
| use super::TransactionError; | ||
| use crate::kernel::{ReaderFeatures, WriterFeatures}; | ||
| use crate::table::state::DeltaTableState; | ||
|
|
||
| static READER_V2: Lazy<HashSet<ReaderFeatures>> = Lazy::new(|| { | ||
| let mut set = HashSet::new(); | ||
| set.insert(ReaderFeatures::ColumnMapping); | ||
| set | ||
| }); | ||
|
|
||
| static WRITER_V2: Lazy<HashSet<WriterFeatures>> = Lazy::new(|| { | ||
| let mut set = HashSet::new(); | ||
| set.insert(WriterFeatures::AppendOnly); | ||
| set.insert(WriterFeatures::Invariants); | ||
| set | ||
| }); | ||
| static WRITER_V3: Lazy<HashSet<WriterFeatures>> = Lazy::new(|| { | ||
| let mut set = WRITER_V2.clone(); | ||
| set.insert(WriterFeatures::CheckConstraints); | ||
| set | ||
| }); | ||
| static WRITER_V4: Lazy<HashSet<WriterFeatures>> = Lazy::new(|| { | ||
| let mut set = WRITER_V3.clone(); | ||
| set.insert(WriterFeatures::ChangeDataFeed); | ||
| set.insert(WriterFeatures::GeneratedColumns); | ||
| set | ||
| }); | ||
| static WRITER_V5: Lazy<HashSet<WriterFeatures>> = Lazy::new(|| { | ||
| let mut set = WRITER_V4.clone(); | ||
| set.insert(WriterFeatures::ColumnMapping); | ||
| set | ||
| }); | ||
| static WRITER_V6: Lazy<HashSet<WriterFeatures>> = Lazy::new(|| { | ||
| let mut set = WRITER_V5.clone(); | ||
| set.insert(WriterFeatures::IdentityColumns); | ||
| set | ||
| }); | ||
|
|
||
| pub struct ProtocolChecker { | ||
| reader_features: HashSet<ReaderFeatures>, | ||
| writer_features: HashSet<WriterFeatures>, | ||
| } | ||
|
|
||
| impl ProtocolChecker { | ||
| /// Create a new protocol checker. | ||
| pub fn new( | ||
| reader_features: HashSet<ReaderFeatures>, | ||
| writer_features: HashSet<WriterFeatures>, | ||
| ) -> Self { | ||
| Self { | ||
| reader_features, | ||
| writer_features, | ||
| } | ||
| } | ||
|
|
||
| pub fn default_reader_version(&self) -> i32 { | ||
| 1 | ||
| } | ||
|
|
||
| pub fn default_writer_version(&self) -> i32 { | ||
| 1 | ||
| } | ||
|
|
||
| /// Check if delta-rs can read form the given delta table. | ||
| pub fn can_read_from(&self, snapshot: &DeltaTableState) -> Result<(), TransactionError> { | ||
| let required_features = match snapshot.min_reader_version() { | ||
| 0 | 1 => None, | ||
| 2 => Some(READER_V2.clone()), | ||
| _ => snapshot.reader_features().cloned(), | ||
| }; | ||
|
|
||
| if let Some(features) = required_features { | ||
| let diff: Vec<_> = features | ||
| .difference(&self.reader_features) | ||
| .cloned() | ||
| .collect(); | ||
| if diff.is_empty() { | ||
| Ok(()) | ||
| } else { | ||
| Err(TransactionError::UnsupportedReaderFeatures(diff)) | ||
| } | ||
| } else { | ||
| return Ok(()); | ||
| } | ||
| } | ||
|
|
||
| /// Check if delta-rs can write to the given delta table. | ||
| pub fn can_write_to(&self, snapshot: &DeltaTableState) -> Result<(), TransactionError> { | ||
| // NOTE: writers must always support all required reader features | ||
| self.can_read_from(snapshot)?; | ||
|
|
||
| let required_features = match snapshot.min_writer_version() { | ||
| 0 | 1 => None, | ||
| 2 => Some(WRITER_V2.clone()), | ||
| 3 => Some(WRITER_V3.clone()), | ||
| 4 => Some(WRITER_V4.clone()), | ||
| 5 => Some(WRITER_V5.clone()), | ||
| 6 => Some(WRITER_V6.clone()), | ||
| _ => snapshot.writer_features().cloned(), | ||
| }; | ||
|
|
||
| if let Some(features) = required_features { | ||
| let diff: Vec<_> = features | ||
| .difference(&self.writer_features) | ||
| .cloned() | ||
| .collect(); | ||
| if diff.is_empty() { | ||
| Ok(()) | ||
| } else { | ||
| Err(TransactionError::UnsupportedWriterFeatures(diff)) | ||
| } | ||
| } else { | ||
| // TODO: check if this is correct | ||
| Ok(()) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /// The global protocol checker instance to validate table versions and features. | ||
| pub static INSTANCE: Lazy<ProtocolChecker> = Lazy::new(|| { | ||
| let reader_features = HashSet::new(); | ||
| // reader_features.insert(ReaderFeatures::ColumnMapping); | ||
|
|
||
| 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::ChangeDataFeed); | ||
| // writer_features.insert(WriterFeatures::GeneratedColumns); | ||
| // writer_features.insert(WriterFeatures::ColumnMapping); | ||
| // writer_features.insert(WriterFeatures::IdentityColumns); | ||
|
|
||
| ProtocolChecker::new(reader_features, writer_features) | ||
| }); |
Oops, something went wrong.