find ancestor before timestamp

we cannot use snapshot_as_of_timestamp() as it finds previously current snapshots but not necessarily an ancestor.
An example is here: https://iceberg.apache.org/docs/nightly/spark-queries/?h=ancestor#history

Author: chinmay-bhat

pyiceberg/table/__init__.py

@@ -113,6 +113,7 @@
     SnapshotLogEntry,
     SnapshotSummaryCollector,
     Summary,
+    ancestor_right_before_timestamp,
     ancestors_of,
     update_snapshot_summaries,
 )
@@ -2050,7 +2051,11 @@ def rollback_to_timestamp(self, timestamp: int) -> ManageSnapshots:
             This for method chaining
         """
         self._commit_if_ref_updates_exist()
-        if (snapshot := self._transaction._table.snapshot_as_of_timestamp(timestamp, inclusive=False)) is None:
+        if (
+            snapshot := ancestor_right_before_timestamp(
+                self._transaction._table.current_snapshot(), self._transaction.table_metadata, timestamp
+            )
+        ) is None:
             raise ValidationError(f"Cannot roll back, no valid snapshot older than: {timestamp}")
 
         update, requirement = self._transaction._set_ref_snapshot(

pyiceberg/table/snapshots.py

@@ -419,6 +419,17 @@ def set_when_positive(properties: Dict[str, str], num: int, property_name: str)
         properties[property_name] = str(num)
 
 
+def ancestor_right_before_timestamp(
+    current_snapshot: Optional[Snapshot], table_metadata: TableMetadata, timestamp_ms: int
+) -> Optional[Snapshot]:
+    """Get the ancestor right before the given timestamp."""
+    if current_snapshot is not None:
+        for ancestor in ancestors_of(current_snapshot, table_metadata):
+            if ancestor.timestamp_ms < timestamp_ms:
+                return ancestor
+    return None
+
+
 def ancestors_of(current_snapshot: Optional[Snapshot], table_metadata: TableMetadata) -> Iterable[Snapshot]:
     """Get the ancestors of and including the given snapshot."""
     if current_snapshot: