Add implementation for load_with_datetime in Python package.

python/Cargo.toml

@@ -21,6 +21,7 @@ env_logger = "0"
 reqwest = { version = "*", features = ["native-tls-vendored"] }
 serde_json = "1"
 arrow  = { version = "5" }
+chrono = "0"
 
 [dependencies.pyo3]
 version = "0.14"

python/Makefile

@@ -21,6 +21,8 @@ build: setup ## Build Python binding of delta-rs
 
 .PHONY: install
 install: build ## Install Python binding of delta-rs
+	$(info --- Uninstall Python binding ---)
+	pip uninstall -y deltalake
 	$(info --- Install Python binding ---)
 	$(eval TARGET_WHEEL := $(shell ls ../target/wheels/deltalake-${PACKAGE_VERSION}-*.whl))
 	pip install $(TARGET_WHEEL)[devel,pandas]

python/deltalake/table.py

@@ -152,6 +152,20 @@ def load_version(self, version: int) -> None:
         """
         self._table.load_version(version)
 
+    def load_with_datetime(self, datetime_string: str) -> None:
+        """
+        Time travel Delta table to the latest version that's created at or before provided `datetime_string` argument.
+        The `datetime_string` argument should be an RFC 3339 and ISO 8601 date and time string.
+
+        Examples:
+        `2018-01-26T18:30:09Z`
+        `2018-12-19T16:39:57-08:00`
+        `2018-01-26T18:30:09.453+00:00`
+
+        :param datetime_string: the identifier of the datetime point of the DeltaTable to load
+        """
+        self._table.load_with_datetime(datetime_string)
+
     def schema(self) -> Schema:
         """
         Get the current schema of the DeltaTable.

python/src/lib.rs

@@ -4,6 +4,7 @@ extern crate arrow;
 extern crate pyo3;
 
 use arrow::datatypes::Schema as ArrowSchema;
+use chrono::{DateTime, FixedOffset, Utc};
 use deltalake::partitions::PartitionFilter;
 use pyo3::create_exception;
 use pyo3::exceptions::PyException;
@@ -25,6 +26,13 @@ impl PyDeltaTableError {
     fn from_tokio(err: tokio::io::Error) -> pyo3::PyErr {
         PyDeltaTableError::new_err(err.to_string())
     }
+
+    fn from_chrono(err: chrono::ParseError) -> pyo3::PyErr {
+        PyDeltaTableError::new_err(format!(
+            "Parse date and time string failed: {}",
+            err.to_string()
+        ))
+    }
 }
 
 #[inline]
@@ -100,6 +108,16 @@ impl RawDeltaTable {
             .map_err(PyDeltaTableError::from_raw)
     }
 
+    pub fn load_with_datetime(&mut self, ds: &str) -> PyResult<()> {
+        let datetime = DateTime::<Utc>::from(
+            DateTime::<FixedOffset>::parse_from_rfc3339(ds)
+                .map_err(PyDeltaTableError::from_chrono)?,
+        );
+        rt()?
+            .block_on(self._table.load_with_datetime(datetime))
+            .map_err(PyDeltaTableError::from_raw)
+    }
+
     pub fn files_by_partitions(
         &self,
         partitions_filters: Vec<(&str, &str, PartitionFilterValue)>,

python/tests/test_table_read.py

@@ -1,3 +1,4 @@
+import os
 from threading import Barrier, Thread
 
 import pandas as pd
@@ -18,6 +19,52 @@ def test_read_simple_table_by_version_to_dict():
     assert dt.to_pyarrow_dataset().to_table().to_pydict() == {"value": [1, 2, 3]}
 
 
+def test_load_with_datetime():
+    log_dir = "../rust/tests/data/simple_table/_delta_log"
+    log_mtime_pair = [
+        ("00000000000000000000.json", 1588398451.0),
+        ("00000000000000000001.json", 1588484851.0),
+        ("00000000000000000002.json", 1588571251.0),
+        ("00000000000000000003.json", 1588657651.0),
+        ("00000000000000000004.json", 1588744051.0),
+    ]
+    for file_name, dt_epoch in log_mtime_pair:
+        file_path = os.path.join(log_dir, file_name)
+        os.utime(file_path, (dt_epoch, dt_epoch))
+
+    table_path = "../rust/tests/data/simple_table"
+    dt = DeltaTable(table_path)
+    dt.load_with_datetime("2020-05-01T00:47:31-07:00")
+    assert dt.version() == 0
+    dt.load_with_datetime("2020-05-02T22:47:31-07:00")
+    assert dt.version() == 1
+    dt.load_with_datetime("2020-05-25T22:47:31-07:00")
+    assert dt.version() == 4
+
+
+def test_load_with_datetime_bad_format():
+    table_path = "../rust/tests/data/simple_table"
+    dt = DeltaTable(table_path)
+    with pytest.raises(Exception) as exception:
+        dt.load_with_datetime("2020-05-01T00:47:31")
+    assert (
+        str(exception.value)
+        == "Parse date and time string failed: premature end of input"
+    )
+    with pytest.raises(Exception) as exception:
+        dt.load_with_datetime("2020-05-01 00:47:31")
+    assert (
+        str(exception.value)
+        == "Parse date and time string failed: input contains invalid characters"
+    )
+    with pytest.raises(Exception) as exception:
+        dt.load_with_datetime("2020-05-01T00:47:31+08")
+    assert (
+        str(exception.value)
+        == "Parse date and time string failed: premature end of input"
+    )
+
+
 def test_read_simple_table_update_incremental():
     table_path = "../rust/tests/data/simple_table"
     dt = DeltaTable(table_path, version=0)

rust/src/delta.rs

@@ -1099,8 +1099,8 @@ impl DeltaTable {
         Ok(())
     }
 
-    /// Time travel Delta table to latest version that's created at or before provided `datetime`
-    /// argument.
+    /// Time travel Delta table to the latest version that's created at or before provided
+    /// `datetime` argument.
     ///
     /// Internally, this methods performs a binary search on all Delta transaction logs.
     pub async fn load_with_datetime(