high level ops and integration test

Author: OussamaSaoudi-db

kernel/src/history_manager/mod.rs

@@ -1,11 +1,15 @@
-use crate::internal_mod;
+//! This module defines the [`LogHistoryManager`], which can be used to perform timestamp queries
+//! over the Delta Log, translating from timestamps to Delta versions.
 
-use error::LogHistoryError;
-use error::TimestampOutOfRangeError;
+use crate::internal_mod;
+use error::{LogHistoryError, TimestampOutOfRangeError};
 use search::{binary_search_by_key_with_bounds, Bound, SearchError};
+use std::cell::RefCell;
 use std::cmp::Ordering;
+use std::collections::HashMap;
 use std::fmt::Debug;
 use std::sync::Arc;
+use url::Url;
 
 use crate::log_segment::LogSegment;
 use crate::path::ParsedLogPath;
@@ -14,8 +18,8 @@ use crate::utils::require;
 use crate::{DeltaResult, Engine, Error as DeltaError, RowVisitor, Version};
 use timestamp_visitor::InCommitTimestampVisitor;
 
-pub(crate) mod error;
 internal_mod!(pub(crate) mod search);
+pub mod error;
 mod timestamp_visitor;
 
 type Timestamp = i64;
@@ -27,8 +31,8 @@ type Timestamp = i64;
 ///
 /// Use this manager to:
 /// - Convert timestamps or timestamp ranges into Delta versions or version ranges
-/// - Perform time travel queries using [`Table::snapshot`]
-/// - Execute timestamp-based change data feed queries using [`Table::table_changes`]
+/// - Perform time travel queries using `Table::snapshot`
+/// - Execute timestamp-based change data feed queries using `Table::table_changes`
 ///
 /// The [`LogHistoryManager`] works with tables regardless of whether they have In-Commit
 /// Timestamps enabled.
@@ -38,14 +42,13 @@ type Timestamp = i64;
 /// Once created, the [`LogHistoryManager`] does not automatically update with newer versions
 /// of the table. All timestamp queries are limited to the state captured in the [`Snapshot`]
 /// provided during construction.
-#[allow(unused)]
 #[derive(Debug)]
-pub(crate) struct LogHistoryManager {
+pub struct LogHistoryManager {
     log_segment: LogSegment,
     snapshot: Arc<Snapshot>,
+    commit_to_timestamp_cache: RefCell<HashMap<Url, Timestamp>>,
 }
 
-#[allow(unused)]
 #[derive(Debug)]
 enum TimestampSearchBounds {
     ExactMatch(Version),
@@ -92,25 +95,41 @@ impl LogHistoryManager {
         Ok(Self {
             log_segment,
             snapshot,
+            commit_to_timestamp_cache: Default::default(),
         })
     }
+    fn update_cache_with_timestamp(&self, commit_file: &ParsedLogPath, value: Timestamp) {
+        self.commit_to_timestamp_cache
+            .borrow_mut()
+            .insert(commit_file.location.location.clone(), value);
+    }
+    fn get_cached_timestamp(&self, commit_file: &ParsedLogPath) -> Option<Timestamp> {
+        self.commit_to_timestamp_cache
+            .borrow()
+            .get(&commit_file.location.location)
+            .copied()
+    }
 
     /// Gets the timestamp for the `commit_file`. If `read_ict` is false ,this returns the file's
     /// modification timestamp. If `read_ict` is true, this reads the file's In-commit timestamp.
-    #[allow(unused)]
     fn commit_file_to_timestamp(
         &self,
         engine: &dyn Engine,
         commit_file: &ParsedLogPath,
         read_ict: bool,
     ) -> Result<Timestamp, LogHistoryError> {
+        if let Some(cached) = self.get_cached_timestamp(commit_file) {
+            return Ok(cached);
+        }
         let commit_timestamp = if read_ict {
             Self::read_in_commit_timestamp(engine, commit_file)?
         } else {
             // By default, the timestamp of a commit is its modification time
             commit_file.location.last_modified
         };
 
+        self.update_cache_with_timestamp(commit_file, commit_timestamp);
+
         Ok(commit_timestamp)
     }
 
@@ -122,7 +141,6 @@ impl LogHistoryManager {
     /// This returns a [`LogHistoryError::InCommitTimestampNotFoundError`] if the in-commit timestamp
     /// is not present in the commit file, or if the CommitInfo is not the first action in the
     /// commit.
-    #[allow(unused)]
     fn read_in_commit_timestamp(
         engine: &dyn Engine,
         commit_file: &ParsedLogPath,
@@ -134,7 +152,7 @@ impl LogHistoryManager {
         };
 
         // Get an iterator over the actions in the commit file
-        let mut action_iter = engine
+        let action_iter = engine
             .json_handler()
             .read_json_files(
                 &[commit_file.location.clone()],
@@ -143,7 +161,7 @@ impl LogHistoryManager {
             )
             .map_err(wrap_err)?;
 
-        // Take the first non-empty engine data batch
+        // Take the first engine data batch
         let batch = action_iter
             .map(|res| res.map_err(wrap_err))
             .next()
@@ -164,11 +182,170 @@ impl LogHistoryManager {
         })
     }
 
+    /// Gets the latest version that occurs before or at the given `timestamp`.
+    ///
+    /// This finds the version whose timestamp is less than or equal to `timestamp`.
+    /// If no such version exists, returns [`LogHistoryError::TimestampOutOfRange`].
+    ///
+    ////// # Examples
+    /// ```rust
+    /// # use delta_kernel::history_manager::error::LogHistoryError;
+    /// # use delta_kernel::engine::sync::SyncEngine;
+    /// # use delta_kernel::Table;
+    /// # use std::sync::Arc;
+    /// # let path = "./tests/data/with_checkpoint_no_last_checkpoint";
+    /// # let engine = Arc::new(SyncEngine::new());
+    /// let table = Table::try_from_uri(path)?;
+    /// let manager = table.history_manager(engine.as_ref())?;
+    ///
+    /// // Get the latest version as of January 1, 2023
+    /// let timestamp = 1672531200000; // Milliseconds since epoch for 2023-01-01
+    /// let version_res = manager.latest_version_as_of(engine.as_ref(), timestamp);
+    /// # Ok::<(), delta_kernel::Error>(())
+    /// ```
+    pub fn latest_version_as_of(
+        &self,
+        engine: &dyn Engine,
+        timestamp: Timestamp,
+    ) -> DeltaResult<Version> {
+        Ok(self.timestamp_to_version(engine, timestamp, Bound::GreatestLower)?)
+    }
+
+    /// Gets the first version that occurs after the given `timestamp` (inclusive).
+    ///
+    /// This finds the version whose timestamp is greater than or equal to `timestamp`.
+    /// If no such version exists, returns [`LogHistoryError::TimestampOutOfRange`].
+    /// # Examples
+    /// ```rust
+    /// # use delta_kernel::engine::sync::SyncEngine;
+    /// # use delta_kernel::Table;
+    /// # use std::sync::Arc;
+    /// # let path = "./tests/data/with_checkpoint_no_last_checkpoint";
+    /// # let engine = Arc::new(SyncEngine::new());
+    /// let table = Table::try_from_uri(path)?;
+    /// let manager = table.history_manager(engine.as_ref())?;
+    ///
+    /// // Find the first version that occurred after January 1, 2023
+    /// let timestamp = 1672531200000; // Milliseconds since epoch for 2023-01-01
+    /// let version_res = manager.first_version_after(engine.as_ref(), timestamp);
+    /// # Ok::<(), delta_kernel::Error>(())
+    /// ```
+    pub fn first_version_after(
+        &self,
+        engine: &dyn Engine,
+        timestamp: Timestamp,
+    ) -> DeltaResult<Version> {
+        Ok(self.timestamp_to_version(engine, timestamp, Bound::LeastUpper)?)
+    }
+
+    /// Converts a timestamp range to a corresponding version range.
+    ///
+    /// This function finds the version range that corresponds to the given timestamp range.
+    /// The returned tuple contains:
+    /// - The first (earliest) version with a timestamp greater than or equal to `start_timestamp`
+    /// - If `end_timestamp` is provided, the version with a timestamp less than or equal to `end_timestamp`.
+    ///
+    /// # Arguments
+    /// * `engine` - The engine used to access version history
+    /// * `start_timestamp` - The lower bound timestamp (inclusive)
+    /// * `end_timestamp` - The optional upper bound timestamp (inclusive), or `None` to indicate no upper bound
+    ///
+    /// # Returns
+    /// A tuple containing the start version and optional end version (inclusive)
+    ///
+    /// # Errors
+    /// Returns [`LogHistoryError::TimestampOutOfRange`] if:
+    /// - No version exists at or after `start_timestamp`
+    /// - `end_timestamp` is provided and no version exists at or before it
+    ///
+    /// Returns [`LogHistoryError::InvalidTimestampRange`] if the entire range [start_timestamp,
+    /// end_timestamp]
+    ///
+    /// # Examples
+    /// ```rust
+    /// # use delta_kernel::engine::sync::SyncEngine;
+    /// # use delta_kernel::Table;
+    /// # use std::sync::Arc;
+    /// # let path = "./tests/data/with_checkpoint_no_last_checkpoint";
+    /// # let engine = Arc::new(SyncEngine::new());
+    ///
+    /// let table = Table::try_from_uri(path)?;
+    /// let manager = table.history_manager(engine.as_ref())?;
+    ///
+    /// // Find versions between January 1, 2023 and March 1, 2023
+    /// let start_timestamp = 1672531200000; // Jan 1, 2023 (milliseconds since epoch)
+    /// let end_timestamp = 1677628800000;   // Mar 1, 2023 (milliseconds since epoch)
+    ///
+    /// let version_range_res =
+    ///     manager.timestamp_range_to_versions(engine.as_ref(), start_timestamp, end_timestamp);
+    /// # Ok::<(), delta_kernel::Error>(())
+    /// ```
+    pub fn timestamp_range_to_versions(
+        &self,
+        engine: &dyn Engine,
+        start_timestamp: Timestamp,
+        end_timestamp: impl Into<Option<Timestamp>>,
+    ) -> DeltaResult<(Version, Option<Version>)> {
+        // Check that the start and end timestamps are valid. Timestamps must be positive
+        let end_timestamp = end_timestamp.into();
+        require!(
+            0 <= start_timestamp,
+            LogHistoryError::InvalidTimestamp(start_timestamp).into()
+        );
+        if let Some(end_timestamp) = end_timestamp {
+            require!(
+                0 <= end_timestamp,
+                LogHistoryError::InvalidTimestamp(end_timestamp).into()
+            );
+            // The `start_timestamp` must be no greater than the `end_timestamp`.
+            require!(
+                start_timestamp <= end_timestamp,
+                LogHistoryError::InvalidTimestampRange {
+                    start_timestamp,
+                    end_timestamp
+                }
+                .into()
+            );
+        }
+
+        // Convert the start timestamp to version
+        let start_version = self.first_version_after(engine, start_timestamp)?;
+
+        // If the end timestamp is present, convert it to an end version
+        let end_version = end_timestamp
+            .map(|end| {
+                let end_version = self.latest_version_as_of(engine, end)?;
+
+                // Verify that the start version is no greater than the end version. This can
+                // happen in the case that the entire timestamp range falls between two commits.
+                // Consider the following history:
+                // |-------------|--------------------|---------------|
+                // v4       start_timestamp      end_timestamp       v5
+                //
+                // The latest version as of the end_timestamp is 4. The first version after the
+                // start_timestamp is 5. Thus in the case where end_version < start_version, we
+                // return and [`LogHistoryError::EmptyTimestampRange`].
+                require!(
+                    start_version <= end_version,
+                    DeltaError::from(LogHistoryError::EmptyTimestampRange {
+                        end_timestamp: end,
+                        start_timestamp,
+                        between_left: end_version,
+                        between_right: start_version
+                    })
+                );
+
+                Ok(end_version)
+            })
+            .transpose()?;
+
+        Ok((start_version, end_version))
+    }
+
     /// Given a timestamp, this function determines the commit range that timestamp conversion
     /// should search. A timestamp search may be conducted over one of two version ranges:
     ///     1) A range of commits whose timestamp is the file modification timestamp
     ///     2) A range of commits whose timestamp is an in-commit timestamp.
-    #[allow(unused)]
     fn get_timestamp_search_bounds(
         &self,
         timestamp: Timestamp,
@@ -236,7 +413,6 @@ impl LogHistoryManager {
     /// happen in the following cases based on the bound:
     /// - `Bound::GreatestLower`: There is no commit whose timestamp is lower than the given `timestamp`.
     /// - `Bound::LeastUpper`: There is no commit whose timestamp is greater than the given `timestamp`.
-    #[allow(unused)]
     fn timestamp_to_version(
         &self,
         engine: &dyn Engine,

kernel/src/history_manager/search.rs

@@ -15,15 +15,13 @@ use delta_kernel_derive::internal_api;
 /// * [`Bound::GreatestLower`] - Finds the largest index `i` such that `values[i] <= key`.
 ///   This represents the last element less than or equal to the search key.
 #[internal_api]
-#[allow(unused)]
 #[derive(Debug, Clone, Copy, PartialEq)]
 pub(crate) enum Bound {
     LeastUpper,
     GreatestLower,
 }
 
 #[internal_api]
-#[allow(unused)]
 #[derive(Debug)]
 pub(crate) enum SearchError<T: Error> {
     OutOfRange,
@@ -119,7 +117,6 @@ pub(crate) enum SearchError<T: Error> {
 /// assert!(matches!(result, Err(SearchError::KeyFunctionError(_))));
 /// ```
 #[internal_api]
-#[allow(unused)]
 pub(crate) fn binary_search_by_key_with_bounds<'a, T, K: Ord + Debug, E: Error>(
     values: &'a [T],
     key: K,

kernel/src/history_manager/timestamp_visitor.rs

@@ -14,14 +14,12 @@ use crate::{DeltaResult, Error, RowVisitor};
 ///
 /// Only one row of the engine data is checked because CommitInfo. This is because in-commit
 /// timestamps requires that the CommitInfo containing the ICT be the first action in the log.
-#[allow(unused)]
 #[derive(Default)]
 pub(crate) struct InCommitTimestampVisitor {
     pub(crate) in_commit_timestamp: Option<i64>,
 }
 
 impl InCommitTimestampVisitor {
-    #[allow(unused)]
     /// Get the schema that the visitor expects the data to have.
     pub(crate) fn schema() -> Arc<Schema> {
         static SCHEMA: LazyLock<Arc<Schema>> = LazyLock::new(|| {

kernel/src/lib.rs

@@ -78,6 +78,7 @@ pub mod checkpoint;
 pub mod engine_data;
 pub mod error;
 pub mod expressions;
+pub mod history_manager;
 pub mod scan;
 pub mod schema;
 pub mod snapshot;
@@ -95,7 +96,6 @@ pub use arrow_compat::*;
 pub(crate) mod kernel_predicates;
 pub(crate) mod utils;
 
-internal_mod!(pub(crate) mod history_manager);
 internal_mod!(pub(crate) mod path);
 internal_mod!(pub(crate) mod log_replay);
 internal_mod!(pub(crate) mod log_segment);

kernel/src/log_segment.rs

@@ -191,7 +191,6 @@ impl LogSegment {
         LogSegment::try_new(ascending_commit_files, vec![], log_root, end_version)
     }
 
-    #[allow(unused)]
     /// Constructs a [`LogSegment`] to be used for timestamp conversion. This [`LogSegment`] will consist
     /// only of contiguous commit files. If an `end_version` is specified, the commit range will
     /// include commits up to the `end_version` commit (inclusive). If present, `limit` specifies the