|
| 1 | +"""Tests for RetrievalJob FeastDataFrame integration.""" |
| 2 | + |
| 3 | +from unittest.mock import Mock |
| 4 | + |
| 5 | +import pandas as pd |
| 6 | +import pyarrow as pa |
| 7 | + |
| 8 | +from feast.dataframe import DataFrameEngine, FeastDataFrame |
| 9 | +from feast.infra.offline_stores.offline_store import RetrievalJob |
| 10 | + |
| 11 | + |
| 12 | +class MockRetrievalJob(RetrievalJob): |
| 13 | + """Mock RetrievalJob for testing.""" |
| 14 | + |
| 15 | + def __init__( |
| 16 | + self, arrow_table: pa.Table, features: list = None, odfvs: list = None |
| 17 | + ): |
| 18 | + self.arrow_table = arrow_table |
| 19 | + self.features = features or [] |
| 20 | + self.odfvs = odfvs or [] |
| 21 | + |
| 22 | + def _to_arrow_internal(self, timeout=None): |
| 23 | + return self.arrow_table |
| 24 | + |
| 25 | + @property |
| 26 | + def full_feature_names(self): |
| 27 | + return False |
| 28 | + |
| 29 | + @property |
| 30 | + def on_demand_feature_views(self): |
| 31 | + return self.odfvs |
| 32 | + |
| 33 | + |
| 34 | +class TestRetrievalJobFeastDataFrame: |
| 35 | + """Test RetrievalJob FeastDataFrame integration.""" |
| 36 | + |
| 37 | + def test_to_feast_df_basic(self): |
| 38 | + """Test basic to_feast_df functionality.""" |
| 39 | + # Create test data |
| 40 | + test_data = pa.table( |
| 41 | + { |
| 42 | + "feature1": [1, 2, 3], |
| 43 | + "feature2": ["a", "b", "c"], |
| 44 | + "timestamp": pd.to_datetime(["2023-01-01", "2023-01-02", "2023-01-03"]), |
| 45 | + } |
| 46 | + ) |
| 47 | + |
| 48 | + # Create mock retrieval job |
| 49 | + job = MockRetrievalJob(test_data, features=["feature1", "feature2"]) |
| 50 | + |
| 51 | + # Test to_feast_df |
| 52 | + feast_df = job.to_feast_df() |
| 53 | + |
| 54 | + # Assertions |
| 55 | + assert isinstance(feast_df, FeastDataFrame) |
| 56 | + assert feast_df.engine == DataFrameEngine.ARROW |
| 57 | + assert isinstance(feast_df.data, pa.Table) |
| 58 | + assert feast_df.data.num_rows == 3 |
| 59 | + assert feast_df.data.num_columns == 3 |
| 60 | + |
| 61 | + def test_to_feast_df_metadata(self): |
| 62 | + """Test to_feast_df metadata population.""" |
| 63 | + # Create test data |
| 64 | + test_data = pa.table({"feature1": [1, 2, 3], "feature2": [4.0, 5.0, 6.0]}) |
| 65 | + |
| 66 | + # Create mock on-demand feature views |
| 67 | + mock_odfv1 = Mock() |
| 68 | + mock_odfv1.name = "odfv1" |
| 69 | + # Mock transform_arrow to return an empty table (no new columns added) |
| 70 | + mock_odfv1.transform_arrow.return_value = pa.table({}) |
| 71 | + |
| 72 | + mock_odfv2 = Mock() |
| 73 | + mock_odfv2.name = "odfv2" |
| 74 | + # Mock transform_arrow to return an empty table (no new columns added) |
| 75 | + mock_odfv2.transform_arrow.return_value = pa.table({}) |
| 76 | + |
| 77 | + # Create mock retrieval job with features and ODFVs |
| 78 | + job = MockRetrievalJob( |
| 79 | + test_data, features=["feature1", "feature2"], odfvs=[mock_odfv1, mock_odfv2] |
| 80 | + ) |
| 81 | + |
| 82 | + # Test to_feast_df |
| 83 | + feast_df = job.to_feast_df() |
| 84 | + |
| 85 | + # Check metadata |
| 86 | + assert "features" in feast_df.metadata |
| 87 | + assert "on_demand_feature_views" in feast_df.metadata |
| 88 | + assert feast_df.metadata["features"] == ["feature1", "feature2"] |
| 89 | + assert feast_df.metadata["on_demand_feature_views"] == ["odfv1", "odfv2"] |
| 90 | + |
| 91 | + def test_to_feast_df_with_timeout(self): |
| 92 | + """Test to_feast_df with timeout parameter.""" |
| 93 | + test_data = pa.table({"feature1": [1, 2, 3]}) |
| 94 | + job = MockRetrievalJob(test_data) |
| 95 | + |
| 96 | + # Test with timeout - should not raise any errors |
| 97 | + feast_df = job.to_feast_df(timeout=30) |
| 98 | + |
| 99 | + assert isinstance(feast_df, FeastDataFrame) |
| 100 | + assert feast_df.engine == DataFrameEngine.ARROW |
| 101 | + |
| 102 | + def test_to_feast_df_empty_metadata(self): |
| 103 | + """Test to_feast_df with empty features and ODFVs.""" |
| 104 | + test_data = pa.table({"feature1": [1, 2, 3]}) |
| 105 | + job = MockRetrievalJob(test_data) # No features or ODFVs provided |
| 106 | + |
| 107 | + feast_df = job.to_feast_df() |
| 108 | + |
| 109 | + # Should handle missing features gracefully |
| 110 | + assert feast_df.metadata["features"] == [] |
| 111 | + assert feast_df.metadata["on_demand_feature_views"] == [] |
| 112 | + |
| 113 | + def test_to_feast_df_preserves_arrow_data(self): |
| 114 | + """Test that to_feast_df preserves the original Arrow data.""" |
| 115 | + # Create test data with specific types |
| 116 | + test_data = pa.table( |
| 117 | + { |
| 118 | + "int_feature": pa.array([1, 2, 3], type=pa.int64()), |
| 119 | + "float_feature": pa.array([1.1, 2.2, 3.3], type=pa.float64()), |
| 120 | + "string_feature": pa.array(["a", "b", "c"], type=pa.string()), |
| 121 | + "bool_feature": pa.array([True, False, True], type=pa.bool_()), |
| 122 | + } |
| 123 | + ) |
| 124 | + |
| 125 | + job = MockRetrievalJob(test_data) |
| 126 | + feast_df = job.to_feast_df() |
| 127 | + |
| 128 | + # Check that the Arrow data is exactly the same |
| 129 | + assert feast_df.data.equals(test_data) |
| 130 | + assert feast_df.data.schema == test_data.schema |
| 131 | + |
| 132 | + # Check column names and types are preserved |
| 133 | + assert feast_df.data.column_names == test_data.column_names |
| 134 | + for i, column in enumerate(test_data.schema): |
| 135 | + assert feast_df.data.schema.field(i).type == column.type |
| 136 | + |
| 137 | + def test_to_df_still_works(self): |
| 138 | + """Test that the original to_df method still works unchanged.""" |
| 139 | + test_data = pa.table({"feature1": [1, 2, 3], "feature2": ["a", "b", "c"]}) |
| 140 | + |
| 141 | + job = MockRetrievalJob(test_data) |
| 142 | + |
| 143 | + # Test to_df returns pandas DataFrame |
| 144 | + df = job.to_df() |
| 145 | + |
| 146 | + assert isinstance(df, pd.DataFrame) |
| 147 | + assert len(df) == 3 |
| 148 | + assert list(df.columns) == ["feature1", "feature2"] |
| 149 | + assert df["feature1"].tolist() == [1, 2, 3] |
| 150 | + assert df["feature2"].tolist() == ["a", "b", "c"] |
| 151 | + |
| 152 | + def test_both_methods_return_same_data(self): |
| 153 | + """Test that to_df and to_feast_df return equivalent data.""" |
| 154 | + test_data = pa.table( |
| 155 | + {"feature1": [1, 2, 3, 4], "feature2": [10.5, 20.5, 30.5, 40.5]} |
| 156 | + ) |
| 157 | + |
| 158 | + job = MockRetrievalJob(test_data) |
| 159 | + |
| 160 | + # Get data from both methods |
| 161 | + df = job.to_df() |
| 162 | + feast_df = job.to_feast_df() |
| 163 | + |
| 164 | + # Convert FeastDataFrame to pandas for comparison |
| 165 | + feast_as_pandas = feast_df.data.to_pandas().reset_index(drop=True) |
| 166 | + |
| 167 | + # Should be equivalent |
| 168 | + pd.testing.assert_frame_equal(df, feast_as_pandas) |
0 commit comments