We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Schema inference of BQ OfflineStore costly
## Expected Behavior 
When making schema inference, I expect there would be an extra minor cost when scanning tables.

## Current Behavior
BQ OfflineStore made a full scan on the entire table, although we are using `limit` statement. According to GCP documentation:
`Applying a LIMIT clause to a SELECT * query does not affect the amount of data read. You are billed for reading all bytes in the entire table, and the query counts against your free tier quota.`

https://cloud.google.com/bigquery/docs/best-practices-costs

## Steps to reproduce
1. Prepare a large feature table
2. feast apply as a feature view

### Specifications

- Version: 0.28.0
- Platform: Linux
- Subsystem:

## Possible Solution
We can add a filter by `timestamp_field` by modifying this [line](https://github.com/feast-dev/feast/blob/master/sdk/python/feast/infra/offline_stores/bigquery_source.py#L183) even though there is no data, the schema could be inferred eventually

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sdk/python/feast/infra/offline_stores/bigquery_source.py`
Content:
```
1 from typing import Callable, Dict, Iterable, List, Optional, Tuple
2 
3 from typeguard import typechecked
4 
5 from feast import type_map
6 from feast.data_source import DataSource
7 from feast.errors import DataSourceNoNameException, DataSourceNotFoundException
8 from feast.feature_logging import LoggingDestination
9 from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto
10 from feast.protos.feast.core.FeatureService_pb2 import (
11     LoggingConfig as LoggingConfigProto,
12 )
13 from feast.protos.feast.core.SavedDataset_pb2 import (
14     SavedDatasetStorage as SavedDatasetStorageProto,
15 )
16 from feast.repo_config import RepoConfig
17 from feast.saved_dataset import SavedDatasetStorage
18 from feast.usage import get_user_agent
19 from feast.value_type import ValueType
20 
21 
22 @typechecked
23 class BigQuerySource(DataSource):
24     def __init__(
25         self,
26         *,
27         name: Optional[str] = None,
28         timestamp_field: Optional[str] = None,
29         table: Optional[str] = None,
30         created_timestamp_column: Optional[str] = "",
31         field_mapping: Optional[Dict[str, str]] = None,
32         query: Optional[str] = None,
33         description: Optional[str] = "",
34         tags: Optional[Dict[str, str]] = None,
35         owner: Optional[str] = "",
36     ):
37         """Create a BigQuerySource from an existing table or query.
38 
39         Args:
40             name (optional): Name for the source. Defaults to the table if not specified, in which
41                 case the table must be specified.
42             timestamp_field (optional): Event timestamp field used for point in time
43                 joins of feature values.
44             table (optional): BigQuery table where the features are stored. Exactly one of 'table'
45                 and 'query' must be specified.
46             table (optional): The BigQuery table where features can be found.
47             created_timestamp_column (optional): Timestamp column when row was created, used for deduplicating rows.
48             field_mapping (optional): A dictionary mapping of column names in this data source to feature names in a feature table
49                 or view. Only used for feature columns, not entities or timestamp columns.
50             query (optional): The query to be executed to obtain the features. Exactly one of 'table'
51                 and 'query' must be specified.
52             description (optional): A human-readable description.
53             tags (optional): A dictionary of key-value pairs to store arbitrary metadata.
54             owner (optional): The owner of the bigquery source, typically the email of the primary
55                 maintainer.
56         Example:
57             >>> from feast import BigQuerySource
58             >>> my_bigquery_source = BigQuerySource(table="gcp_project:bq_dataset.bq_table")
59         """
60         if table is None and query is None:
61             raise ValueError('No "table" or "query" argument provided.')
62 
63         self.bigquery_options = BigQueryOptions(table=table, query=query)
64 
65         # If no name, use the table as the default name.
66         if name is None and table is None:
67             raise DataSourceNoNameException()
68         name = name or table
69         assert name
70 
71         super().__init__(
72             name=name,
73             timestamp_field=timestamp_field,
74             created_timestamp_column=created_timestamp_column,
75             field_mapping=field_mapping,
76             description=description,
77             tags=tags,
78             owner=owner,
79         )
80 
81     # Note: Python requires redefining hash in child classes that override __eq__
82     def __hash__(self):
83         return super().__hash__()
84 
85     def __eq__(self, other):
86         if not isinstance(other, BigQuerySource):
87             raise TypeError(
88                 "Comparisons should only involve BigQuerySource class objects."
89             )
90 
91         return (
92             super().__eq__(other)
93             and self.table == other.table
94             and self.query == other.query
95         )
96 
97     @property
98     def table(self):
99         return self.bigquery_options.table
100 
101     @property
102     def query(self):
103         return self.bigquery_options.query
104 
105     @staticmethod
106     def from_proto(data_source: DataSourceProto):
107         assert data_source.HasField("bigquery_options")
108 
109         return BigQuerySource(
110             name=data_source.name,
111             field_mapping=dict(data_source.field_mapping),
112             table=data_source.bigquery_options.table,
113             timestamp_field=data_source.timestamp_field,
114             created_timestamp_column=data_source.created_timestamp_column,
115             query=data_source.bigquery_options.query,
116             description=data_source.description,
117             tags=dict(data_source.tags),
118             owner=data_source.owner,
119         )
120 
121     def to_proto(self) -> DataSourceProto:
122         data_source_proto = DataSourceProto(
123             name=self.name,
124             type=DataSourceProto.BATCH_BIGQUERY,
125             field_mapping=self.field_mapping,
126             bigquery_options=self.bigquery_options.to_proto(),
127             description=self.description,
128             tags=self.tags,
129             owner=self.owner,
130             timestamp_field=self.timestamp_field,
131             created_timestamp_column=self.created_timestamp_column,
132         )
133 
134         return data_source_proto
135 
136     def validate(self, config: RepoConfig):
137         if not self.query:
138             from google.api_core.exceptions import NotFound
139             from google.cloud import bigquery
140 
141             client = bigquery.Client()
142             try:
143                 client.get_table(self.table)
144             except NotFound:
145                 raise DataSourceNotFoundException(self.table)
146 
147     def get_table_query_string(self) -> str:
148         """Returns a string that can directly be used to reference this table in SQL"""
149         if self.table:
150             return f"`{self.table}`"
151         else:
152             return f"({self.query})"
153 
154     @staticmethod
155     def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:
156         return type_map.bq_to_feast_value_type
157 
158     def get_table_column_names_and_types(
159         self, config: RepoConfig
160     ) -> Iterable[Tuple[str, str]]:
161         try:
162             from google.api_core import client_info as http_client_info
163         except ImportError as e:
164             from feast.errors import FeastExtrasDependencyImportError
165 
166             raise FeastExtrasDependencyImportError("gcp", str(e))
167 
168         from google.cloud import bigquery
169 
170         project_id = (
171             config.offline_store.billing_project_id or config.offline_store.project_id
172         )
173         client = bigquery.Client(
174             project=project_id,
175             location=config.offline_store.location,
176             client_info=http_client_info.ClientInfo(user_agent=get_user_agent()),
177         )
178         if self.table:
179             schema = client.get_table(self.table).schema
180             if not isinstance(schema[0], bigquery.schema.SchemaField):
181                 raise TypeError("Could not parse BigQuery table schema.")
182         else:
183             bq_columns_query = f"SELECT * FROM ({self.query}) LIMIT 1"
184             queryRes = client.query(bq_columns_query).result()
185             schema = queryRes.schema
186 
187         name_type_pairs: List[Tuple[str, str]] = []
188         for field in schema:
189             bq_type_as_str = field.field_type
190             if field.mode == "REPEATED":
191                 bq_type_as_str = "ARRAY<" + bq_type_as_str + ">"
192             name_type_pairs.append((field.name, bq_type_as_str))
193 
194         return name_type_pairs
195 
196 
197 class BigQueryOptions:
198     """
199     Configuration options for a BigQuery data source.
200     """
201 
202     def __init__(
203         self,
204         table: Optional[str],
205         query: Optional[str],
206     ):
207         self.table = table or ""
208         self.query = query or ""
209 
210     @classmethod
211     def from_proto(cls, bigquery_options_proto: DataSourceProto.BigQueryOptions):
212         """
213         Creates a BigQueryOptions from a protobuf representation of a BigQuery option
214 
215         Args:
216             bigquery_options_proto: A protobuf representation of a DataSource
217 
218         Returns:
219             Returns a BigQueryOptions object based on the bigquery_options protobuf
220         """
221         bigquery_options = cls(
222             table=bigquery_options_proto.table,
223             query=bigquery_options_proto.query,
224         )
225 
226         return bigquery_options
227 
228     def to_proto(self) -> DataSourceProto.BigQueryOptions:
229         """
230         Converts an BigQueryOptionsProto object to its protobuf representation.
231 
232         Returns:
233             BigQueryOptionsProto protobuf
234         """
235         bigquery_options_proto = DataSourceProto.BigQueryOptions(
236             table=self.table,
237             query=self.query,
238         )
239 
240         return bigquery_options_proto
241 
242 
243 class SavedDatasetBigQueryStorage(SavedDatasetStorage):
244     _proto_attr_name = "bigquery_storage"
245 
246     bigquery_options: BigQueryOptions
247 
248     def __init__(self, table: str):
249         self.bigquery_options = BigQueryOptions(table=table, query=None)
250 
251     @staticmethod
252     def from_proto(storage_proto: SavedDatasetStorageProto) -> SavedDatasetStorage:
253         return SavedDatasetBigQueryStorage(
254             table=BigQueryOptions.from_proto(storage_proto.bigquery_storage).table
255         )
256 
257     def to_proto(self) -> SavedDatasetStorageProto:
258         return SavedDatasetStorageProto(
259             bigquery_storage=self.bigquery_options.to_proto()
260         )
261 
262     def to_data_source(self) -> DataSource:
263         return BigQuerySource(table=self.bigquery_options.table)
264 
265 
266 class BigQueryLoggingDestination(LoggingDestination):
267     _proto_kind = "bigquery_destination"
268 
269     table: str
270 
271     def __init__(self, *, table_ref):
272         self.table = table_ref
273 
274     @classmethod
275     def from_proto(cls, config_proto: LoggingConfigProto) -> "LoggingDestination":
276         return BigQueryLoggingDestination(
277             table_ref=config_proto.bigquery_destination.table_ref,
278         )
279 
280     def to_data_source(self) -> DataSource:
281         return BigQuerySource(table=self.table)
282 
283     def to_proto(self) -> LoggingConfigProto:
284         return LoggingConfigProto(
285             bigquery_destination=LoggingConfigProto.BigQueryDestination(
286                 table_ref=self.table
287             )
288         )
289
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/sdk/python/feast/infra/offline_stores/bigquery_source.py b/sdk/python/feast/infra/offline_stores/bigquery_source.py
--- a/sdk/python/feast/infra/offline_stores/bigquery_source.py
+++ b/sdk/python/feast/infra/offline_stores/bigquery_source.py
@@ -180,9 +180,9 @@
             if not isinstance(schema[0], bigquery.schema.SchemaField):
                 raise TypeError("Could not parse BigQuery table schema.")
         else:
-            bq_columns_query = f"SELECT * FROM ({self.query}) LIMIT 1"
-            queryRes = client.query(bq_columns_query).result()
-            schema = queryRes.schema
+            bq_columns_query = f"SELECT * FROM ({self.query}) LIMIT 0"
+            query_res = client.query(bq_columns_query).result()
+            schema = query_res.schema
 
         name_type_pairs: List[Tuple[str, str]] = []
         for field in schema:

{"golden_diff": "diff --git a/sdk/python/feast/infra/offline_stores/bigquery_source.py b/sdk/python/feast/infra/offline_stores/bigquery_source.py\n--- a/sdk/python/feast/infra/offline_stores/bigquery_source.py\n+++ b/sdk/python/feast/infra/offline_stores/bigquery_source.py\n@@ -180,9 +180,9 @@\n             if not isinstance(schema[0], bigquery.schema.SchemaField):\n                 raise TypeError(\"Could not parse BigQuery table schema.\")\n         else:\n-            bq_columns_query = f\"SELECT * FROM ({self.query}) LIMIT 1\"\n-            queryRes = client.query(bq_columns_query).result()\n-            schema = queryRes.schema\n+            bq_columns_query = f\"SELECT * FROM ({self.query}) LIMIT 0\"\n+            query_res = client.query(bq_columns_query).result()\n+            schema = query_res.schema\n \n         name_type_pairs: List[Tuple[str, str]] = []\n         for field in schema:\n", "issue": "Schema inference of BQ OfflineStore costly\n## Expected Behavior \r\nWhen making schema inference, I expect there would be an extra minor cost when scanning tables.\r\n\r\n## Current Behavior\r\nBQ OfflineStore made a full scan on the entire table, although we are using `limit` statement. According to GCP documentation:\r\n`Applying a LIMIT clause to a SELECT * query does not affect the amount of data read. You are billed for reading all bytes in the entire table, and the query counts against your free tier quota.`\r\n\r\nhttps://cloud.google.com/bigquery/docs/best-practices-costs\r\n\r\n## Steps to reproduce\r\n1. Prepare a large feature table\r\n2. feast apply as a feature view\r\n\r\n### Specifications\r\n\r\n- Version: 0.28.0\r\n- Platform: Linux\r\n- Subsystem:\r\n\r\n## Possible Solution\r\nWe can add a filter by `timestamp_field` by modifying this [line](https://github.com/feast-dev/feast/blob/master/sdk/python/feast/infra/offline_stores/bigquery_source.py#L183) even though there is no data, the schema could be inferred eventually\n", "before_files": [{"content": "from typing import Callable, Dict, Iterable, List, Optional, Tuple\n\nfrom typeguard import typechecked\n\nfrom feast import type_map\nfrom feast.data_source import DataSource\nfrom feast.errors import DataSourceNoNameException, DataSourceNotFoundException\nfrom feast.feature_logging import LoggingDestination\nfrom feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto\nfrom feast.protos.feast.core.FeatureService_pb2 import (\n    LoggingConfig as LoggingConfigProto,\n)\nfrom feast.protos.feast.core.SavedDataset_pb2 import (\n    SavedDatasetStorage as SavedDatasetStorageProto,\n)\nfrom feast.repo_config import RepoConfig\nfrom feast.saved_dataset import SavedDatasetStorage\nfrom feast.usage import get_user_agent\nfrom feast.value_type import ValueType\n\n\n@typechecked\nclass BigQuerySource(DataSource):\n    def __init__(\n        self,\n        *,\n        name: Optional[str] = None,\n        timestamp_field: Optional[str] = None,\n        table: Optional[str] = None,\n        created_timestamp_column: Optional[str] = \"\",\n        field_mapping: Optional[Dict[str, str]] = None,\n        query: Optional[str] = None,\n        description: Optional[str] = \"\",\n        tags: Optional[Dict[str, str]] = None,\n        owner: Optional[str] = \"\",\n    ):\n        \"\"\"Create a BigQuerySource from an existing table or query.\n\n        Args:\n            name (optional): Name for the source. Defaults to the table if not specified, in which\n                case the table must be specified.\n            timestamp_field (optional): Event timestamp field used for point in time\n                joins of feature values.\n            table (optional): BigQuery table where the features are stored. Exactly one of 'table'\n                and 'query' must be specified.\n            table (optional): The BigQuery table where features can be found.\n            created_timestamp_column (optional): Timestamp column when row was created, used for deduplicating rows.\n            field_mapping (optional): A dictionary mapping of column names in this data source to feature names in a feature table\n                or view. Only used for feature columns, not entities or timestamp columns.\n            query (optional): The query to be executed to obtain the features. Exactly one of 'table'\n                and 'query' must be specified.\n            description (optional): A human-readable description.\n            tags (optional): A dictionary of key-value pairs to store arbitrary metadata.\n            owner (optional): The owner of the bigquery source, typically the email of the primary\n                maintainer.\n        Example:\n            >>> from feast import BigQuerySource\n            >>> my_bigquery_source = BigQuerySource(table=\"gcp_project:bq_dataset.bq_table\")\n        \"\"\"\n        if table is None and query is None:\n            raise ValueError('No \"table\" or \"query\" argument provided.')\n\n        self.bigquery_options = BigQueryOptions(table=table, query=query)\n\n        # If no name, use the table as the default name.\n        if name is None and table is None:\n            raise DataSourceNoNameException()\n        name = name or table\n        assert name\n\n        super().__init__(\n            name=name,\n            timestamp_field=timestamp_field,\n            created_timestamp_column=created_timestamp_column,\n            field_mapping=field_mapping,\n            description=description,\n            tags=tags,\n            owner=owner,\n        )\n\n    # Note: Python requires redefining hash in child classes that override __eq__\n    def __hash__(self):\n        return super().__hash__()\n\n    def __eq__(self, other):\n        if not isinstance(other, BigQuerySource):\n            raise TypeError(\n                \"Comparisons should only involve BigQuerySource class objects.\"\n            )\n\n        return (\n            super().__eq__(other)\n            and self.table == other.table\n            and self.query == other.query\n        )\n\n    @property\n    def table(self):\n        return self.bigquery_options.table\n\n    @property\n    def query(self):\n        return self.bigquery_options.query\n\n    @staticmethod\n    def from_proto(data_source: DataSourceProto):\n        assert data_source.HasField(\"bigquery_options\")\n\n        return BigQuerySource(\n            name=data_source.name,\n            field_mapping=dict(data_source.field_mapping),\n            table=data_source.bigquery_options.table,\n            timestamp_field=data_source.timestamp_field,\n            created_timestamp_column=data_source.created_timestamp_column,\n            query=data_source.bigquery_options.query,\n            description=data_source.description,\n            tags=dict(data_source.tags),\n            owner=data_source.owner,\n        )\n\n    def to_proto(self) -> DataSourceProto:\n        data_source_proto = DataSourceProto(\n            name=self.name,\n            type=DataSourceProto.BATCH_BIGQUERY,\n            field_mapping=self.field_mapping,\n            bigquery_options=self.bigquery_options.to_proto(),\n            description=self.description,\n            tags=self.tags,\n            owner=self.owner,\n            timestamp_field=self.timestamp_field,\n            created_timestamp_column=self.created_timestamp_column,\n        )\n\n        return data_source_proto\n\n    def validate(self, config: RepoConfig):\n        if not self.query:\n            from google.api_core.exceptions import NotFound\n            from google.cloud import bigquery\n\n            client = bigquery.Client()\n            try:\n                client.get_table(self.table)\n            except NotFound:\n                raise DataSourceNotFoundException(self.table)\n\n    def get_table_query_string(self) -> str:\n        \"\"\"Returns a string that can directly be used to reference this table in SQL\"\"\"\n        if self.table:\n            return f\"`{self.table}`\"\n        else:\n            return f\"({self.query})\"\n\n    @staticmethod\n    def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:\n        return type_map.bq_to_feast_value_type\n\n    def get_table_column_names_and_types(\n        self, config: RepoConfig\n    ) -> Iterable[Tuple[str, str]]:\n        try:\n            from google.api_core import client_info as http_client_info\n        except ImportError as e:\n            from feast.errors import FeastExtrasDependencyImportError\n\n            raise FeastExtrasDependencyImportError(\"gcp\", str(e))\n\n        from google.cloud import bigquery\n\n        project_id = (\n            config.offline_store.billing_project_id or config.offline_store.project_id\n        )\n        client = bigquery.Client(\n            project=project_id,\n            location=config.offline_store.location,\n            client_info=http_client_info.ClientInfo(user_agent=get_user_agent()),\n        )\n        if self.table:\n            schema = client.get_table(self.table).schema\n            if not isinstance(schema[0], bigquery.schema.SchemaField):\n                raise TypeError(\"Could not parse BigQuery table schema.\")\n        else:\n            bq_columns_query = f\"SELECT * FROM ({self.query}) LIMIT 1\"\n            queryRes = client.query(bq_columns_query).result()\n            schema = queryRes.schema\n\n        name_type_pairs: List[Tuple[str, str]] = []\n        for field in schema:\n            bq_type_as_str = field.field_type\n            if field.mode == \"REPEATED\":\n                bq_type_as_str = \"ARRAY<\" + bq_type_as_str + \">\"\n            name_type_pairs.append((field.name, bq_type_as_str))\n\n        return name_type_pairs\n\n\nclass BigQueryOptions:\n    \"\"\"\n    Configuration options for a BigQuery data source.\n    \"\"\"\n\n    def __init__(\n        self,\n        table: Optional[str],\n        query: Optional[str],\n    ):\n        self.table = table or \"\"\n        self.query = query or \"\"\n\n    @classmethod\n    def from_proto(cls, bigquery_options_proto: DataSourceProto.BigQueryOptions):\n        \"\"\"\n        Creates a BigQueryOptions from a protobuf representation of a BigQuery option\n\n        Args:\n            bigquery_options_proto: A protobuf representation of a DataSource\n\n        Returns:\n            Returns a BigQueryOptions object based on the bigquery_options protobuf\n        \"\"\"\n        bigquery_options = cls(\n            table=bigquery_options_proto.table,\n            query=bigquery_options_proto.query,\n        )\n\n        return bigquery_options\n\n    def to_proto(self) -> DataSourceProto.BigQueryOptions:\n        \"\"\"\n        Converts an BigQueryOptionsProto object to its protobuf representation.\n\n        Returns:\n            BigQueryOptionsProto protobuf\n        \"\"\"\n        bigquery_options_proto = DataSourceProto.BigQueryOptions(\n            table=self.table,\n            query=self.query,\n        )\n\n        return bigquery_options_proto\n\n\nclass SavedDatasetBigQueryStorage(SavedDatasetStorage):\n    _proto_attr_name = \"bigquery_storage\"\n\n    bigquery_options: BigQueryOptions\n\n    def __init__(self, table: str):\n        self.bigquery_options = BigQueryOptions(table=table, query=None)\n\n    @staticmethod\n    def from_proto(storage_proto: SavedDatasetStorageProto) -> SavedDatasetStorage:\n        return SavedDatasetBigQueryStorage(\n            table=BigQueryOptions.from_proto(storage_proto.bigquery_storage).table\n        )\n\n    def to_proto(self) -> SavedDatasetStorageProto:\n        return SavedDatasetStorageProto(\n            bigquery_storage=self.bigquery_options.to_proto()\n        )\n\n    def to_data_source(self) -> DataSource:\n        return BigQuerySource(table=self.bigquery_options.table)\n\n\nclass BigQueryLoggingDestination(LoggingDestination):\n    _proto_kind = \"bigquery_destination\"\n\n    table: str\n\n    def __init__(self, *, table_ref):\n        self.table = table_ref\n\n    @classmethod\n    def from_proto(cls, config_proto: LoggingConfigProto) -> \"LoggingDestination\":\n        return BigQueryLoggingDestination(\n            table_ref=config_proto.bigquery_destination.table_ref,\n        )\n\n    def to_data_source(self) -> DataSource:\n        return BigQuerySource(table=self.table)\n\n    def to_proto(self) -> LoggingConfigProto:\n        return LoggingConfigProto(\n            bigquery_destination=LoggingConfigProto.BigQueryDestination(\n                table_ref=self.table\n            )\n        )\n", "path": "sdk/python/feast/infra/offline_stores/bigquery_source.py"}], "after_files": [{"content": "from typing import Callable, Dict, Iterable, List, Optional, Tuple\n\nfrom typeguard import typechecked\n\nfrom feast import type_map\nfrom feast.data_source import DataSource\nfrom feast.errors import DataSourceNoNameException, DataSourceNotFoundException\nfrom feast.feature_logging import LoggingDestination\nfrom feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto\nfrom feast.protos.feast.core.FeatureService_pb2 import (\n    LoggingConfig as LoggingConfigProto,\n)\nfrom feast.protos.feast.core.SavedDataset_pb2 import (\n    SavedDatasetStorage as SavedDatasetStorageProto,\n)\nfrom feast.repo_config import RepoConfig\nfrom feast.saved_dataset import SavedDatasetStorage\nfrom feast.usage import get_user_agent\nfrom feast.value_type import ValueType\n\n\n@typechecked\nclass BigQuerySource(DataSource):\n    def __init__(\n        self,\n        *,\n        name: Optional[str] = None,\n        timestamp_field: Optional[str] = None,\n        table: Optional[str] = None,\n        created_timestamp_column: Optional[str] = \"\",\n        field_mapping: Optional[Dict[str, str]] = None,\n        query: Optional[str] = None,\n        description: Optional[str] = \"\",\n        tags: Optional[Dict[str, str]] = None,\n        owner: Optional[str] = \"\",\n    ):\n        \"\"\"Create a BigQuerySource from an existing table or query.\n\n        Args:\n            name (optional): Name for the source. Defaults to the table if not specified, in which\n                case the table must be specified.\n            timestamp_field (optional): Event timestamp field used for point in time\n                joins of feature values.\n            table (optional): BigQuery table where the features are stored. Exactly one of 'table'\n                and 'query' must be specified.\n            table (optional): The BigQuery table where features can be found.\n            created_timestamp_column (optional): Timestamp column when row was created, used for deduplicating rows.\n            field_mapping (optional): A dictionary mapping of column names in this data source to feature names in a feature table\n                or view. Only used for feature columns, not entities or timestamp columns.\n            query (optional): The query to be executed to obtain the features. Exactly one of 'table'\n                and 'query' must be specified.\n            description (optional): A human-readable description.\n            tags (optional): A dictionary of key-value pairs to store arbitrary metadata.\n            owner (optional): The owner of the bigquery source, typically the email of the primary\n                maintainer.\n        Example:\n            >>> from feast import BigQuerySource\n            >>> my_bigquery_source = BigQuerySource(table=\"gcp_project:bq_dataset.bq_table\")\n        \"\"\"\n        if table is None and query is None:\n            raise ValueError('No \"table\" or \"query\" argument provided.')\n\n        self.bigquery_options = BigQueryOptions(table=table, query=query)\n\n        # If no name, use the table as the default name.\n        if name is None and table is None:\n            raise DataSourceNoNameException()\n        name = name or table\n        assert name\n\n        super().__init__(\n            name=name,\n            timestamp_field=timestamp_field,\n            created_timestamp_column=created_timestamp_column,\n            field_mapping=field_mapping,\n            description=description,\n            tags=tags,\n            owner=owner,\n        )\n\n    # Note: Python requires redefining hash in child classes that override __eq__\n    def __hash__(self):\n        return super().__hash__()\n\n    def __eq__(self, other):\n        if not isinstance(other, BigQuerySource):\n            raise TypeError(\n                \"Comparisons should only involve BigQuerySource class objects.\"\n            )\n\n        return (\n            super().__eq__(other)\n            and self.table == other.table\n            and self.query == other.query\n        )\n\n    @property\n    def table(self):\n        return self.bigquery_options.table\n\n    @property\n    def query(self):\n        return self.bigquery_options.query\n\n    @staticmethod\n    def from_proto(data_source: DataSourceProto):\n        assert data_source.HasField(\"bigquery_options\")\n\n        return BigQuerySource(\n            name=data_source.name,\n            field_mapping=dict(data_source.field_mapping),\n            table=data_source.bigquery_options.table,\n            timestamp_field=data_source.timestamp_field,\n            created_timestamp_column=data_source.created_timestamp_column,\n            query=data_source.bigquery_options.query,\n            description=data_source.description,\n            tags=dict(data_source.tags),\n            owner=data_source.owner,\n        )\n\n    def to_proto(self) -> DataSourceProto:\n        data_source_proto = DataSourceProto(\n            name=self.name,\n            type=DataSourceProto.BATCH_BIGQUERY,\n            field_mapping=self.field_mapping,\n            bigquery_options=self.bigquery_options.to_proto(),\n            description=self.description,\n            tags=self.tags,\n            owner=self.owner,\n            timestamp_field=self.timestamp_field,\n            created_timestamp_column=self.created_timestamp_column,\n        )\n\n        return data_source_proto\n\n    def validate(self, config: RepoConfig):\n        if not self.query:\n            from google.api_core.exceptions import NotFound\n            from google.cloud import bigquery\n\n            client = bigquery.Client()\n            try:\n                client.get_table(self.table)\n            except NotFound:\n                raise DataSourceNotFoundException(self.table)\n\n    def get_table_query_string(self) -> str:\n        \"\"\"Returns a string that can directly be used to reference this table in SQL\"\"\"\n        if self.table:\n            return f\"`{self.table}`\"\n        else:\n            return f\"({self.query})\"\n\n    @staticmethod\n    def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:\n        return type_map.bq_to_feast_value_type\n\n    def get_table_column_names_and_types(\n        self, config: RepoConfig\n    ) -> Iterable[Tuple[str, str]]:\n        try:\n            from google.api_core import client_info as http_client_info\n        except ImportError as e:\n            from feast.errors import FeastExtrasDependencyImportError\n\n            raise FeastExtrasDependencyImportError(\"gcp\", str(e))\n\n        from google.cloud import bigquery\n\n        project_id = (\n            config.offline_store.billing_project_id or config.offline_store.project_id\n        )\n        client = bigquery.Client(\n            project=project_id,\n            location=config.offline_store.location,\n            client_info=http_client_info.ClientInfo(user_agent=get_user_agent()),\n        )\n        if self.table:\n            schema = client.get_table(self.table).schema\n            if not isinstance(schema[0], bigquery.schema.SchemaField):\n                raise TypeError(\"Could not parse BigQuery table schema.\")\n        else:\n            bq_columns_query = f\"SELECT * FROM ({self.query}) LIMIT 0\"\n            query_res = client.query(bq_columns_query).result()\n            schema = query_res.schema\n\n        name_type_pairs: List[Tuple[str, str]] = []\n        for field in schema:\n            bq_type_as_str = field.field_type\n            if field.mode == \"REPEATED\":\n                bq_type_as_str = \"ARRAY<\" + bq_type_as_str + \">\"\n            name_type_pairs.append((field.name, bq_type_as_str))\n\n        return name_type_pairs\n\n\nclass BigQueryOptions:\n    \"\"\"\n    Configuration options for a BigQuery data source.\n    \"\"\"\n\n    def __init__(\n        self,\n        table: Optional[str],\n        query: Optional[str],\n    ):\n        self.table = table or \"\"\n        self.query = query or \"\"\n\n    @classmethod\n    def from_proto(cls, bigquery_options_proto: DataSourceProto.BigQueryOptions):\n        \"\"\"\n        Creates a BigQueryOptions from a protobuf representation of a BigQuery option\n\n        Args:\n            bigquery_options_proto: A protobuf representation of a DataSource\n\n        Returns:\n            Returns a BigQueryOptions object based on the bigquery_options protobuf\n        \"\"\"\n        bigquery_options = cls(\n            table=bigquery_options_proto.table,\n            query=bigquery_options_proto.query,\n        )\n\n        return bigquery_options\n\n    def to_proto(self) -> DataSourceProto.BigQueryOptions:\n        \"\"\"\n        Converts an BigQueryOptionsProto object to its protobuf representation.\n\n        Returns:\n            BigQueryOptionsProto protobuf\n        \"\"\"\n        bigquery_options_proto = DataSourceProto.BigQueryOptions(\n            table=self.table,\n            query=self.query,\n        )\n\n        return bigquery_options_proto\n\n\nclass SavedDatasetBigQueryStorage(SavedDatasetStorage):\n    _proto_attr_name = \"bigquery_storage\"\n\n    bigquery_options: BigQueryOptions\n\n    def __init__(self, table: str):\n        self.bigquery_options = BigQueryOptions(table=table, query=None)\n\n    @staticmethod\n    def from_proto(storage_proto: SavedDatasetStorageProto) -> SavedDatasetStorage:\n        return SavedDatasetBigQueryStorage(\n            table=BigQueryOptions.from_proto(storage_proto.bigquery_storage).table\n        )\n\n    def to_proto(self) -> SavedDatasetStorageProto:\n        return SavedDatasetStorageProto(\n            bigquery_storage=self.bigquery_options.to_proto()\n        )\n\n    def to_data_source(self) -> DataSource:\n        return BigQuerySource(table=self.bigquery_options.table)\n\n\nclass BigQueryLoggingDestination(LoggingDestination):\n    _proto_kind = \"bigquery_destination\"\n\n    table: str\n\n    def __init__(self, *, table_ref):\n        self.table = table_ref\n\n    @classmethod\n    def from_proto(cls, config_proto: LoggingConfigProto) -> \"LoggingDestination\":\n        return BigQueryLoggingDestination(\n            table_ref=config_proto.bigquery_destination.table_ref,\n        )\n\n    def to_data_source(self) -> DataSource:\n        return BigQuerySource(table=self.table)\n\n    def to_proto(self) -> LoggingConfigProto:\n        return LoggingConfigProto(\n            bigquery_destination=LoggingConfigProto.BigQueryDestination(\n                table_ref=self.table\n            )\n        )\n", "path": "sdk/python/feast/infra/offline_stores/bigquery_source.py"}]}