feat: handle protocol compatibility (#1807)

# Description

In preparation for further improvements to protocol support, this PR
introduces a `ProtocolChecker` which validates that we can read form /
write to / commit to a specific table. So far everything is expressed in
table features in the hopes, that this keeps on giving correct
behaviours as we add more table features.

The existing support for append only is integrated and extended to
handle enablement logic according to the protocol.

---------

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

Author: roeap

crates/deltalake-core/Cargo.toml

@@ -130,6 +130,7 @@ tempfile = "3"
 tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
 utime = "0.3"
 hyper = { version = "0.14", features = ["server"] }
+criterion = "0.5"
 
 [features]
 azure = ["object_store/azure"]

crates/deltalake-core/benches/read_checkpoint.rs

@@ -1,6 +1,6 @@
 use criterion::{criterion_group, criterion_main, Criterion};
-use deltalake::table::state::DeltaTableState;
-use deltalake::DeltaTableConfig;
+use deltalake_core::table::state::DeltaTableState;
+use deltalake_core::DeltaTableConfig;
 use std::fs::File;
 use std::io::Read;
 

crates/deltalake-core/src/kernel/actions/types.rs

@@ -1,4 +1,5 @@
 use std::collections::{HashMap, HashSet};
+use std::fmt;
 use std::str::FromStr;
 // use std::io::{Cursor, Read};
 // use std::sync::Arc;
@@ -225,6 +226,24 @@ impl From<String> for ReaderFeatures {
     }
 }
 
+impl AsRef<str> for ReaderFeatures {
+    fn as_ref(&self) -> &str {
+        match self {
+            ReaderFeatures::ColumnMapping => "columnMapping",
+            ReaderFeatures::DeleteionVecotrs => "deletionVectors",
+            ReaderFeatures::TimestampWithoutTimezone => "timestampNtz",
+            ReaderFeatures::V2Checkpoint => "v2Checkpoint",
+            ReaderFeatures::Other(f) => f,
+        }
+    }
+}
+
+impl fmt::Display for ReaderFeatures {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.as_ref())
+    }
+}
+
 /// Features table writers can support as well as let users know
 /// what is supported
 #[derive(Serialize, Deserialize, Debug, Clone, Eq, PartialEq, Hash)]
@@ -303,6 +322,33 @@ impl From<String> for WriterFeatures {
     }
 }
 
+impl AsRef<str> for WriterFeatures {
+    fn as_ref(&self) -> &str {
+        match self {
+            WriterFeatures::AppendOnly => "appendOnly",
+            WriterFeatures::Invariants => "invariants",
+            WriterFeatures::CheckConstraints => "checkConstraints",
+            WriterFeatures::ChangeDataFeed => "changeDataFeed",
+            WriterFeatures::GeneratedColumns => "generatedColumns",
+            WriterFeatures::ColumnMapping => "columnMapping",
+            WriterFeatures::IdentityColumns => "identityColumns",
+            WriterFeatures::DeleteionVecotrs => "deletionVectors",
+            WriterFeatures::RowTracking => "rowTracking",
+            WriterFeatures::TimestampWithoutTimezone => "timestampNtz",
+            WriterFeatures::DomainMetadata => "domainMetadata",
+            WriterFeatures::V2Checkpoint => "v2Checkpoint",
+            WriterFeatures::IcebergCompatV1 => "icebergCompatV1",
+            WriterFeatures::Other(f) => f,
+        }
+    }
+}
+
+impl fmt::Display for WriterFeatures {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.as_ref())
+    }
+}
+
 #[cfg(all(not(feature = "parquet2"), feature = "parquet"))]
 impl From<&parquet::record::Field> for WriterFeatures {
     fn from(value: &parquet::record::Field) -> Self {

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

@@ -7,8 +7,7 @@ use std::sync::Arc;
 use futures::future::BoxFuture;
 use serde_json::{Map, Value};
 
-use super::transaction::commit;
-use super::{MAX_SUPPORTED_READER_VERSION, MAX_SUPPORTED_WRITER_VERSION};
+use super::transaction::{commit, PROTOCOL};
 use crate::errors::{DeltaResult, DeltaTableError};
 use crate::kernel::{Action, DataType, Metadata, Protocol, StructField, StructType};
 use crate::logstore::{LogStore, LogStoreRef};
@@ -245,8 +244,8 @@ impl CreateBuilder {
                 _ => unreachable!(),
             })
             .unwrap_or_else(|| Protocol {
-                min_reader_version: MAX_SUPPORTED_READER_VERSION,
-                min_writer_version: MAX_SUPPORTED_WRITER_VERSION,
+                min_reader_version: PROTOCOL.default_reader_version(),
+                min_writer_version: PROTOCOL.default_writer_version(),
                 writer_features: None,
                 reader_features: None,
             });
@@ -391,8 +390,14 @@ mod tests {
             .await
             .unwrap();
         assert_eq!(table.version(), 0);
-        assert_eq!(table.get_min_reader_version(), MAX_SUPPORTED_READER_VERSION);
-        assert_eq!(table.get_min_writer_version(), MAX_SUPPORTED_WRITER_VERSION);
+        assert_eq!(
+            table.get_min_reader_version(),
+            PROTOCOL.default_reader_version()
+        );
+        assert_eq!(
+            table.get_min_writer_version(),
+            PROTOCOL.default_writer_version()
+        );
         assert_eq!(table.schema().unwrap(), &schema);
 
         // check we can overwrite default settings via adding actions

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

@@ -34,6 +34,8 @@ use parquet::file::properties::WriterProperties;
 use serde::Serialize;
 use serde_json::Value;
 
+use super::datafusion_utils::Expression;
+use super::transaction::PROTOCOL;
 use crate::delta_datafusion::expr::fmt_expr_to_sql;
 use crate::delta_datafusion::{find_files, register_store, DeltaScanBuilder};
 use crate::errors::{DeltaResult, DeltaTableError};
@@ -44,8 +46,6 @@ use crate::protocol::DeltaOperation;
 use crate::table::state::DeltaTableState;
 use crate::DeltaTable;
 
-use super::datafusion_utils::Expression;
-
 /// Delete Records from the Delta Table.
 /// See this module's documentation for more information
 pub struct DeleteBuilder {
@@ -274,6 +274,8 @@ impl std::future::IntoFuture for DeleteBuilder {
         let mut this = self;
 
         Box::pin(async move {
+            PROTOCOL.can_write_to(&this.snapshot)?;
+
             let state = this.state.unwrap_or_else(|| {
                 let session = SessionContext::new();
 

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

@@ -6,6 +6,7 @@ use datafusion::physical_plan::coalesce_partitions::CoalescePartitionsExec;
 use datafusion::physical_plan::{ExecutionPlan, SendableRecordBatchStream};
 use futures::future::BoxFuture;
 
+use super::transaction::PROTOCOL;
 use crate::errors::{DeltaResult, DeltaTableError};
 use crate::logstore::LogStoreRef;
 use crate::table::state::DeltaTableState;
@@ -46,6 +47,8 @@ impl std::future::IntoFuture for LoadBuilder {
         let this = self;
 
         Box::pin(async move {
+            PROTOCOL.can_read_from(&this.snapshot)?;
+
             let table = DeltaTable::new_with_state(this.log_store, this.snapshot);
             let schema = table.state.arrow_schema()?;
             let projection = this

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

@@ -64,7 +64,7 @@ use serde::Serialize;
 use serde_json::Value;
 
 use super::datafusion_utils::{into_expr, maybe_into_expr, Expression};
-use super::transaction::commit;
+use super::transaction::{commit, PROTOCOL};
 use crate::delta_datafusion::expr::{fmt_expr_to_sql, parse_predicate_expression};
 use crate::delta_datafusion::{register_store, DeltaScanBuilder};
 use crate::kernel::{Action, Remove};
@@ -1208,6 +1208,8 @@ impl std::future::IntoFuture for MergeBuilder {
         let mut this = self;
 
         Box::pin(async move {
+            PROTOCOL.can_write_to(&this.snapshot)?;
+
             let state = this.state.unwrap_or_else(|| {
                 let session = SessionContext::new();
 

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

@@ -50,11 +50,6 @@ pub mod write;
 #[cfg(all(feature = "arrow", feature = "parquet"))]
 pub mod writer;
 
-/// Maximum supported writer version
-pub const MAX_SUPPORTED_WRITER_VERSION: i32 = 1;
-/// Maximum supported reader version
-pub const MAX_SUPPORTED_READER_VERSION: i32 = 1;
-
 /// High level interface for executing commands against a DeltaTable
 pub struct DeltaOps(pub DeltaTable);
 

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

@@ -38,7 +38,7 @@ use parquet::errors::ParquetError;
 use parquet::file::properties::WriterProperties;
 use serde::{Deserialize, Serialize};
 
-use super::transaction::commit;
+use super::transaction::{commit, PROTOCOL};
 use super::writer::{PartitionWriter, PartitionWriterConfig};
 use crate::errors::{DeltaResult, DeltaTableError};
 use crate::kernel::{Action, Remove};
@@ -260,6 +260,8 @@ impl<'a> std::future::IntoFuture for OptimizeBuilder<'a> {
         let this = self;
 
         Box::pin(async move {
+            PROTOCOL.can_write_to(&this.snapshot)?;
+
             let writer_properties = this.writer_properties.unwrap_or_else(|| {
                 WriterProperties::builder()
                     .set_compression(Compression::ZSTD(ZstdLevel::try_new(4).unwrap()))

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

@@ -245,7 +245,6 @@ async fn execute(
             datetime: datetime_to_restore.map(|time| -> i64 { time.timestamp_millis() }),
         },
         &actions,
-        &snapshot,
         None,
     )
     .await?;