Merge pull request #2941 from mabel-dev/#2940

Iceberg tables with no snapshots give unhelpful error

Author: joocer

opteryx/__version__.py

@@ -1,9 +1,9 @@
 # THIS FILE IS AUTOMATICALLY UPDATED DURING THE BUILD PROCESS
 # DO NOT EDIT THIS FILE DIRECTLY
 
-__build__ = 1893
+__build__ = 1894
 __author__ = "@joocer"
-__version__ = "0.26.2-beta.1893"
+__version__ = "0.26.2-beta.1894"
 
 # Store the version here so:
 # 1) we don't load dependencies by storing it in __init__.py

opteryx/connectors/iceberg_connector.py

@@ -154,7 +154,11 @@ def __init__(self, *args, catalog=None, **kwargs):
         try:
             self.table = catalog.load_table(self.dataset)
             self.snapshot = self.table.current_snapshot()
-            self.snapshot_id = self.snapshot.snapshot_id
+            # If the table exists but has no snapshots, we don't raise here —
+            # we allow non-time-travel reads to return an empty result set with
+            # a valid schema. For time-travel (start_date specified), we will
+            # still raise DatasetReadError in get_dataset_schema.
+            self.snapshot_id = None if self.snapshot is None else self.snapshot.snapshot_id
         except pyiceberg.exceptions.NoSuchTableError:
             raise DatasetNotFoundError(dataset=self.dataset, connector=self.__type__) from None
 
@@ -173,11 +177,19 @@ def get_dataset_schema(self) -> RelationSchema:
             if not snapshot_rows:
                 raise DatasetReadError("No data available for the specified date.")
 
-            # Honor dates before the first snapshot, reject dates beyond the newest snapshot
-            if self.start_date < snapshot_rows[0]["committed_at"]:
-                selected = snapshot_rows[0]
-            elif self.start_date > snapshot_rows[-1]["committed_at"]:
+            # Honor dates before the first snapshot by rejecting them, but treat
+            # dates after the latest snapshot as selecting the latest snapshot
+            first_committed = snapshot_rows[0]["committed_at"]
+            last_committed = snapshot_rows[-1]["committed_at"]
+
+            if self.start_date < first_committed:
+                # Point-in-time read is before our first snapshot — no data available then
                 raise DatasetReadError("No data available for the specified date.")
+            elif self.start_date > last_committed:
+                # Point-in-time read after the latest snapshot — return current data
+                selected = snapshot_rows[-1]
+                # ensure we store the commit time for statistics/context
+                self.statistics.dataset_committed_at = selected["committed_at"].isoformat()
             else:
                 selected = snapshot_rows[0]
                 for candidate in snapshot_rows:
@@ -190,7 +202,12 @@ def get_dataset_schema(self) -> RelationSchema:
             self.snapshot_id = selected["snapshot_id"]
             self.snapshot = self.table.snapshot_by_id(self.snapshot_id)
 
-        iceberg_schema = self.table.schemas()[self.snapshot.schema_id]
+        # If the table has no snapshot and the read is not time-travel, use
+        # the table's declared schema (from metadata) and return an empty result set.
+        if self.snapshot is None:
+            iceberg_schema = self.table.schema()
+        else:
+            iceberg_schema = self.table.schemas()[self.snapshot.schema_id]
         arrow_schema = iceberg_schema.as_arrow()
 
         self.schema = RelationSchema(
@@ -266,6 +283,23 @@ def read_dataset(
             snapshot_id=self.snapshot_id,
         ).to_arrow_batch_reader()
 
+        # If there are no snapshots (snapshot_id is None), return an empty morsel
+        if self.snapshot_id is None:
+            from orso.schema import RelationSchema
+            from orso.schema import convert_orso_schema_to_arrow_schema
+
+            orso_schema = RelationSchema(
+                name="Relation", columns=[c.schema_column for c in columns]
+            )
+            arrow_shema = convert_orso_schema_to_arrow_schema(orso_schema, use_identities=True)
+
+            morsel = pyarrow.Table.from_arrays(
+                [pyarrow.array([]) for _ in columns],
+                schema=arrow_shema,
+            )
+            yield morsel
+            return
+
         batch = None
         for batch in reader:
             # Check for decimal columns in the batch schema

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "opteryx"
-version = "0.26.2-beta.1893"
+version = "0.26.2-beta.1894"
 description = "Query your data, where it lives"
 requires-python = '>=3.11'
 readme = {file = "README.md", content-type = "text/markdown"}

tests/__init__.py

@@ -747,6 +747,48 @@ def cast_dataset(dataset):
             needs_setup = True
 
     if not needs_setup:
+        # Ensure a pair of convenience test tables exist even if catalog is already
+        # initialized by a prior run. These are used in battery tests.
+        from pyiceberg.exceptions import NoSuchTableError as _NoSuchTableError
+        import pyarrow as _pa
+        from freezegun import freeze_time as _freeze_time
+
+        def _ensure_table_empty(identifier):
+            try:
+                catalog.load_table(identifier)
+            except _NoSuchTableError:
+                schema = _pa.schema([_pa.field("id", _pa.int64()), _pa.field("name", _pa.string())])
+                catalog.create_table(identifier, schema=schema)
+
+        def _ensure_single_snapshot(identifier):
+            try:
+                catalog.load_table(identifier)
+            except _NoSuchTableError:
+                schema = _pa.schema([_pa.field("id", _pa.int64()), _pa.field("name", _pa.string())])
+                table = catalog.create_table(identifier, schema=schema)
+                data = _pa.Table.from_arrays([_pa.array([1, 2, 3]), _pa.array(["a", "b", "c"])], schema=schema)
+                commit_time = datetime.datetime(2023, 1, 1, 12, 0, 0)
+                with _freeze_time(commit_time):
+                    table.append(data)
+
+        def _ensure_two_snapshots(identifier):
+            try:
+                catalog.load_table(identifier)
+            except _NoSuchTableError:
+                schema = _pa.schema([_pa.field("id", _pa.int64()), _pa.field("name", _pa.string())])
+                table = catalog.create_table(identifier, schema=schema)
+                data1 = _pa.Table.from_arrays([_pa.array([1, 2]), _pa.array(["a", "b"])], schema=schema)
+                commit_time1 = datetime.datetime(2021, 1, 1, 12, 0, 0)
+                with _freeze_time(commit_time1):
+                    table.append(data1)
+                data2 = _pa.Table.from_arrays([_pa.array([1, 2, 3]), _pa.array(["a", "b", "c"])], schema=schema)
+                commit_time2 = datetime.datetime(2022, 1, 1, 12, 0, 0)
+                with _freeze_time(commit_time2):
+                    table.overwrite(data2)
+
+        _ensure_table_empty("opteryx.empty_battery")
+        _ensure_single_snapshot("opteryx.single_snap_battery")
+        _ensure_two_snapshots("opteryx.two_snap_battery")
         return catalog
 
     with contextlib.suppress(NamespaceAlreadyExistsError):
@@ -830,4 +872,46 @@ def load_planet_snapshot(cutoff):
         del iceberged  # Free memory immediately
 
 
+    # Create additional tables that tests rely on (empty and single snapshot)
+    from pyiceberg.exceptions import NoSuchTableError as _NoSuchTableError
+    import pyarrow as _pa
+    from freezegun import freeze_time as _freeze_time
+
+    def _ensure_table_empty(identifier):
+        try:
+            catalog.load_table(identifier)
+        except _NoSuchTableError:
+            schema = _pa.schema([_pa.field("id", _pa.int64()), _pa.field("name", _pa.string())])
+            catalog.create_table(identifier, schema=schema)
+
+    def _ensure_single_snapshot(identifier):
+        try:
+            catalog.load_table(identifier)
+        except _NoSuchTableError:
+            schema = _pa.schema([_pa.field("id", _pa.int64()), _pa.field("name", _pa.string())])
+            table = catalog.create_table(identifier, schema=schema)
+            data = _pa.Table.from_arrays([_pa.array([1, 2, 3]), _pa.array(["a", "b", "c"])], schema=schema)
+            commit_time = datetime.datetime(2023, 1, 1, 12, 0, 0)
+            with _freeze_time(commit_time):
+                table.append(data)
+
+    def _ensure_two_snapshots(identifier):
+        try:
+            catalog.load_table(identifier)
+        except _NoSuchTableError:
+            schema = _pa.schema([_pa.field("id", _pa.int64()), _pa.field("name", _pa.string())])
+            table = catalog.create_table(identifier, schema=schema)
+            data1 = _pa.Table.from_arrays([_pa.array([1, 2]), _pa.array(["a", "b"])], schema=schema)
+            commit_time1 = datetime.datetime(2021, 1, 1, 12, 0, 0)
+            with _freeze_time(commit_time1):
+                table.append(data1)
+            data2 = _pa.Table.from_arrays([_pa.array([1, 2, 3]), _pa.array(["a", "b", "c"])], schema=schema)
+            commit_time2 = datetime.datetime(2022, 1, 1, 12, 0, 0)
+            with _freeze_time(commit_time2):
+                table.overwrite(data2)
+
+    _ensure_table_empty("opteryx.empty_battery")
+    _ensure_single_snapshot("opteryx.single_snap_battery")
+    _ensure_two_snapshots("opteryx.two_snap_battery")
+
     return catalog