py: support listening to case sensitive views

abhizer · abhizer · commit 3357be103606 · 2025-09-02T10:00:50.000Z
Signed-off-by: Abhinav Gyawali &lt;22275402+abhizer@users.noreply.github.com&gt;
diff --git a/python/feldera/_callback_runner.py b/python/feldera/_callback_runner.py
@@ -39,14 +39,12 @@ def run(self):
         """
 
         pipeline = self.client.get_pipeline(self.pipeline_name)
-        schema = pipeline.program_info["schema"]
 
-        if schema:
-            schemas = [relation for relation in schema["inputs"] + schema["outputs"]]
-            for schema in schemas:
-                if schema["name"] == self.view_name:
-                    self.schema = schema
-                    break
+        schemas = pipeline.tables + pipeline.views
+        for schema in schemas:
+            if schema.name == self.view_name:
+                self.schema = schema
+                break
 
         if self.schema is None:
             raise ValueError(
@@ -66,7 +64,10 @@ def run(self):
             case _CallbackRunnerInstruction.PipelineStarted:
                 # listen to the pipeline
                 gen_obj = self.client.listen_to_pipeline(
-                    self.pipeline_name, self.view_name, format="json"
+                    self.pipeline_name,
+                    self.view_name,
+                    format="json",
+                    case_sensitive=self.schema.case_sensitive,
                 )
 
                 # if there is a queue set up, inform the main thread that the listener has been started, and it can
@@ -83,7 +84,7 @@ def run(self):
                     seq_no: Optional[int] = chunk.get("sequence_number")
                     if data is not None and seq_no is not None:
                         self.callback(
-                            dataframe_from_response([data], self.schema), seq_no
+                            dataframe_from_response([data], self.schema.fields), seq_no
                         )
 
                     if self.queue:
diff --git a/python/feldera/_helpers.py b/python/feldera/_helpers.py
@@ -60,7 +60,7 @@ def ensure_dataframe_has_columns(df: pd.DataFrame):
         )
 
 
-def dataframe_from_response(buffer: list[list[dict]], schema: dict):
+def dataframe_from_response(buffer: list[list[dict]], fields: list[dict]):
     """
     Converts the response from Feldera to a pandas DataFrame.
     """
@@ -70,7 +70,7 @@ def dataframe_from_response(buffer: list[list[dict]], schema: dict):
     decimal_col = []
     uuid_col = []
 
-    for column in schema["fields"]:
+    for column in fields:
         column_name = column["name"]
         if not column["case_sensitive"]:
             column_name = column_name.lower()
diff --git a/python/feldera/rest/feldera_client.py b/python/feldera/rest/feldera_client.py
@@ -830,6 +830,7 @@ def listen_to_pipeline(
         backpressure: bool = True,
         array: bool = False,
         timeout: Optional[float] = None,
+        case_sensitive: bool = False,
     ):
         """
         Listen for updates to views for pipeline, yields the chunks of data
@@ -845,6 +846,7 @@ def listen_to_pipeline(
             "json" format, the default value is False
 
         :param timeout: The amount of time in seconds to listen to the stream for
+        :param case_sensitive: True if the table name is case sensitive, False by default
         """
 
         params = {
@@ -855,6 +857,8 @@ def listen_to_pipeline(
         if format == "json":
             params["array"] = _prepare_boolean_input(array)
 
+        table_name = f'"{table_name}"' if case_sensitive else table_name
+
         resp = self.http.post(
             path=f"/pipelines/{pipeline_name}/egress/{table_name}",
             params=params,
diff --git a/python/tests/platform/test_shared_pipeline.py b/python/tests/platform/test_shared_pipeline.py
@@ -6,6 +6,7 @@
 import io
 import tempfile
 import zipfile
+import sys
 
 from tests.shared_test_pipeline import SharedTestPipeline
 from tests import TEST_CLIENT, enterprise_only
@@ -19,6 +20,7 @@ def test_create_pipeline(self):
         """
         CREATE TABLE tbl(id INT) WITH ('materialized' = 'true');
         CREATE MATERIALIZED VIEW v0 AS SELECT * FROM tbl;
+        CREATE MATERIALIZED VIEW "V0" AS SELECT * FROM tbl WHERE id % 2 <> 0;
         """
         pass
 
@@ -65,6 +67,26 @@ def test_get_pipeline_stats(self):
         assert stats.get("inputs") is not None
         assert stats.get("outputs") is not None
 
+    def test_case_sensitive_views_listen(self):
+        all_stream = self.pipeline.listen("v0")
+        odd_stream = self.pipeline.listen("V0")
+
+        self.pipeline.start()
+        self.pipeline.input_json("tbl", [{"id": i} for i in range(10)])
+        self.pipeline.wait_for_completion()
+
+        all = all_stream.to_dict()
+        odd = odd_stream.to_dict()
+
+        expected_all = list(self.pipeline.query("select * from v0"))
+        expected_odd = list(self.pipeline.query('select * from "V0"'))
+
+        def extract_ids(x):
+            return sorted(i["id"] for i in x)
+
+        assert extract_ids(all) == extract_ids(expected_all)
+        assert extract_ids(odd) == extract_ids(expected_odd)
+
     def test_adhoc_query_text(self):
         data = "1\n2\n"
         self.pipeline.start()
