refactor: re-organize top level modules

python/src/lib.rs

@@ -19,7 +19,6 @@ use deltalake::action::{
 use deltalake::arrow::compute::concat_batches;
 use deltalake::arrow::record_batch::RecordBatch;
 use deltalake::arrow::{self, datatypes::Schema as ArrowSchema};
-use deltalake::builder::DeltaTableBuilder;
 use deltalake::checkpoints::create_checkpoint;
 use deltalake::datafusion::prelude::SessionContext;
 use deltalake::delta_datafusion::DeltaDataChecker;
@@ -28,6 +27,7 @@ use deltalake::operations::optimize::OptimizeBuilder;
 use deltalake::operations::transaction::commit;
 use deltalake::operations::vacuum::VacuumBuilder;
 use deltalake::partitions::PartitionFilter;
+use deltalake::DeltaTableBuilder;
 use deltalake::{DeltaOps, Invariant, Schema};
 use pyo3::exceptions::{PyIOError, PyRuntimeError, PyValueError};
 use pyo3::prelude::*;

rust/benches/read_checkpoint.rs

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

rust/src/action/checkpoints.rs

@@ -18,10 +18,10 @@ use std::iter::Iterator;
 use std::ops::Add;
 
 use super::{Action, Add as AddAction, MetaData, Protocol, ProtocolError, Txn};
-use crate::delta_arrow::delta_log_schema_for_table;
+use crate::arrow_convert::delta_log_schema_for_table;
 use crate::schema::*;
 use crate::storage::DeltaObjectStore;
-use crate::table_state::DeltaTableState;
+use crate::table::state::DeltaTableState;
 use crate::{open_table_with_version, time_utils, CheckPoint, DeltaTable};
 
 type SchemaPath = Vec<String>;

rust/src/action/mod.rs

@@ -23,10 +23,10 @@ use std::borrow::Borrow;
 use std::collections::HashMap;
 use std::hash::{Hash, Hasher};
 
-use crate::delta_config::IsolationLevel;
+use crate::config::IsolationLevel;
 use crate::errors::DeltaResult;
 use crate::storage::ObjectStoreRef;
-use crate::{delta::CheckPoint, schema::*, DeltaTableMetaData};
+use crate::{schema::*, table::CheckPoint, DeltaTableMetaData};
 
 /// Error returned when an invalid Delta log action is encountered.
 #[allow(missing_docs)]

rust/src/delta_datafusion.rs

@@ -61,7 +61,7 @@ use crate::action::{self, Add};
 use crate::builder::ensure_table_uri;
 use crate::errors::{DeltaResult, DeltaTableError};
 use crate::storage::ObjectStoreRef;
-use crate::table_state::DeltaTableState;
+use crate::table::state::DeltaTableState;
 use crate::{open_table, open_table_with_storage_options, DeltaTable, Invariant, SchemaDataType};
 
 impl From<DeltaTableError> for DataFusionError {

rust/src/lib.rs

@@ -83,34 +83,24 @@ compile_error!(
 );
 
 pub mod action;
-pub mod builder;
 pub mod data_catalog;
-pub mod delta;
-pub mod delta_config;
 pub mod errors;
 pub mod operations;
-pub mod partitions;
 pub mod schema;
 pub mod storage;
-pub mod table_state;
+pub mod table;
 pub mod time_utils;
 
-#[cfg(all(feature = "arrow"))]
-pub mod table_state_arrow;
-
-#[cfg(all(feature = "arrow", feature = "parquet"))]
-pub mod delta_arrow;
 #[cfg(feature = "datafusion")]
 pub mod delta_datafusion;
 #[cfg(all(feature = "arrow", feature = "parquet"))]
 pub mod writer;
 
-pub use self::builder::*;
+pub use self::config::*;
 pub use self::data_catalog::{get_data_catalog, DataCatalog, DataCatalogError};
-pub use self::delta::*;
-pub use self::delta_config::*;
-pub use self::partitions::*;
+pub use self::schema::partitions::*;
 pub use self::schema::*;
+pub use self::table::*;
 pub use errors::*;
 pub use object_store::{path::Path, Error as ObjectStoreError, ObjectMeta, ObjectStore};
 pub use operations::DeltaOps;

rust/src/operations/create.rs

@@ -11,7 +11,7 @@ use super::transaction::commit;
 use super::{MAX_SUPPORTED_READER_VERSION, MAX_SUPPORTED_WRITER_VERSION};
 use crate::action::{Action, DeltaOperation, MetaData, Protocol, SaveMode};
 use crate::builder::ensure_table_uri;
-use crate::delta_config::DeltaConfigKey;
+use crate::config::DeltaConfigKey;
 use crate::errors::{DeltaResult, DeltaTableError};
 use crate::schema::{SchemaDataType, SchemaField, SchemaTypeStruct};
 use crate::storage::DeltaObjectStore;
@@ -322,7 +322,7 @@ impl std::future::IntoFuture for CreateBuilder {
 #[cfg(all(test, feature = "parquet"))]
 mod tests {
     use super::*;
-    use crate::delta_config::DeltaConfigKey;
+    use crate::config::DeltaConfigKey;
     use crate::operations::DeltaOps;
     use crate::writer::test_utils::get_delta_schema;
     use tempdir::TempDir;

rust/src/operations/delete.rs

@@ -61,7 +61,7 @@ use crate::errors::{DeltaResult, DeltaTableError};
 use crate::operations::transaction::commit;
 use crate::operations::write::write_execution_plan;
 use crate::storage::{DeltaObjectStore, ObjectStoreRef};
-use crate::table_state::DeltaTableState;
+use crate::table::state::DeltaTableState;
 use crate::DeltaTable;
 
 const PATH_COLUMN: &str = "__delta_rs_path";

rust/src/operations/filesystem_check.rs

@@ -28,7 +28,7 @@ use crate::action::{Action, Add, DeltaOperation, Remove};
 use crate::errors::{DeltaResult, DeltaTableError};
 use crate::operations::transaction::commit;
 use crate::storage::DeltaObjectStore;
-use crate::table_state::DeltaTableState;
+use crate::table::state::DeltaTableState;
 use crate::DeltaTable;
 
 /// Audit the Delta Table's active files with the underlying file system.

rust/src/operations/load.rs

@@ -8,7 +8,7 @@ use futures::future::BoxFuture;
 
 use crate::errors::{DeltaResult, DeltaTableError};
 use crate::storage::DeltaObjectStore;
-use crate::table_state::DeltaTableState;
+use crate::table::state::DeltaTableState;
 use crate::DeltaTable;
 
 #[derive(Debug, Clone)]

rust/src/operations/optimize.rs

@@ -46,7 +46,7 @@ use super::writer::{PartitionWriter, PartitionWriterConfig};
 use crate::action::{self, Action, DeltaOperation};
 use crate::errors::{DeltaResult, DeltaTableError};
 use crate::storage::ObjectStoreRef;
-use crate::table_state::DeltaTableState;
+use crate::table::state::DeltaTableState;
 use crate::writer::utils::arrow_schema_without_partitions;
 use crate::{crate_version, DeltaTable, ObjectMeta, PartitionFilter};
 

rust/src/operations/transaction/conflict_checker.rs

@@ -6,10 +6,10 @@ use object_store::ObjectStore;
 
 use super::CommitInfo;
 use crate::action::{Action, Add, DeltaOperation, MetaData, Protocol, Remove};
-use crate::delta_config::IsolationLevel;
+use crate::config::IsolationLevel;
 use crate::errors::{DeltaResult, DeltaTableError};
 use crate::storage::commit_uri_from_version;
-use crate::table_state::DeltaTableState;
+use crate::table::state::DeltaTableState;
 
 #[cfg(feature = "datafusion")]
 use super::state::AddContainer;
@@ -53,8 +53,8 @@ pub enum CommitConflictError {
     ///   you may need to upgrade your Delta Lake version.
     /// - When multiple writers are creating or replacing a table at the same time.
     /// - When multiple writers are writing to an empty path at the same time.
-    #[error("Protocol changed since last commit.")]
-    ProtocolChanged,
+    #[error("Protocol changed since last commit: {0}")]
+    ProtocolChanged(String),
 
     /// Error returned when the table requires an unsupported writer version
     #[error("Delta-rs does not support writer version {0}")]
@@ -392,10 +392,18 @@ impl<'a> ConflictChecker<'a> {
     /// to read and write against the protocol set by the committed transaction.
     fn check_protocol_compatibility(&self) -> Result<(), CommitConflictError> {
         for p in self.winning_commit_summary.protocol() {
-            if self.txn_info.read_snapshot.min_reader_version() < p.min_reader_version
-                || self.txn_info.read_snapshot.min_writer_version() < p.min_writer_version
-            {
-                return Err(CommitConflictError::ProtocolChanged);
+            let (win_read, curr_read) = (
+                p.min_reader_version,
+                self.txn_info.read_snapshot.min_reader_version(),
+            );
+            let (win_write, curr_write) = (
+                p.min_writer_version,
+                self.txn_info.read_snapshot.min_writer_version(),
+            );
+            if curr_read < win_read || win_write < curr_write {
+                return Err(CommitConflictError::ProtocolChanged(
+                    format!("reqired read/write {win_read}/{win_write}, current read/write {curr_read}/{curr_write}"),
+                ));
             };
         }
         if !self.winning_commit_summary.protocol().is_empty()
@@ -405,7 +413,9 @@ impl<'a> ConflictChecker<'a> {
                 .iter()
                 .any(|a| matches!(a, Action::protocol(_)))
         {
-            return Err(CommitConflictError::ProtocolChanged);
+            return Err(CommitConflictError::ProtocolChanged(
+                "protocol changed".into(),
+            ));
         };
         Ok(())
     }
@@ -818,7 +828,10 @@ mod tests {
             vec![tu::create_protocol_action(None, None)],
             false,
         );
-        assert!(matches!(result, Err(CommitConflictError::ProtocolChanged)));
+        assert!(matches!(
+            result,
+            Err(CommitConflictError::ProtocolChanged(_))
+        ));
 
         // taint whole table
         // `read_whole_table` should disallow any concurrent change, even if the change

rust/src/operations/transaction/mod.rs

@@ -9,7 +9,7 @@ use crate::action::{Action, CommitInfo, DeltaOperation};
 use crate::crate_version;
 use crate::errors::{DeltaResult, DeltaTableError};
 use crate::storage::commit_uri_from_version;
-use crate::table_state::DeltaTableState;
+use crate::table::state::DeltaTableState;
 
 mod conflict_checker;
 #[cfg(feature = "datafusion")]
@@ -164,7 +164,8 @@ pub async fn commit(
 ) -> DeltaResult<i64> {
     let tmp_commit = prepare_commit(storage, &operation, actions, app_metadata).await?;
 
-    let max_attempts = 5;
+    // TODO make max attempts configurable
+    let max_attempts = 25;
     let mut attempt_number = 1;
 
     while attempt_number <= max_attempts {

rust/src/operations/transaction/state.rs

@@ -24,7 +24,7 @@ use crate::delta_datafusion::{
     get_null_of_arrow_type, logical_expr_to_physical_expr, to_correct_scalar_value,
 };
 use crate::errors::{DeltaResult, DeltaTableError};
-use crate::table_state::DeltaTableState;
+use crate::table::state::DeltaTableState;
 
 impl DeltaTableState {
     /// Get the table schema as an [`ArrowSchemaRef`]

rust/src/operations/transaction/test_utils.rs

@@ -1,7 +1,7 @@
 #![allow(unused)]
 use super::{prepare_commit, try_commit_transaction, CommitInfo};
 use crate::action::{Action, Add, DeltaOperation, MetaData, Protocol, Remove, SaveMode};
-use crate::table_state::DeltaTableState;
+use crate::table::state::DeltaTableState;
 use crate::{
     DeltaTable, DeltaTableBuilder, DeltaTableMetaData, Schema, SchemaDataType, SchemaField,
 };

rust/src/operations/vacuum.rs

@@ -32,7 +32,7 @@ use object_store::{path::Path, ObjectStore};
 
 use crate::errors::{DeltaResult, DeltaTableError};
 use crate::storage::DeltaObjectStore;
-use crate::table_state::DeltaTableState;
+use crate::table::state::DeltaTableState;
 use crate::DeltaTable;
 
 /// Errors that can occur during vacuum

rust/src/operations/write.rs

@@ -32,14 +32,14 @@ use super::writer::{DeltaWriter, WriterConfig};
 use super::MAX_SUPPORTED_WRITER_VERSION;
 use super::{transaction::commit, CreateBuilder};
 use crate::action::{Action, Add, DeltaOperation, Remove, SaveMode};
-use crate::delta::DeltaTable;
 use crate::delta_datafusion::DeltaDataChecker;
 use crate::errors::{DeltaResult, DeltaTableError};
 use crate::schema::Schema;
 use crate::storage::{DeltaObjectStore, ObjectStoreRef};
-use crate::table_state::DeltaTableState;
+use crate::table::state::DeltaTableState;
 use crate::writer::record_batch::divide_by_partition_values;
 use crate::writer::utils::PartitionPath;
+use crate::DeltaTable;
 
 #[derive(thiserror::Error, Debug)]
 enum WriteError {

rust/src/schema.rs → rust/src/schema/mod.rs

@@ -8,6 +8,10 @@ use std::collections::HashMap;
 
 use crate::errors::DeltaTableError;
 
+#[cfg(all(feature = "arrow", feature = "parquet"))]
+pub mod arrow_convert;
+pub mod partitions;
+
 /// Type alias for a string expected to match a GUID/UUID format
 pub type Guid = String;
 

rust/src/partitions.rs → rust/src/schema/partitions.rs

@@ -2,7 +2,7 @@
 
 use std::convert::TryFrom;
 
-use super::schema::SchemaDataType;
+use super::SchemaDataType;
 use crate::errors::DeltaTableError;
 use std::cmp::Ordering;
 use std::collections::HashMap;
@@ -189,20 +189,18 @@ pub struct DeltaTablePartition<'a> {
     pub value: &'a str,
 }
 
-/**
- * Create a DeltaTable partition from a HivePartition string.
- *
- * A HivePartition string is represented by a "key=value" format.
- *
- * ```rust
- * use deltalake::DeltaTablePartition;
- *
- * let hive_part = "ds=2023-01-01";
- * let partition = DeltaTablePartition::try_from(hive_part).unwrap();
- * assert_eq!("ds", partition.key);
- * assert_eq!("2023-01-01", partition.value);
- * ```
- */
+/// Create a DeltaTable partition from a HivePartition string.
+///
+/// A HivePartition string is represented by a "key=value" format.
+///
+/// ```rust
+/// use deltalake::DeltaTablePartition;
+///
+/// let hive_part = "ds=2023-01-01";
+/// let partition = DeltaTablePartition::try_from(hive_part).unwrap();
+/// assert_eq!("ds", partition.key);
+/// assert_eq!("2023-01-01", partition.value);
+/// ```
 impl<'a> TryFrom<&'a str> for DeltaTablePartition<'a> {
     type Error = DeltaTableError;
 
@@ -223,20 +221,18 @@ impl<'a> TryFrom<&'a str> for DeltaTablePartition<'a> {
 }
 
 impl<'a> DeltaTablePartition<'a> {
-    /**
-     * Try to create a DeltaTable partition from a partition value kv pair.
-     *
-     * ```rust
-     * use deltalake::DeltaTablePartition;
-     *
-     * let value = (&"ds".to_string(), &Some("2023-01-01".to_string()));
-     * let null_default = "1979-01-01";
-     * let partition = DeltaTablePartition::from_partition_value(value, null_default);
-     *
-     * assert_eq!("ds", partition.key);
-     * assert_eq!("2023-01-01", partition.value);
-     * ```
-     */
+    /// Try to create a DeltaTable partition from a partition value kv pair.
+    ///
+    /// ```rust
+    /// use deltalake::DeltaTablePartition;
+    ///
+    /// let value = (&"ds".to_string(), &Some("2023-01-01".to_string()));
+    /// let null_default = "1979-01-01";
+    /// let partition = DeltaTablePartition::from_partition_value(value, null_default);
+    ///
+    /// assert_eq!("ds", partition.key);
+    /// assert_eq!("2023-01-01", partition.value);
+    /// ```
     pub fn from_partition_value(
         partition_value: (&'a String, &'a Option<String>),
         default_for_null: &'a str,

rust/src/builder.rs → rust/src/table/builder.rs

@@ -9,7 +9,7 @@ use object_store::DynObjectStore;
 use serde::{Deserialize, Serialize};
 use url::Url;
 
-use crate::delta::DeltaTable;
+use super::DeltaTable;
 use crate::errors::{DeltaResult, DeltaTableError};
 use crate::storage::config::StorageOptions;
 use crate::storage::{DeltaObjectStore, ObjectStoreRef};