add snapshot_by_timestamp API

Author: chinmay-bhat

pyiceberg/table/__init__.py

@@ -1290,6 +1290,10 @@ def snapshot_by_name(self, name: str) -> Optional[Snapshot]:
             return self.snapshot_by_id(ref.snapshot_id)
         return None
 
+    def snapshot_by_timestamp(self, timestamp_ms: int) -> Optional[Snapshot]:
+        """Get the snapshot right before the given timestamp, or None if there is no matching snapshot."""
+        return self.metadata.snapshot_by_timestamp(timestamp_ms)
+
     def history(self) -> List[SnapshotLogEntry]:
         """Get the snapshot history of this table."""
         return self.metadata.snapshot_log

pyiceberg/table/metadata.py

@@ -230,6 +230,14 @@ def snapshot_by_id(self, snapshot_id: int) -> Optional[Snapshot]:
         """Get the snapshot by snapshot_id."""
         return next((snapshot for snapshot in self.snapshots if snapshot.snapshot_id == snapshot_id), None)
 
+    def snapshot_by_timestamp(self, timestamp_ms: int) -> Optional[Snapshot]:
+        """Get the snapshot right before the given timestamp."""
+        result, prev_timestamp = None, 0
+        for snapshot in self.snapshots:
+            if prev_timestamp < snapshot.timestamp_ms < timestamp_ms:
+                result, prev_timestamp = snapshot, snapshot.timestamp_ms
+        return result if result else None
+
     def schema_by_id(self, schema_id: int) -> Optional[Schema]:
         """Get the schema by schema_id."""
         return next((schema for schema in self.schemas if schema.schema_id == schema_id), None)