chore: remove peek_next_commit from LogStore (delta-io#4059)

# Description

After removing deprecated methods from `DeltaTable` we no longer require
or use `peek_next_commit`. Comments hint at that we were looking forward
to remove this anyways. So this PR removed `peek_next_commit` from
`LogStore` and removes associated code.

Signed-off-by: Robert Pack <robstar.pack@gmail.com>

Author: roeap

crates/core/src/lib.rs

@@ -196,7 +196,6 @@ mod tests {
     use futures::TryStreamExt as _;
 
     use super::*;
-    use crate::table::PeekCommit;
     use test_utils::file_paths_from;
 
     #[tokio::test]
@@ -623,39 +622,6 @@ mod tests {
         assert_eq!(history3.len(), 5);
     }
 
-    #[tokio::test]
-    async fn test_poll_table_commits() -> DeltaResult<()> {
-        let table_path = std::path::Path::new("../test/tests/data/simple_table_with_checkpoint")
-            .canonicalize()
-            .unwrap();
-        let table_url = url::Url::from_directory_path(table_path).unwrap();
-        let mut table = crate::open_table_with_version(table_url, 9).await.unwrap();
-        assert_eq!(table.version(), Some(9));
-        let peek = table
-            .log_store()
-            .peek_next_commit(table.version().unwrap())
-            .await
-            .unwrap();
-        assert!(matches!(peek, PeekCommit::New(..)));
-
-        if let PeekCommit::New(version, actions) = peek {
-            assert_eq!(table.version(), Some(9));
-            assert_eq!(version, 10);
-            assert_eq!(actions.len(), 2);
-
-            table.update_incremental(None).await.unwrap();
-            assert_eq!(table.version(), Some(10));
-        };
-
-        let peek = table
-            .log_store()
-            .peek_next_commit(table.version().unwrap())
-            .await
-            .unwrap();
-        assert!(matches!(peek, PeekCommit::UpToDate));
-        Ok(())
-    }
-
     #[tokio::test]
     async fn test_read_vacuumed_log() {
         let table_path = std::path::Path::new("../test/tests/data/checkpoints_vacuumed")

crates/core/src/logstore/mod.rs

@@ -216,16 +216,6 @@ pub enum CommitOrBytes {
     LogBytes(Bytes),
 }
 
-/// The next commit that's available from underlying storage
-///
-#[derive(Debug)]
-pub enum PeekCommit {
-    /// The next commit version and associated actions
-    New(i64, Vec<Action>),
-    /// Provided DeltaVersion is up to date
-    UpToDate,
-}
-
 /// Configuration parameters for a log store
 #[derive(Debug, Clone)]
 pub struct LogStoreConfig {
@@ -308,19 +298,6 @@ pub trait LogStore: Send + Sync + AsAny {
     /// Find latest version currently stored in the delta log.
     async fn get_latest_version(&self, start_version: i64) -> DeltaResult<i64>;
 
-    /// Get the list of actions for the next commit
-    async fn peek_next_commit(&self, current_version: i64) -> DeltaResult<PeekCommit> {
-        let next_version = current_version + 1;
-        let commit_log_bytes = match self.read_commit_entry(next_version).await {
-            Ok(Some(bytes)) => Ok(bytes),
-            Ok(None) => return Ok(PeekCommit::UpToDate),
-            Err(err) => Err(err),
-        }?;
-
-        let actions = crate::logstore::get_actions(next_version, &commit_log_bytes);
-        Ok(PeekCommit::New(next_version, actions?))
-    }
-
     /// Get object store, can pass operation_id for object stores linked to an operation
     fn object_store(&self, operation_id: Option<Uuid>) -> Arc<dyn ObjectStore>;
 
@@ -463,10 +440,6 @@ impl<T: LogStore + ?Sized> LogStore for Arc<T> {
         T::get_latest_version(self, start_version).await
     }
 
-    async fn peek_next_commit(&self, current_version: i64) -> DeltaResult<PeekCommit> {
-        T::peek_next_commit(self, current_version).await
-    }
-
     fn object_store(&self, operation_id: Option<Uuid>) -> Arc<dyn ObjectStore> {
         T::object_store(self, operation_id)
     }
@@ -975,31 +948,6 @@ pub(crate) mod tests {
         );
     }
 
-    /// <https://github.com/delta-io/delta-rs/issues/3297>:w
-    #[tokio::test]
-    async fn test_peek_with_invalid_json() -> DeltaResult<()> {
-        use crate::logstore::object_store::memory::InMemory;
-        let memory_store = Arc::new(InMemory::new());
-        let log_path = Path::from("delta-table/_delta_log/00000000000000000001.json");
-
-        let log_content = r#"{invalid_json"#;
-
-        memory_store
-            .put(&log_path, log_content.into())
-            .await
-            .expect("Failed to write log file");
-
-        let table_uri = url::Url::parse("memory:///delta-table").unwrap();
-        let table = crate::DeltaTableBuilder::from_url(table_uri.clone())
-            .unwrap()
-            .with_storage_backend(memory_store, table_uri)
-            .build()?;
-
-        let result = table.log_store().peek_next_commit(0).await;
-        assert!(result.is_err());
-        Ok(())
-    }
-
     /// Collect list stream
     pub(crate) async fn flatten_list_stream(
         storage: &object_store::DynObjectStore,

crates/core/src/table/mod.rs

@@ -25,9 +25,6 @@ use crate::logstore::{
 use crate::partitions::PartitionFilter;
 use crate::{DeltaResult, DeltaTableBuilder, DeltaTableError};
 
-// NOTE: this use can go away when peek_next_commit is removed off of [DeltaTable]
-pub use crate::logstore::PeekCommit;
-
 pub mod builder;
 pub mod config;
 pub mod state;