feat: handle protocol compatibility

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?;