We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
spark source complain about "table or view not found" error.
## Expected Behavior 
spark offline store

## Current Behavior
if set table of SparkSource with pattern "db.table"，feast will complain about "table or view not found" error.

## Steps to reproduce
1. feast init test_repo
2. modify example.py to use SparkSource, and set table of SparkSource with pattern "db.table"
3. configure featue_store.yaml
4. feast apply

### Specifications

- Version: 0.20.1
- Platform: ubuntu 18/04
- Subsystem:

## Possible Solution


--- 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/contrib/spark_offline_store/spark_source.py`
Content:
```
1 import logging
2 import traceback
3 import warnings
4 from enum import Enum
5 from typing import Any, Callable, Dict, Iterable, Optional, Tuple
6 
7 from pyspark.sql import SparkSession
8 
9 from feast.data_source import DataSource
10 from feast.errors import DataSourceNoNameException
11 from feast.infra.offline_stores.offline_utils import get_temp_entity_table_name
12 from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto
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.type_map import spark_to_feast_value_type
19 from feast.value_type import ValueType
20 
21 logger = logging.getLogger(__name__)
22 
23 
24 class SparkSourceFormat(Enum):
25     csv = "csv"
26     json = "json"
27     parquet = "parquet"
28     delta = "delta"
29     avro = "avro"
30 
31 
32 class SparkSource(DataSource):
33     def __init__(
34         self,
35         *,
36         name: Optional[str] = None,
37         table: Optional[str] = None,
38         query: Optional[str] = None,
39         path: Optional[str] = None,
40         file_format: Optional[str] = None,
41         event_timestamp_column: Optional[str] = None,
42         created_timestamp_column: Optional[str] = None,
43         field_mapping: Optional[Dict[str, str]] = None,
44         date_partition_column: Optional[str] = None,
45         description: Optional[str] = "",
46         tags: Optional[Dict[str, str]] = None,
47         owner: Optional[str] = "",
48         timestamp_field: Optional[str] = None,
49     ):
50         # If no name, use the table_ref as the default name
51         _name = name
52         if not _name:
53             if table:
54                 _name = table
55             else:
56                 raise DataSourceNoNameException()
57 
58         if date_partition_column:
59             warnings.warn(
60                 (
61                     "The argument 'date_partition_column' is not supported for Spark sources."
62                     "It will be removed in Feast 0.24+"
63                 ),
64                 DeprecationWarning,
65             )
66 
67         super().__init__(
68             name=_name,
69             event_timestamp_column=event_timestamp_column,
70             created_timestamp_column=created_timestamp_column,
71             field_mapping=field_mapping,
72             description=description,
73             tags=tags,
74             owner=owner,
75             timestamp_field=timestamp_field,
76         )
77         warnings.warn(
78             "The spark data source API is an experimental feature in alpha development. "
79             "This API is unstable and it could and most probably will be changed in the future.",
80             RuntimeWarning,
81         )
82         self.spark_options = SparkOptions(
83             table=table,
84             query=query,
85             path=path,
86             file_format=file_format,
87         )
88 
89     @property
90     def table(self):
91         """
92         Returns the table of this feature data source
93         """
94         return self.spark_options.table
95 
96     @property
97     def query(self):
98         """
99         Returns the query of this feature data source
100         """
101         return self.spark_options.query
102 
103     @property
104     def path(self):
105         """
106         Returns the path of the spark data source file.
107         """
108         return self.spark_options.path
109 
110     @property
111     def file_format(self):
112         """
113         Returns the file format of this feature data source.
114         """
115         return self.spark_options.file_format
116 
117     @staticmethod
118     def from_proto(data_source: DataSourceProto) -> Any:
119         assert data_source.HasField("spark_options")
120         spark_options = SparkOptions.from_proto(data_source.spark_options)
121 
122         return SparkSource(
123             name=data_source.name,
124             field_mapping=dict(data_source.field_mapping),
125             table=spark_options.table,
126             query=spark_options.query,
127             path=spark_options.path,
128             file_format=spark_options.file_format,
129             timestamp_field=data_source.timestamp_field,
130             created_timestamp_column=data_source.created_timestamp_column,
131             description=data_source.description,
132             tags=dict(data_source.tags),
133             owner=data_source.owner,
134         )
135 
136     def to_proto(self) -> DataSourceProto:
137         data_source_proto = DataSourceProto(
138             name=self.name,
139             type=DataSourceProto.BATCH_SPARK,
140             data_source_class_type="feast.infra.offline_stores.contrib.spark_offline_store.spark_source.SparkSource",
141             field_mapping=self.field_mapping,
142             spark_options=self.spark_options.to_proto(),
143             description=self.description,
144             tags=self.tags,
145             owner=self.owner,
146         )
147 
148         data_source_proto.timestamp_field = self.timestamp_field
149         data_source_proto.created_timestamp_column = self.created_timestamp_column
150 
151         return data_source_proto
152 
153     def validate(self, config: RepoConfig):
154         self.get_table_column_names_and_types(config)
155 
156     @staticmethod
157     def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:
158         return spark_to_feast_value_type
159 
160     def get_table_column_names_and_types(
161         self, config: RepoConfig
162     ) -> Iterable[Tuple[str, str]]:
163         from feast.infra.offline_stores.contrib.spark_offline_store.spark import (
164             get_spark_session_or_start_new_with_repoconfig,
165         )
166 
167         spark_session = get_spark_session_or_start_new_with_repoconfig(
168             store_config=config.offline_store
169         )
170         df = spark_session.sql(f"SELECT * FROM {self.get_table_query_string()}")
171         return (
172             (fields["name"], fields["type"])
173             for fields in df.schema.jsonValue()["fields"]
174         )
175 
176     def get_table_query_string(self) -> str:
177         """Returns a string that can directly be used to reference this table in SQL"""
178         if self.table:
179             # Backticks make sure that spark sql knows this a table reference.
180             return f"`{self.table}`"
181         if self.query:
182             return f"({self.query})"
183 
184         # If both the table query string and the actual query are null, we can load from file.
185         spark_session = SparkSession.getActiveSession()
186         if spark_session is None:
187             raise AssertionError("Could not find an active spark session.")
188         try:
189             df = spark_session.read.format(self.file_format).load(self.path)
190         except Exception:
191             logger.exception(
192                 "Spark read of file source failed.\n" + traceback.format_exc()
193             )
194         tmp_table_name = get_temp_entity_table_name()
195         df.createOrReplaceTempView(tmp_table_name)
196 
197         return f"`{tmp_table_name}`"
198 
199 
200 class SparkOptions:
201     allowed_formats = [format.value for format in SparkSourceFormat]
202 
203     def __init__(
204         self,
205         table: Optional[str],
206         query: Optional[str],
207         path: Optional[str],
208         file_format: Optional[str],
209     ):
210         # Check that only one of the ways to load a spark dataframe can be used. We have
211         # to treat empty string and null the same due to proto (de)serialization.
212         if sum([(not (not arg)) for arg in [table, query, path]]) != 1:
213             raise ValueError(
214                 "Exactly one of params(table, query, path) must be specified."
215             )
216         if path:
217             if not file_format:
218                 raise ValueError(
219                     "If 'path' is specified, then 'file_format' is required."
220                 )
221             if file_format not in self.allowed_formats:
222                 raise ValueError(
223                     f"'file_format' should be one of {self.allowed_formats}"
224                 )
225 
226         self._table = table
227         self._query = query
228         self._path = path
229         self._file_format = file_format
230 
231     @property
232     def table(self):
233         return self._table
234 
235     @table.setter
236     def table(self, table):
237         self._table = table
238 
239     @property
240     def query(self):
241         return self._query
242 
243     @query.setter
244     def query(self, query):
245         self._query = query
246 
247     @property
248     def path(self):
249         return self._path
250 
251     @path.setter
252     def path(self, path):
253         self._path = path
254 
255     @property
256     def file_format(self):
257         return self._file_format
258 
259     @file_format.setter
260     def file_format(self, file_format):
261         self._file_format = file_format
262 
263     @classmethod
264     def from_proto(cls, spark_options_proto: DataSourceProto.SparkOptions):
265         """
266         Creates a SparkOptions from a protobuf representation of a spark option
267         args:
268             spark_options_proto: a protobuf representation of a datasource
269         Returns:
270             Returns a SparkOptions object based on the spark_options protobuf
271         """
272         spark_options = cls(
273             table=spark_options_proto.table,
274             query=spark_options_proto.query,
275             path=spark_options_proto.path,
276             file_format=spark_options_proto.file_format,
277         )
278 
279         return spark_options
280 
281     def to_proto(self) -> DataSourceProto.SparkOptions:
282         """
283         Converts an SparkOptionsProto object to its protobuf representation.
284         Returns:
285             SparkOptionsProto protobuf
286         """
287         spark_options_proto = DataSourceProto.SparkOptions(
288             table=self.table,
289             query=self.query,
290             path=self.path,
291             file_format=self.file_format,
292         )
293 
294         return spark_options_proto
295 
296 
297 class SavedDatasetSparkStorage(SavedDatasetStorage):
298     _proto_attr_name = "spark_storage"
299 
300     spark_options: SparkOptions
301 
302     def __init__(
303         self,
304         table: Optional[str] = None,
305         query: Optional[str] = None,
306         path: Optional[str] = None,
307         file_format: Optional[str] = None,
308     ):
309         self.spark_options = SparkOptions(
310             table=table,
311             query=query,
312             path=path,
313             file_format=file_format,
314         )
315 
316     @staticmethod
317     def from_proto(storage_proto: SavedDatasetStorageProto) -> SavedDatasetStorage:
318         spark_options = SparkOptions.from_proto(storage_proto.spark_storage)
319         return SavedDatasetSparkStorage(
320             table=spark_options.table,
321             query=spark_options.query,
322             path=spark_options.path,
323             file_format=spark_options.file_format,
324         )
325 
326     def to_proto(self) -> SavedDatasetStorageProto:
327         return SavedDatasetStorageProto(spark_storage=self.spark_options.to_proto())
328 
329     def to_data_source(self) -> DataSource:
330         return SparkSource(
331             table=self.spark_options.table,
332             query=self.spark_options.query,
333             path=self.spark_options.path,
334             file_format=self.spark_options.file_format,
335         )
336
```
--- 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/contrib/spark_offline_store/spark_source.py b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark_source.py
--- a/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark_source.py
+++ b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark_source.py
@@ -177,7 +177,8 @@
         """Returns a string that can directly be used to reference this table in SQL"""
         if self.table:
             # Backticks make sure that spark sql knows this a table reference.
-            return f"`{self.table}`"
+            table = ".".join([f"`{x}`" for x in self.table.split(".")])
+            return table
         if self.query:
             return f"({self.query})"

{"golden_diff": "diff --git a/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark_source.py b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark_source.py\n--- a/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark_source.py\n+++ b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark_source.py\n@@ -177,7 +177,8 @@\n         \"\"\"Returns a string that can directly be used to reference this table in SQL\"\"\"\n         if self.table:\n             # Backticks make sure that spark sql knows this a table reference.\n-            return f\"`{self.table}`\"\n+            table = \".\".join([f\"`{x}`\" for x in self.table.split(\".\")])\n+            return table\n         if self.query:\n             return f\"({self.query})\"\n", "issue": "spark source complain about \"table or view not found\" error.\n## Expected Behavior \r\nspark offline store\r\n\r\n## Current Behavior\r\nif set table of SparkSource with pattern \"db.table\"\uff0cfeast will complain about \"table or view not found\" error.\r\n\r\n## Steps to reproduce\r\n1. feast init test_repo\r\n2. modify example.py to use SparkSource, and set table of SparkSource with pattern \"db.table\"\r\n3. configure featue_store.yaml\r\n4. feast apply\r\n\r\n### Specifications\r\n\r\n- Version: 0.20.1\r\n- Platform: ubuntu 18/04\r\n- Subsystem:\r\n\r\n## Possible Solution\r\n\n", "before_files": [{"content": "import logging\nimport traceback\nimport warnings\nfrom enum import Enum\nfrom typing import Any, Callable, Dict, Iterable, Optional, Tuple\n\nfrom pyspark.sql import SparkSession\n\nfrom feast.data_source import DataSource\nfrom feast.errors import DataSourceNoNameException\nfrom feast.infra.offline_stores.offline_utils import get_temp_entity_table_name\nfrom feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto\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.type_map import spark_to_feast_value_type\nfrom feast.value_type import ValueType\n\nlogger = logging.getLogger(__name__)\n\n\nclass SparkSourceFormat(Enum):\n    csv = \"csv\"\n    json = \"json\"\n    parquet = \"parquet\"\n    delta = \"delta\"\n    avro = \"avro\"\n\n\nclass SparkSource(DataSource):\n    def __init__(\n        self,\n        *,\n        name: Optional[str] = None,\n        table: Optional[str] = None,\n        query: Optional[str] = None,\n        path: Optional[str] = None,\n        file_format: Optional[str] = None,\n        event_timestamp_column: Optional[str] = None,\n        created_timestamp_column: Optional[str] = None,\n        field_mapping: Optional[Dict[str, str]] = None,\n        date_partition_column: Optional[str] = None,\n        description: Optional[str] = \"\",\n        tags: Optional[Dict[str, str]] = None,\n        owner: Optional[str] = \"\",\n        timestamp_field: Optional[str] = None,\n    ):\n        # If no name, use the table_ref as the default name\n        _name = name\n        if not _name:\n            if table:\n                _name = table\n            else:\n                raise DataSourceNoNameException()\n\n        if date_partition_column:\n            warnings.warn(\n                (\n                    \"The argument 'date_partition_column' is not supported for Spark sources.\"\n                    \"It will be removed in Feast 0.24+\"\n                ),\n                DeprecationWarning,\n            )\n\n        super().__init__(\n            name=_name,\n            event_timestamp_column=event_timestamp_column,\n            created_timestamp_column=created_timestamp_column,\n            field_mapping=field_mapping,\n            description=description,\n            tags=tags,\n            owner=owner,\n            timestamp_field=timestamp_field,\n        )\n        warnings.warn(\n            \"The spark data source API is an experimental feature in alpha development. \"\n            \"This API is unstable and it could and most probably will be changed in the future.\",\n            RuntimeWarning,\n        )\n        self.spark_options = SparkOptions(\n            table=table,\n            query=query,\n            path=path,\n            file_format=file_format,\n        )\n\n    @property\n    def table(self):\n        \"\"\"\n        Returns the table of this feature data source\n        \"\"\"\n        return self.spark_options.table\n\n    @property\n    def query(self):\n        \"\"\"\n        Returns the query of this feature data source\n        \"\"\"\n        return self.spark_options.query\n\n    @property\n    def path(self):\n        \"\"\"\n        Returns the path of the spark data source file.\n        \"\"\"\n        return self.spark_options.path\n\n    @property\n    def file_format(self):\n        \"\"\"\n        Returns the file format of this feature data source.\n        \"\"\"\n        return self.spark_options.file_format\n\n    @staticmethod\n    def from_proto(data_source: DataSourceProto) -> Any:\n        assert data_source.HasField(\"spark_options\")\n        spark_options = SparkOptions.from_proto(data_source.spark_options)\n\n        return SparkSource(\n            name=data_source.name,\n            field_mapping=dict(data_source.field_mapping),\n            table=spark_options.table,\n            query=spark_options.query,\n            path=spark_options.path,\n            file_format=spark_options.file_format,\n            timestamp_field=data_source.timestamp_field,\n            created_timestamp_column=data_source.created_timestamp_column,\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_SPARK,\n            data_source_class_type=\"feast.infra.offline_stores.contrib.spark_offline_store.spark_source.SparkSource\",\n            field_mapping=self.field_mapping,\n            spark_options=self.spark_options.to_proto(),\n            description=self.description,\n            tags=self.tags,\n            owner=self.owner,\n        )\n\n        data_source_proto.timestamp_field = self.timestamp_field\n        data_source_proto.created_timestamp_column = self.created_timestamp_column\n\n        return data_source_proto\n\n    def validate(self, config: RepoConfig):\n        self.get_table_column_names_and_types(config)\n\n    @staticmethod\n    def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:\n        return spark_to_feast_value_type\n\n    def get_table_column_names_and_types(\n        self, config: RepoConfig\n    ) -> Iterable[Tuple[str, str]]:\n        from feast.infra.offline_stores.contrib.spark_offline_store.spark import (\n            get_spark_session_or_start_new_with_repoconfig,\n        )\n\n        spark_session = get_spark_session_or_start_new_with_repoconfig(\n            store_config=config.offline_store\n        )\n        df = spark_session.sql(f\"SELECT * FROM {self.get_table_query_string()}\")\n        return (\n            (fields[\"name\"], fields[\"type\"])\n            for fields in df.schema.jsonValue()[\"fields\"]\n        )\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            # Backticks make sure that spark sql knows this a table reference.\n            return f\"`{self.table}`\"\n        if self.query:\n            return f\"({self.query})\"\n\n        # If both the table query string and the actual query are null, we can load from file.\n        spark_session = SparkSession.getActiveSession()\n        if spark_session is None:\n            raise AssertionError(\"Could not find an active spark session.\")\n        try:\n            df = spark_session.read.format(self.file_format).load(self.path)\n        except Exception:\n            logger.exception(\n                \"Spark read of file source failed.\\n\" + traceback.format_exc()\n            )\n        tmp_table_name = get_temp_entity_table_name()\n        df.createOrReplaceTempView(tmp_table_name)\n\n        return f\"`{tmp_table_name}`\"\n\n\nclass SparkOptions:\n    allowed_formats = [format.value for format in SparkSourceFormat]\n\n    def __init__(\n        self,\n        table: Optional[str],\n        query: Optional[str],\n        path: Optional[str],\n        file_format: Optional[str],\n    ):\n        # Check that only one of the ways to load a spark dataframe can be used. We have\n        # to treat empty string and null the same due to proto (de)serialization.\n        if sum([(not (not arg)) for arg in [table, query, path]]) != 1:\n            raise ValueError(\n                \"Exactly one of params(table, query, path) must be specified.\"\n            )\n        if path:\n            if not file_format:\n                raise ValueError(\n                    \"If 'path' is specified, then 'file_format' is required.\"\n                )\n            if file_format not in self.allowed_formats:\n                raise ValueError(\n                    f\"'file_format' should be one of {self.allowed_formats}\"\n                )\n\n        self._table = table\n        self._query = query\n        self._path = path\n        self._file_format = file_format\n\n    @property\n    def table(self):\n        return self._table\n\n    @table.setter\n    def table(self, table):\n        self._table = table\n\n    @property\n    def query(self):\n        return self._query\n\n    @query.setter\n    def query(self, query):\n        self._query = query\n\n    @property\n    def path(self):\n        return self._path\n\n    @path.setter\n    def path(self, path):\n        self._path = path\n\n    @property\n    def file_format(self):\n        return self._file_format\n\n    @file_format.setter\n    def file_format(self, file_format):\n        self._file_format = file_format\n\n    @classmethod\n    def from_proto(cls, spark_options_proto: DataSourceProto.SparkOptions):\n        \"\"\"\n        Creates a SparkOptions from a protobuf representation of a spark option\n        args:\n            spark_options_proto: a protobuf representation of a datasource\n        Returns:\n            Returns a SparkOptions object based on the spark_options protobuf\n        \"\"\"\n        spark_options = cls(\n            table=spark_options_proto.table,\n            query=spark_options_proto.query,\n            path=spark_options_proto.path,\n            file_format=spark_options_proto.file_format,\n        )\n\n        return spark_options\n\n    def to_proto(self) -> DataSourceProto.SparkOptions:\n        \"\"\"\n        Converts an SparkOptionsProto object to its protobuf representation.\n        Returns:\n            SparkOptionsProto protobuf\n        \"\"\"\n        spark_options_proto = DataSourceProto.SparkOptions(\n            table=self.table,\n            query=self.query,\n            path=self.path,\n            file_format=self.file_format,\n        )\n\n        return spark_options_proto\n\n\nclass SavedDatasetSparkStorage(SavedDatasetStorage):\n    _proto_attr_name = \"spark_storage\"\n\n    spark_options: SparkOptions\n\n    def __init__(\n        self,\n        table: Optional[str] = None,\n        query: Optional[str] = None,\n        path: Optional[str] = None,\n        file_format: Optional[str] = None,\n    ):\n        self.spark_options = SparkOptions(\n            table=table,\n            query=query,\n            path=path,\n            file_format=file_format,\n        )\n\n    @staticmethod\n    def from_proto(storage_proto: SavedDatasetStorageProto) -> SavedDatasetStorage:\n        spark_options = SparkOptions.from_proto(storage_proto.spark_storage)\n        return SavedDatasetSparkStorage(\n            table=spark_options.table,\n            query=spark_options.query,\n            path=spark_options.path,\n            file_format=spark_options.file_format,\n        )\n\n    def to_proto(self) -> SavedDatasetStorageProto:\n        return SavedDatasetStorageProto(spark_storage=self.spark_options.to_proto())\n\n    def to_data_source(self) -> DataSource:\n        return SparkSource(\n            table=self.spark_options.table,\n            query=self.spark_options.query,\n            path=self.spark_options.path,\n            file_format=self.spark_options.file_format,\n        )\n", "path": "sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark_source.py"}], "after_files": [{"content": "import logging\nimport traceback\nimport warnings\nfrom enum import Enum\nfrom typing import Any, Callable, Dict, Iterable, Optional, Tuple\n\nfrom pyspark.sql import SparkSession\n\nfrom feast.data_source import DataSource\nfrom feast.errors import DataSourceNoNameException\nfrom feast.infra.offline_stores.offline_utils import get_temp_entity_table_name\nfrom feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto\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.type_map import spark_to_feast_value_type\nfrom feast.value_type import ValueType\n\nlogger = logging.getLogger(__name__)\n\n\nclass SparkSourceFormat(Enum):\n    csv = \"csv\"\n    json = \"json\"\n    parquet = \"parquet\"\n    delta = \"delta\"\n    avro = \"avro\"\n\n\nclass SparkSource(DataSource):\n    def __init__(\n        self,\n        *,\n        name: Optional[str] = None,\n        table: Optional[str] = None,\n        query: Optional[str] = None,\n        path: Optional[str] = None,\n        file_format: Optional[str] = None,\n        event_timestamp_column: Optional[str] = None,\n        created_timestamp_column: Optional[str] = None,\n        field_mapping: Optional[Dict[str, str]] = None,\n        date_partition_column: Optional[str] = None,\n        description: Optional[str] = \"\",\n        tags: Optional[Dict[str, str]] = None,\n        owner: Optional[str] = \"\",\n        timestamp_field: Optional[str] = None,\n    ):\n        # If no name, use the table_ref as the default name\n        _name = name\n        if not _name:\n            if table:\n                _name = table\n            else:\n                raise DataSourceNoNameException()\n\n        if date_partition_column:\n            warnings.warn(\n                (\n                    \"The argument 'date_partition_column' is not supported for Spark sources.\"\n                    \"It will be removed in Feast 0.24+\"\n                ),\n                DeprecationWarning,\n            )\n\n        super().__init__(\n            name=_name,\n            event_timestamp_column=event_timestamp_column,\n            created_timestamp_column=created_timestamp_column,\n            field_mapping=field_mapping,\n            description=description,\n            tags=tags,\n            owner=owner,\n            timestamp_field=timestamp_field,\n        )\n        warnings.warn(\n            \"The spark data source API is an experimental feature in alpha development. \"\n            \"This API is unstable and it could and most probably will be changed in the future.\",\n            RuntimeWarning,\n        )\n        self.spark_options = SparkOptions(\n            table=table,\n            query=query,\n            path=path,\n            file_format=file_format,\n        )\n\n    @property\n    def table(self):\n        \"\"\"\n        Returns the table of this feature data source\n        \"\"\"\n        return self.spark_options.table\n\n    @property\n    def query(self):\n        \"\"\"\n        Returns the query of this feature data source\n        \"\"\"\n        return self.spark_options.query\n\n    @property\n    def path(self):\n        \"\"\"\n        Returns the path of the spark data source file.\n        \"\"\"\n        return self.spark_options.path\n\n    @property\n    def file_format(self):\n        \"\"\"\n        Returns the file format of this feature data source.\n        \"\"\"\n        return self.spark_options.file_format\n\n    @staticmethod\n    def from_proto(data_source: DataSourceProto) -> Any:\n        assert data_source.HasField(\"spark_options\")\n        spark_options = SparkOptions.from_proto(data_source.spark_options)\n\n        return SparkSource(\n            name=data_source.name,\n            field_mapping=dict(data_source.field_mapping),\n            table=spark_options.table,\n            query=spark_options.query,\n            path=spark_options.path,\n            file_format=spark_options.file_format,\n            timestamp_field=data_source.timestamp_field,\n            created_timestamp_column=data_source.created_timestamp_column,\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_SPARK,\n            data_source_class_type=\"feast.infra.offline_stores.contrib.spark_offline_store.spark_source.SparkSource\",\n            field_mapping=self.field_mapping,\n            spark_options=self.spark_options.to_proto(),\n            description=self.description,\n            tags=self.tags,\n            owner=self.owner,\n        )\n\n        data_source_proto.timestamp_field = self.timestamp_field\n        data_source_proto.created_timestamp_column = self.created_timestamp_column\n\n        return data_source_proto\n\n    def validate(self, config: RepoConfig):\n        self.get_table_column_names_and_types(config)\n\n    @staticmethod\n    def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:\n        return spark_to_feast_value_type\n\n    def get_table_column_names_and_types(\n        self, config: RepoConfig\n    ) -> Iterable[Tuple[str, str]]:\n        from feast.infra.offline_stores.contrib.spark_offline_store.spark import (\n            get_spark_session_or_start_new_with_repoconfig,\n        )\n\n        spark_session = get_spark_session_or_start_new_with_repoconfig(\n            store_config=config.offline_store\n        )\n        df = spark_session.sql(f\"SELECT * FROM {self.get_table_query_string()}\")\n        return (\n            (fields[\"name\"], fields[\"type\"])\n            for fields in df.schema.jsonValue()[\"fields\"]\n        )\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            # Backticks make sure that spark sql knows this a table reference.\n            table = \".\".join([f\"`{x}`\" for x in self.table.split(\".\")])\n            return table\n        if self.query:\n            return f\"({self.query})\"\n\n        # If both the table query string and the actual query are null, we can load from file.\n        spark_session = SparkSession.getActiveSession()\n        if spark_session is None:\n            raise AssertionError(\"Could not find an active spark session.\")\n        try:\n            df = spark_session.read.format(self.file_format).load(self.path)\n        except Exception:\n            logger.exception(\n                \"Spark read of file source failed.\\n\" + traceback.format_exc()\n            )\n        tmp_table_name = get_temp_entity_table_name()\n        df.createOrReplaceTempView(tmp_table_name)\n\n        return f\"`{tmp_table_name}`\"\n\n\nclass SparkOptions:\n    allowed_formats = [format.value for format in SparkSourceFormat]\n\n    def __init__(\n        self,\n        table: Optional[str],\n        query: Optional[str],\n        path: Optional[str],\n        file_format: Optional[str],\n    ):\n        # Check that only one of the ways to load a spark dataframe can be used. We have\n        # to treat empty string and null the same due to proto (de)serialization.\n        if sum([(not (not arg)) for arg in [table, query, path]]) != 1:\n            raise ValueError(\n                \"Exactly one of params(table, query, path) must be specified.\"\n            )\n        if path:\n            if not file_format:\n                raise ValueError(\n                    \"If 'path' is specified, then 'file_format' is required.\"\n                )\n            if file_format not in self.allowed_formats:\n                raise ValueError(\n                    f\"'file_format' should be one of {self.allowed_formats}\"\n                )\n\n        self._table = table\n        self._query = query\n        self._path = path\n        self._file_format = file_format\n\n    @property\n    def table(self):\n        return self._table\n\n    @table.setter\n    def table(self, table):\n        self._table = table\n\n    @property\n    def query(self):\n        return self._query\n\n    @query.setter\n    def query(self, query):\n        self._query = query\n\n    @property\n    def path(self):\n        return self._path\n\n    @path.setter\n    def path(self, path):\n        self._path = path\n\n    @property\n    def file_format(self):\n        return self._file_format\n\n    @file_format.setter\n    def file_format(self, file_format):\n        self._file_format = file_format\n\n    @classmethod\n    def from_proto(cls, spark_options_proto: DataSourceProto.SparkOptions):\n        \"\"\"\n        Creates a SparkOptions from a protobuf representation of a spark option\n        args:\n            spark_options_proto: a protobuf representation of a datasource\n        Returns:\n            Returns a SparkOptions object based on the spark_options protobuf\n        \"\"\"\n        spark_options = cls(\n            table=spark_options_proto.table,\n            query=spark_options_proto.query,\n            path=spark_options_proto.path,\n            file_format=spark_options_proto.file_format,\n        )\n\n        return spark_options\n\n    def to_proto(self) -> DataSourceProto.SparkOptions:\n        \"\"\"\n        Converts an SparkOptionsProto object to its protobuf representation.\n        Returns:\n            SparkOptionsProto protobuf\n        \"\"\"\n        spark_options_proto = DataSourceProto.SparkOptions(\n            table=self.table,\n            query=self.query,\n            path=self.path,\n            file_format=self.file_format,\n        )\n\n        return spark_options_proto\n\n\nclass SavedDatasetSparkStorage(SavedDatasetStorage):\n    _proto_attr_name = \"spark_storage\"\n\n    spark_options: SparkOptions\n\n    def __init__(\n        self,\n        table: Optional[str] = None,\n        query: Optional[str] = None,\n        path: Optional[str] = None,\n        file_format: Optional[str] = None,\n    ):\n        self.spark_options = SparkOptions(\n            table=table,\n            query=query,\n            path=path,\n            file_format=file_format,\n        )\n\n    @staticmethod\n    def from_proto(storage_proto: SavedDatasetStorageProto) -> SavedDatasetStorage:\n        spark_options = SparkOptions.from_proto(storage_proto.spark_storage)\n        return SavedDatasetSparkStorage(\n            table=spark_options.table,\n            query=spark_options.query,\n            path=spark_options.path,\n            file_format=spark_options.file_format,\n        )\n\n    def to_proto(self) -> SavedDatasetStorageProto:\n        return SavedDatasetStorageProto(spark_storage=self.spark_options.to_proto())\n\n    def to_data_source(self) -> DataSource:\n        return SparkSource(\n            table=self.spark_options.table,\n            query=self.spark_options.query,\n            path=self.spark_options.path,\n            file_format=self.spark_options.file_format,\n        )\n", "path": "sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark_source.py"}]}