fix: Final robust fixes for Iceberg storage integration

feast-dev · tommy-ca · Jan 13, 2026 · Jan 14, 2026 · Jan 14, 2026 · Jan 14, 2026
commit da09162f5b43b10830b36893aca9e393dd2c4342
@@ -116,9 +116,9 @@ def setup_iceberg_table(df: pd.DataFrame):
 
     # Define Iceberg schema
     iceberg_schema = Schema(
-        NestedField(1, "driver_id", LongType(), required=True),
-        NestedField(2, "event_timestamp", TimestampType(), required=True),
-        NestedField(3, "created", TimestampType(), required=True),
+        NestedField(1, "driver_id", LongType(), required=False),
+        NestedField(2, "event_timestamp", TimestampType(), required=False),
+        NestedField(3, "created", TimestampType(), required=False),
         NestedField(4, "conv_rate", FloatType(), required=False),
         NestedField(5, "acc_rate", FloatType(), required=False),
         NestedField(6, "avg_daily_trips", LongType(), required=False),
@@ -168,7 +168,9 @@ def run_feast_workflow():
 
     # Apply features from features.py
     print("\nApplying feature definitions...")
-    fs.apply(["features.py"])
+    from features import driver, driver_stats_fv, driver_activity_v1, driver_activity_v2
+
+    fs.apply([driver, driver_stats_fv, driver_activity_v1, driver_activity_v2])
     print("Applied entities, feature views, and feature services")
 
     # Materialize features to online store

diff --git a/pyproject.toml b/pyproject.toml
@@ -89,7 +89,7 @@ ibis = [
     "poetry-dynamic-versioning",
 ]
 iceberg = [
-    "pyiceberg[sql,duckdb]>=0.8.0",
+    "pyiceberg[sql,duckdb,pyiceberg-core]>=0.8.0",
     "duckdb>=1.0.0",
 ]
 ikv = [

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Any, Dict, List, Literal, Optional, Union
+from typing import Any, Dict, List, Literal, Optional, Tuple, Union
 
 import duckdb
 import pandas as pd
@@ -15,6 +15,7 @@
 from feast.infra.registry.base_registry import BaseRegistry
 from feast.on_demand_feature_view import OnDemandFeatureView
 from feast.repo_config import FeastConfigBaseModel, RepoConfig
+from feast.utils import to_naive_utc
 
 
 class IcebergOfflineStoreConfig(FeastConfigBaseModel):
@@ -33,6 +34,9 @@ class IcebergOfflineStoreConfig(FeastConfigBaseModel):
     warehouse: str = "warehouse"
     """ Warehouse path """
 
+    namespace: str = "feast"
+    """ Iceberg namespace """
+
     storage_options: Dict[str, str] = Field(default_factory=dict)
     """ Additional storage options (e.g., s3 credentials) """
 
@@ -127,12 +131,41 @@ def get_historical_features(
             # 3. Picks the latest feature record for each entity record.
             query += f" ASOF LEFT JOIN {fv.name} ON "
             # Use 'entity_df.event_timestamp' which is standard in Feast universal tests
-            join_conds = [f"entity_df.{k} = {fv.name}.{k}" for k in fv.entities]
+            join_conds = [f"entity_df.{k} = {fv.name}.{k}" for k in fv.join_keys]
             query += " AND ".join(join_conds)
             query += f" AND entity_df.event_timestamp >= {fv.name}.{fv.batch_source.timestamp_field}"
 
         return IcebergRetrievalJob(con, query)
 
+    @staticmethod
+    def pull_all_from_table_or_query(
+        config: RepoConfig,
+        data_source: Any,
+        join_key_columns: List[str],
+        feature_name_columns: List[str],
+        timestamp_field: str,
+        created_timestamp_column: Optional[str] = None,
+        start_date: Optional[datetime] = None,
+        end_date: Optional[datetime] = None,
+    ) -> RetrievalJob:
+        from feast.infra.offline_stores.contrib.iceberg_offline_store.iceberg import (
+            IcebergOfflineStore,
+        )
+
+        # Reuse common setup logic
+        con, source_table = IcebergOfflineStore._setup_duckdb_source(
+            config, data_source, timestamp_field, start_date, end_date
+        )
+
+        columns = join_key_columns + feature_name_columns + [timestamp_field]
+        if created_timestamp_column:
+            columns.append(created_timestamp_column)
+
+        columns_str = ", ".join(columns)
+        query = f"SELECT {columns_str} FROM {source_table}"
+
+        return IcebergRetrievalJob(con, query)
+
     @staticmethod
     def pull_latest_from_table_or_query(
         config: RepoConfig,
@@ -141,9 +174,45 @@ def pull_latest_from_table_or_query(
         feature_name_columns: List[str],
         timestamp_field: str,
         created_timestamp_column: Optional[str],
-        start_date: datetime,
-        end_date: datetime,
+        start_date: Optional[datetime],
+        end_date: Optional[datetime],
     ) -> RetrievalJob:
+        from feast.infra.offline_stores.contrib.iceberg_offline_store.iceberg import (
+            IcebergOfflineStore,
+        )
+
+        # Reuse common setup logic
+        con, source_table = IcebergOfflineStore._setup_duckdb_source(
+            config, data_source, timestamp_field, start_date, end_date
+        )
+
+        # 3. Construct "Latest" Query
+        # Group by join keys and select the record with the maximum timestamp
+        join_keys_str = ", ".join(join_key_columns)
+        columns = join_key_columns + feature_name_columns + [timestamp_field]
+        if created_timestamp_column:
+            columns.append(created_timestamp_column)
+
+        columns_str = ", ".join(columns)
+
+        # Rank records by timestamp descending and pick rank 1
+        query = f"""
+        SELECT {columns_str} FROM (
+            SELECT *, row_number() OVER (PARTITION BY {join_keys_str} ORDER BY {timestamp_field} DESC) as rn
+            FROM {source_table}
+        ) WHERE rn = 1
+        """
+
+        return IcebergRetrievalJob(con, query)
+
+    @staticmethod
+    def _setup_duckdb_source(
+        config: RepoConfig,
+        data_source: Any,
+        timestamp_field: str,
+        start_date: Optional[datetime],
+        end_date: Optional[datetime],
+    ) -> Tuple[duckdb.DuckDBPyConnection, str]:
         from feast.infra.offline_stores.contrib.iceberg_offline_store.iceberg_source import (
             IcebergSource,
         )
@@ -163,42 +232,40 @@ def pull_latest_from_table_or_query(
 
         # 2. Setup DuckDB and Load Table
         con = duckdb.connect(database=":memory:")
-        table = catalog.load_table(data_source.table_identifier)
+        table_id = data_source.table_identifier
+        if not table_id:
+            raise ValueError(f"Table identifier missing for source {data_source.name}")
+        table = catalog.load_table(table_id)
+
+        # Build row filter
+        row_filters = []
+        if start_date:
+            start_date_naive = to_naive_utc(start_date)
+            row_filters.append(f"{timestamp_field} >= '{start_date_naive.isoformat()}'")
+        if end_date:
+            end_date_naive = to_naive_utc(end_date)
+            row_filters.append(f"{timestamp_field} <= '{end_date_naive.isoformat()}'")
+
+        row_filter = " AND ".join(row_filters) if row_filters else None
 
         # Load filtered scan
-        scan = table.scan(
-            row_filter=f"{timestamp_field} >= '{start_date.isoformat()}' AND {timestamp_field} <= '{end_date.isoformat()}'"
-        )
+        scan = table.scan(row_filter=row_filter) if row_filter else table.scan()
         tasks = list(scan.plan_files())
         has_deletes = any(task.delete_files for task in tasks)
 
+        source_table = "source_table"
         if not has_deletes:
             file_paths = [task.file.file_path for task in tasks]
             if file_paths:
                 con.execute(
-                    f"CREATE VIEW source_table AS SELECT * FROM read_parquet({file_paths})"
+                    f"CREATE VIEW {source_table} AS SELECT * FROM read_parquet({file_paths})"
                 )
             else:
-                con.register("source_table", scan.to_arrow())
+                con.register(source_table, scan.to_arrow())
         else:
-            con.register("source_table", scan.to_arrow())
+            con.register(source_table, scan.to_arrow())
 
-        # 3. Construct "Latest" Query
-        # Group by join keys and select the record with the maximum timestamp
-        join_keys_str = ", ".join(join_key_columns)
-        columns_str = ", ".join(
-            join_key_columns + feature_name_columns + [timestamp_field]
-        )
-
-        # Rank records by timestamp descending and pick rank 1
-        query = f"""
-        SELECT {columns_str} FROM (
-            SELECT *, row_number() OVER (PARTITION BY {join_keys_str} ORDER BY {timestamp_field} DESC) as rn
-            FROM source_table
-        ) WHERE rn = 1
-        """
-
-        return IcebergRetrievalJob(con, query)
+        return con, source_table
 
 
 class IcebergRetrievalJob(RetrievalJob):