[py] support for UUID type

abhizer · ryzhyk · commit f4f33fff7849 · 2025-01-16T19:07:46.000Z
Signed-off-by: Abhinav Gyawali &lt;22275402+abhizer@users.noreply.github.com&gt;
diff --git a/demo/project_demo08-DebeziumJDBC/run.py b/demo/project_demo08-DebeziumJDBC/run.py
@@ -334,7 +334,7 @@ def generate_inputs(pipeline: Pipeline):
                     "f8": date_time.strftime("%F"),
                     # "f9": list("bar".encode('utf-8')),
                     "f10": {"id": i + batch, "f1": True, "f2": "foo", "f4": 10.5},
-                    "f11": str(uuid.uuid4())
+                    "f11": str(uuid.uuid4()),
                 }
             )
         inserts = [{"insert": element} for element in data]
diff --git a/python/feldera/_helpers.py b/python/feldera/_helpers.py
@@ -1,3 +1,5 @@
+import uuid
+
 import pandas as pd
 from decimal import Decimal
 
@@ -8,6 +10,8 @@ def sql_type_to_pandas_type(sql_type: str):
     """
 
     match sql_type.upper():
+        case "UUID":
+            return None
         case "BOOLEAN":
             return "boolean"
         case "TINYINT":
@@ -64,6 +68,7 @@ def dataframe_from_response(buffer: list[list[dict]], schema: dict):
     pd_schema = {}
 
     decimal_col = []
+    uuid_col = []
 
     for column in schema["fields"]:
         column_name = column["name"]
@@ -72,6 +77,8 @@ def dataframe_from_response(buffer: list[list[dict]], schema: dict):
         column_type = column["columntype"]["type"]
         if column_type == "DECIMAL":
             decimal_col.append(column_name)
+        elif column_type == "UUID":
+            uuid_col.append(column_name)
 
         pd_schema[column_name] = sql_type_to_pandas_type(column_type)
 
@@ -89,6 +96,12 @@ def dataframe_from_response(buffer: list[list[dict]], schema: dict):
                 if datum[col] is not None:
                     datum[col] = Decimal(datum[col])
 
+    if len(uuid_col) != 0:
+        for datum in data:
+            for col in uuid_col:
+                if datum[col] is not None:
+                    datum[col] = uuid.UUID(datum[col])
+
     df = pd.DataFrame(data)
     df = df.astype(pd_schema)
 
diff --git a/python/feldera/rest/_httprequests.py b/python/feldera/rest/_httprequests.py
@@ -14,7 +14,8 @@
 
 
 def json_serialize(body: Any) -> str:
-    return json.dumps(body) if body else "" if body == "" else "null"
+    # serialize as string if this object cannot be serialized (e.g. UUID)
+    return json.dumps(body, default=str) if body else "" if body == "" else "null"
 
 
 class HttpRequests:
diff --git a/python/tests/test_pipeline.py b/python/tests/test_pipeline.py
@@ -17,12 +17,15 @@ def test_delete_all_pipelines(self):
         for pipeline in pipelines:
             TEST_CLIENT.delete_pipeline(pipeline.name)
 
-    def test_create_pipeline(self, name: str = "blah", delete=False):
+    def test_create_pipeline(
+        self, name: str = "blah", delete=False, runtime_config: dict = {}
+    ):
+        self.test_delete_all_pipelines()
         sql = """
         CREATE TABLE tbl(id INT) WITH ('append_only' = 'true');
         CREATE VIEW V AS SELECT * FROM tbl;
         """
-        pipeline = Pipeline(name, sql, "", "", {}, {})
+        pipeline = Pipeline(name, sql, "", "", {}, runtime_config)
         pipeline = TEST_CLIENT.create_pipeline(pipeline)
 
         if delete:
@@ -84,12 +87,12 @@ def test_get_pipeline(self):
 
     def test_get_pipeline_config(self):
         name = str(uuid.uuid4())
-        self.test_create_pipeline(name, False)
+        self.test_create_pipeline(name, False, {"workers": 2, "storage": False})
         config = TEST_CLIENT.get_runtime_config(name)
 
         assert config is not None
-        assert config.get("workers") is not None
-        assert config.get("storage") is not None
+        assert config.get("workers") == 2
+        assert not config.get("storage")
 
         TEST_CLIENT.delete_pipeline(name)
 
diff --git a/python/tests/test_pipeline_builder.py b/python/tests/test_pipeline_builder.py
@@ -1,6 +1,8 @@
 import os
 import time
 import unittest
+import uuid
+
 import pandas as pd
 from kafka import KafkaProducer, KafkaConsumer
 from kafka.admin import KafkaAdminClient, NewTopic
@@ -1057,6 +1059,30 @@ def test_connector_orchestration(self):
         pipeline.shutdown()
         pipeline.delete()
 
+    def test_uuid(self):
+        name = "test_uuid"
+        sql = """
+        CREATE TABLE t0(c0 UUID);
+        CREATE MATERIALIZED VIEW v0 AS SELECT c0 FROM t0;
+        """
+
+        pipeline = PipelineBuilder(TEST_CLIENT, name, sql=sql).create_or_replace()
+        out = pipeline.listen("v0")
+        pipeline.start()
+
+        data = [{"c0": uuid.uuid4()}]
+        pipeline.input_json("t0", data)
+
+        for datum in data:
+            datum.update({"insert_delete": 1})
+
+        pipeline.wait_for_completion(True)
+
+        got = out.to_dict()
+        pipeline.shutdown()
+
+        assert got == data
+
 
 if __name__ == "__main__":
     unittest.main()
diff --git a/python/uv.lock b/python/uv.lock

Original file line number	Diff line number	Diff line change
`@@ -334,7 +334,7 @@ def generate_inputs(pipeline: Pipeline):`
`334`	`334`	`"f8": date_time.strftime("%F"),`
`335`	`335`	`# "f9": list("bar".encode('utf-8')),`
`336`	`336`	`"f10": {"id": i + batch, "f1": True, "f2": "foo", "f4": 10.5},`
`337`		`- "f11": str(uuid.uuid4())`
	`337`	`+ "f11": str(uuid.uuid4()),`
`338`	`338`	`}`
`339`	`339`	`)`
`340`	`340`	`inserts = [{"insert": element} for element in data]`