feast-dev
diff --git a/‎sdk/python/feast/data_source.py‎
Lines changed: 3 additions & 1 deletion b/‎sdk/python/feast/data_source.py‎
Lines changed: 3 additions & 1 deletion
diff --git a/‎sdk/python/feast/feature_store.py‎
Lines changed: 0 additions & 4 deletions b/‎sdk/python/feast/feature_store.py‎
Lines changed: 0 additions & 4 deletions
diff --git a/‎sdk/python/feast/offline_store.py‎
Lines changed: 5 additions & 6 deletions b/‎sdk/python/feast/offline_store.py‎
Lines changed: 5 additions & 6 deletions
@@ -172,6 +172,7 @@ def to_proto(self) -> DataSourceProto.BigQueryOptions:
 
         bigquery_options_proto = DataSourceProto.BigQueryOptions(
             table_ref=self.table_ref,
+            query=self.query,
         )
 
         return bigquery_options_proto
@@ -461,13 +462,14 @@ def from_proto(data_source):
                 created_timestamp_column=data_source.created_timestamp_column,
                 date_partition_column=data_source.date_partition_column,
             )
-        elif data_source.bigquery_options.table_ref:
+        elif (data_source.bigquery_options.table_ref or data_source.bigquery_options.query):
             data_source_obj = BigQuerySource(
                 field_mapping=data_source.field_mapping,
                 table_ref=data_source.bigquery_options.table_ref,
                 event_timestamp_column=data_source.event_timestamp_column,
                 created_timestamp_column=data_source.created_timestamp_column,
                 date_partition_column=data_source.date_partition_column,
+                query=data_source.bigquery_options.query,
             )
         elif (
             data_source.kafka_options.bootstrap_servers
 
@@ -217,10 +217,6 @@ def materialize(
                 raise NotImplementedError(
                     "This function is not yet implemented for File data sources"
                 )
-            if not feature_view.input.table_ref:
-                raise NotImplementedError(
-                    f"This function is only implemented for FeatureViews with a table_ref; {feature_view.name} does not have one."
-                )
             (
                 entity_names,
                 feature_names,
 
@@ -125,11 +125,10 @@ def pull_latest_from_table(
         end_date: datetime,
     ) -> pyarrow.Table:
         assert isinstance(data_source, BigQuerySource)
-        table_ref = data_source.table_ref
-        if table_ref is None:
-            raise ValueError(
-                "This function can only be called on a FeatureView with a table_ref"
-            )
+        if data_source.table_ref:
+            from_table = f"`{data_source.table_ref}`"
+        else:
+            from_table = f"({data_source.query})"
 
         partition_by_entity_string = ", ".join(entity_names)
         if partition_by_entity_string != "":
@@ -145,7 +144,7 @@ def pull_latest_from_table(
         FROM (
             SELECT {field_string},
             ROW_NUMBER() OVER({partition_by_entity_string} ORDER BY {timestamp_desc_string}) AS _feast_row
-            FROM `{table_ref}`
+            FROM {from_table}
             WHERE {event_timestamp_column} BETWEEN TIMESTAMP('{start_date}') AND TIMESTAMP('{end_date}')
         )
         WHERE _feast_row = 1
Original file line number	Diff line number	Diff line change
`@@ -217,10 +217,6 @@ def materialize(`
`217`	`217`	`raise NotImplementedError(`
`218`	`218`	`"This function is not yet implemented for File data sources"`
`219`	`219`	`)`
`220`		`- if not feature_view.input.table_ref:`
`221`		`- raise NotImplementedError(`
`222`		`- f"This function is only implemented for FeatureViews with a table_ref; {feature_view.name} does not have one."`
`223`		`- )`
`224`	`220`	`(`
`225`	`221`	`entity_names,`
`226`	`222`	`feature_names,`