Add WHERE clause to Iceberg materialized view

Author: kesmit13

singlestoredb/fusion/handlers/iceberg_materialized_view.py

@@ -21,6 +21,7 @@ class CreateIcebergMaterializedView(SQLHandler):
         ON iceberg_table
         [ catalog ]
         [ storage ]
+        [ where ]
     ;
 
     # If not exists
@@ -42,6 +43,9 @@ class CreateIcebergMaterializedView(SQLHandler):
     _link_config = S3 CONFIG '<link-config>'
     _link_creds = CREDENTIALS '<link-creds>'
 
+    # Where clause
+    where = WHERE <expr>
+
     Description
     -----------
     Create an Iceberg materialized view that syncs data from an Iceberg table
@@ -107,6 +111,10 @@ def run(self, params: Dict[str, Any]) -> Optional[result.FusionSQLResult]:
                 '[catalog.]database.table',
             )
 
+        where = ''
+        if params.get('where'):
+            where = f' WHERE {params["where"]}'
+
         # Iceberg expects lowercase
         if iceberg_database:
             iceberg_database = iceberg_database.lower()
@@ -180,21 +188,26 @@ def run(self, params: Dict[str, Any]) -> Optional[result.FusionSQLResult]:
         with connect() as conn:
             with conn.cursor() as cur:
                 # Infer and create the pipeline.
-                # It also creates a table (and optionally a view in case of merge pipeline) with the same name
+                # It also creates a table (and optionally a view in case of
+                # merge pipeline) with the same name
                 cur.execute(rf'''
                     CREATE INFERRED PIPELINE `{pipeline_name}` AS
                         LOAD DATA S3 '{table_id}'
                         CONFIG '{config_json}'
                         CREDENTIALS '{creds_json}'
                         FORMAT ICEBERG
                         OPTIONS = 'merge'
+                        {where}
                 ''')
 
                 # Start the pipeline
                 cur.execute(rf'START PIPELINE `{pipeline_name}`')
 
                 # Create view with user-provided name
-                cur.execute(rf'CREATE VIEW `{view_database}`.`{view_table}` AS SELECT * FROM `{pipeline_name}`')
+                cur.execute(rf'''
+                    CREATE VIEW `{view_database}`.`{view_table}`
+                        AS SELECT * FROM `{pipeline_name}`
+                ''')
 
         # Return result
         res = result.FusionSQLResult()