fix: Phase 5.1 - Fix offline/online store bugs from code audit

- Fix duplicate query building in offline store get_historical_features - Fix online store schema to use IntegerType instead of Arrow pa.int32 - Update plan.md with comprehensive Phase 5 breakdown - Add PHASE5_STATUS.md tracking document Bug Fixes: - Offline store: Removed duplicate SELECT and FROM clauses (lines 111-130) - Online store: Changed entity_hash type from pa.int32() to IntegerType() All ruff checks passed. Ready for Phase 5.2 (integration tests).
feast-dev · tommy-ca · Jan 13, 2026 · Jan 14, 2026 · Jan 14, 2026 · Jan 14, 2026
commit 8ce4bd85f1cf80cdb5799185a613752bfe84ca76
@@ -0,0 +1,221 @@
+# Iceberg Phase 5: Code Audit, Bug Fixes & Integration Plan
+
+## Status: Bug Fixes COMPLETE ✅ | Tests & Docs IN PROGRESS
+
+### Completed in This Session
+
+#### 1. Comprehensive Code Audit ✅
+**Offline Store (`iceberg.py` - 232 lines)**:
+- ✅ **Found**: Duplicate query building bug (lines 111-130)
+- ✅ **Fixed**: Removed duplicate SELECT clause and FROM clause
+- ✅ **Result**: Clean query building with single pass
+
+**Online Store (`iceberg.py` - 541 lines)**:
+- ✅ **Found**: Arrow type used in Iceberg schema (line 332)
+- ✅ **Fixed**: Changed `pa.int32()` to `IntegerType()` from pyiceberg.types
+- ✅ **Result**: Proper Iceberg type usage throughout
+
+**Data Source (`iceberg_source.py` - 132 lines)**:
+- ✅ No issues found
+- ✅ Complete protobuf serialization
+- ✅ Proper type mapping
+
+#### 2. Code Quality Verification ✅
+```bash
+uv run ruff check --fix sdk/python/feast/infra/offline_stores/contrib/iceberg_offline_store/
+uv run ruff check --fix sdk/python/feast/infra/online_stores/contrib/iceberg_online_store/
+# Result: All checks passed!
+```
+
+#### 3. Plan.md Updated ✅
+- Added comprehensive Phase 5 breakdown
+- Documented bug fixes
+- Outlined test plan
+- Specified R2 documentation requirements
+- Created local example specifications
+
+### Remaining Tasks
+
+#### High Priority
+
+**Integration Tests** (3 tasks):
+1. Create `sdk/python/tests/integration/offline_store/test_iceberg_offline_store.py`
+   - Point-in-time correct feature retrieval
+   - COW vs MOR read strategy selection
+   - Materialization queries
+
+2. Create `sdk/python/tests/integration/online_store/test_iceberg_online_store.py`
+   - Online write and read consistency
+   - Entity hash partitioning
+   - Latest record selection
+
+3. Create `sdk/python/tests/integration/feature_repos/universal/online_store/iceberg.py`
+   - IcebergOnlineStoreCreator for universal tests
+   - Register in AVAILABLE_ONLINE_STORES
+
+**Local Development Example** (1 task):
+4. Create `examples/iceberg-local/` with complete working example
+   - SQLite catalog + DuckDB engine
+   - Sample data generation
+   - End-to-end workflow
+   - README with step-by-step instructions
+
+#### Medium Priority
+
+**Cloudflare R2 Documentation** (2 tasks):
+5. Add R2 section to `docs/reference/offline-stores/iceberg.md`
+   - R2-compatible S3 configuration
+   - R2 Data Catalog setup
+
+6. Add R2 section to `docs/reference/online-stores/iceberg.md`
+   - R2 storage options
+   - Virtual addressing configuration
+
+### Bug Fixes Applied
+
+#### Fix 1: Duplicate Query Building (Offline Store)
+
+**Before** (lines 111-130):
+```python
+query = "SELECT entity_df.*"
+for fv in feature_views:
+    for feature in fv.features:
+        feature_name = feature.name
+        if full_feature_names:
+            feature_name = f"{fv.name}__{feature.name}"
+        query += f", {fv.name}.{feature.name} AS {feature_name}"
+
+query += " FROM entity_df"  # First FROM
+for fv in feature_views:     # Duplicate loop!
+    for feature in fv.features:
+        feature_name = feature.name
+        if full_feature_names:
+            feature_name = f"{fv.name}__{feature.name}"
+        query += f", {fv.name}.{feature.name} AS {feature_name}"
+
+query += " FROM entity_df"  # Duplicate FROM!
+for fv in feature_views:
+    # ASOF JOIN logic...
+```
+
+**After** (fixed):
+```python
+query = "SELECT entity_df.*"
+for fv in feature_views:
+    # Add all features from the feature view to SELECT clause
+    for feature in fv.features:
+        feature_name = feature.name
+        if full_feature_names:
+            feature_name = f"{fv.name}__{feature.name}"
+        query += f", {fv.name}.{feature.name} AS {feature_name}"
+
+query += " FROM entity_df"  # Single FROM
+for fv in feature_views:
+    # ASOF JOIN logic...
+```
+
+**Impact**: Fixes incorrect SQL generation that would have caused query errors.
+
+#### Fix 2: Iceberg Type Usage (Online Store)
+
+**Before** (line 332):
+```python
+NestedField(field_id=2, name="entity_hash", type=pa.int32(), required=True),
+```
+
+**After** (fixed):
+```python
+from pyiceberg.types import IntegerType
+
+NestedField(field_id=2, name="entity_hash", type=IntegerType(), required=True),
+```
+
+**Impact**: Uses proper Iceberg types instead of Arrow types for schema definition.
+
+### Test Plan Summary
+
+#### Standalone Tests Structure
+```
+sdk/python/tests/integration/
+├── offline_store/
+│   └── test_iceberg_offline_store.py (NEW)
+├── online_store/
+│   └── test_iceberg_online_store.py (NEW)
+└── feature_repos/universal/
+    └── online_store/
+        └── iceberg.py (NEW - IcebergOnlineStoreCreator)
+```
+
+#### Test Coverage
+- ✅ Local SQLite catalog (no external dependencies)
+- ✅ Point-in-time correct feature retrieval
+- ✅ COW/MOR hybrid strategy
+- ✅ Entity hash partitioning
+- ✅ Online write/read consistency
+- ✅ Latest record selection
+
+### R2 Configuration Summary
+
+#### R2-Compatible S3 Storage Options
+```yaml
+storage_options:
+    s3.endpoint: https://<account-id>.r2.cloudflarestorage.com
+    s3.access-key-id: ${R2_ACCESS_KEY_ID}
+    s3.secret-access-key: ${R2_SECRET_ACCESS_KEY}
+    s3.region: auto
+    s3.force-virtual-addressing: true
+```
+
+#### R2 Data Catalog (Beta)
+```yaml
+catalog_type: rest
+uri: <r2-catalog-uri>
+warehouse: <r2-warehouse-name>
+storage_options:
+    token: ${R2_DATA_CATALOG_TOKEN}
+```
+
+### Local Example Structure
+```
+examples/iceberg-local/
+├── feature_store.yaml     # SQLite catalog config
+├── features.py           # Entity and feature view definitions
+├── run_example.py        # Complete end-to-end workflow
+└── README.md            # Step-by-step instructions
+```
+
+### Next Steps
+
+1. **Commit bug fixes** (Phase 5.1 complete)
+2. **Create integration tests** (Phase 5.2 - in queue)
+3. **Add R2 documentation** (Phase 5.3 - in queue)
+4. **Create local example** (Phase 5.4 - in queue)
+
+### Files Modified
+
+**Bug Fixes** (2 files):
+1. `sdk/python/feast/infra/offline_stores/contrib/iceberg_offline_store/iceberg.py`
+   - Removed duplicate query building (lines 111-130)
+
+2. `sdk/python/feast/infra/online_stores/contrib/iceberg_online_store/iceberg.py`
+   - Fixed IntegerType import and usage (line 332)
+
+**Documentation** (1 file):
+3. `docs/specs/plan.md`
+   - Added comprehensive Phase 5 breakdown
+
+### Success Metrics
+
+- ✅ Ruff checks: All passed
+- ✅ No syntax errors
+- ✅ Proper type usage throughout
+- ✅ Clean query generation
+- ⏳ Integration tests: Pending
+- ⏳ R2 documentation: Pending
+- ⏳ Local example: Pending
+
+---
+
+**Phase 5.1 Status**: ✅ COMPLETE  
+**Phase 5.2-5.4 Status**: 📋 PLANNED  
+**Overall Phase 5 Progress**: 30% (3/10 tasks complete)
@@ -246,24 +246,76 @@ All documentation objectives achieved. Ready for final commit.
 
 ---
 
-### Phase 5: Maintenance & Monitoring (PLANNED)
-- [ ] Create comprehensive documentation:
-    - [ ] Add `docs/reference/offline-stores/iceberg.md` with configuration examples.
-    - [ ] Add `docs/reference/online-stores/iceberg.md` with performance characteristics.
-    - [ ] Add quickstart guide for Iceberg setup.
-- [ ] Final audit:
-    - [ ] Review type mappings in `feast/type_map.py` for completeness.
-    - [ ] Performance benchmarking against other offline stores.
-    - [ ] Security audit for catalog credentials handling.
-- [ ] Update CHANGELOG.md with new feature.
-- [ ] **Checkpoint**: Documentation review and merge.
-
-### Phase 5: Maintenance & Monitoring
-- [ ] Monitor upstream dependency releases:
-    - [ ] pyiceberg upgrades (watch for v0.9+ for Pydantic fixes).
-    - [ ] testcontainers-python upgrades (deprecation fixes).
-- [ ] Set up CI/CD for Iceberg tests.
-- [ ] Community feedback integration.
+### Phase 5: Code Audit, Bug Fixes & Integration Tests (IN PROGRESS)
+
+**Status**: Bug fixes and integration tests implementation
+
+**Completion Target**: 2026-01-14
+
+#### Phase 5.1: Code Audit & Bug Fixes
+
+**Audit Findings**:
+- ✅ Offline Store: Duplicate query building bug found (lines 111-130)
+- ✅ Online Store: Incorrect Arrow type in Iceberg schema (line 332)
+- ✅ Test Infrastructure: Already registered in AVAILABLE_OFFLINE_STORES
+
+**Bug Fixes**:
+- [ ] Fix duplicate query building in offline store `get_historical_features`
+- [ ] Fix Iceberg schema builder to use `IntegerType()` instead of `pa.int32()`
+- [ ] Verify all type mappings are complete in `type_map.py`
+
+#### Phase 5.2: Integration Tests
+
+**Standalone Tests**:
+- [ ] Create `test_iceberg_offline_store.py` - Isolated offline store tests
+- [ ] Create `test_iceberg_online_store.py` - Isolated online store tests
+- [ ] Create `IcebergOnlineStoreCreator` for universal test framework
+
+**Test Coverage**:
+- [ ] Point-in-time correct feature retrieval
+- [ ] COW vs MOR read strategy selection
+- [ ] Entity hash partitioning functionality
+- [ ] Online write and read consistency
+- [ ] Latest record selection per entity
+
+#### Phase 5.3: Cloudflare R2 Documentation
+
+**R2 Configuration Examples**:
+- [ ] Add R2 + REST catalog example to offline store docs
+- [ ] Add R2 + REST catalog example to online store docs
+- [ ] Create dedicated R2 quickstart guide (`iceberg_cloudflare_r2.md`)
+
+**Coverage**:
+- [ ] R2-compatible S3 endpoint configuration
+- [ ] R2 Data Catalog (native Iceberg catalog) setup
+- [ ] Authentication with R2 access keys
+- [ ] Force virtual addressing for R2 compatibility
+
+#### Phase 5.4: Local Development Example
+
+**Complete Example** (`examples/iceberg-local/`):
+- [ ] Create `feature_store.yaml` with SQLite catalog
+- [ ] Create `features.py` with entity and feature view definitions
+- [ ] Create `run_example.py` with end-to-end workflow
+- [ ] Create `README.md` with step-by-step instructions
+
+**Example Demonstrates**:
+- [ ] Local SQLite catalog + DuckDB engine setup
+- [ ] Sample data generation and Iceberg table creation
+- [ ] Feature definition and application
+- [ ] Materialization to online store
+- [ ] Online feature retrieval
+- [ ] Historical feature retrieval (point-in-time correct)
+
+#### **Checkpoint**: Phase 5 Complete when all tests pass and examples run
+
+---
+
+### Phase 6: Maintenance & Monitoring (FUTURE)
+- [ ] Monitor upstream dependency releases
+- [ ] Set up CI/CD for Iceberg tests
+- [ ] Community feedback integration
+- [ ] Performance benchmarking
 
 ## Design Specifications
 - [Offline Store Spec](iceberg_offline_store.md)

@@ -111,16 +111,7 @@ def get_historical_features(
         # 4. Construct ASOF join query with feature name handling
         query = "SELECT entity_df.*"
         for fv in feature_views:
-            # Join all features from the feature view
-            for feature in fv.features:
-                feature_name = feature.name
-                if full_feature_names:
-                    feature_name = f"{fv.name}__{feature.name}"
-                query += f", {fv.name}.{feature.name} AS {feature_name}"
-
-        query += " FROM entity_df"
-        for fv in feature_views:
-            # Join all features from the feature view
+            # Add all features from the feature view to SELECT clause
             for feature in fv.features:
                 feature_name = feature.name
                 if full_feature_names:
@@ -211,7 +202,12 @@ def pull_latest_from_table_or_query(
 
 
 class IcebergRetrievalJob(RetrievalJob):
-    def __init__(self, con: duckdb.DuckDBPyConnection, query: str, full_feature_names: bool = False):
+    def __init__(
+        self,
+        con: duckdb.DuckDBPyConnection,
+        query: str,
+        full_feature_names: bool = False,
+    ):
         self.con = con
         self.query = query
         self._full_feature_names = full_feature_names

@@ -325,11 +325,15 @@ def _build_online_schema(
         self, table: FeatureView, config: IcebergOnlineStoreConfig
     ) -> Schema:
         """Build Iceberg schema for online table."""
+        from pyiceberg.types import IntegerType
+
         fields = [
             NestedField(
                 field_id=1, name="entity_key", type=StringType(), required=True
             ),
-            NestedField(field_id=2, name="entity_hash", type=pa.int32(), required=True),
+            NestedField(
+                field_id=2, name="entity_hash", type=IntegerType(), required=True
+            ),
             NestedField(
                 field_id=3, name="event_ts", type=TimestampType(), required=True
             ),