From 6990d1779d64225bdd1547786c4016efffa8b630 Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Sun, 24 Aug 2025 02:57:43 +0300
Subject: [PATCH 01/19] implement python gRPC server code in iceberg sink

---
 cmd/pyiceberg_receiver/__main__.py         | 17 +++++++++++++++++
 cmd/pyiceberg_receiver/iceberg_receiver.py |  6 ++++++
 2 files changed, 23 insertions(+)
 create mode 100644 cmd/pyiceberg_receiver/__main__.py
 create mode 100644 cmd/pyiceberg_receiver/iceberg_receiver.py

diff --git a/cmd/pyiceberg_receiver/__main__.py b/cmd/pyiceberg_receiver/__main__.py
new file mode 100644
index 0000000..3473653
--- /dev/null
+++ b/cmd/pyiceberg_receiver/__main__.py
@@ -0,0 +1,17 @@
+import grpc
+from concurrent import futures
+from iceberg_receiver import Receiver
+from pgwatch_pb2_grpc import add_ReceiverServicer_to_server
+
+def serve(port: int):
+    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
+    add_ReceiverServicer_to_server(
+        Receiver(),
+        server,
+    )
+    server.add_insecure_port(f"0.0.0.0:{port}")
+    server.start()
+    server.wait_for_termination()
+
+if __name__ == "__main__":
+    serve(1234)
\ No newline at end of file
diff --git a/cmd/pyiceberg_receiver/iceberg_receiver.py b/cmd/pyiceberg_receiver/iceberg_receiver.py
new file mode 100644
index 0000000..80f168d
--- /dev/null
+++ b/cmd/pyiceberg_receiver/iceberg_receiver.py
@@ -0,0 +1,6 @@
+from pgwatch_pb2_grpc import ReceiverServicer
+from pgwatch_pb2 import Reply
+
+class Receiver(ReceiverServicer):
+    def UpdateMeasurements(self, request, context):
+        return super().UpdateMeasurements(request, context)
\ No newline at end of file

From 7305331b926db5bdb38171fa266a4462685d09cc Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Sun, 24 Aug 2025 02:58:42 +0300
Subject: [PATCH 02/19] update .gitignore

---
 .gitignore | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/.gitignore b/.gitignore
index 2d90b05..404625b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,3 +1,5 @@
 ./*/*/*.txt
 .vscode/
-*pb.go
\ No newline at end of file
+*pb.go
+*pb2*.py
+__pycache__/
\ No newline at end of file

From c698c751a2d6f9f1965d08225e844570c3f1fa89 Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Sun, 24 Aug 2025 02:59:56 +0300
Subject: [PATCH 03/19] implement iceberg table initalization logic

---
 cmd/pyiceberg_receiver/__main__.py | 61 ++++++++++++++++++++++++++++--
 1 file changed, 58 insertions(+), 3 deletions(-)

diff --git a/cmd/pyiceberg_receiver/__main__.py b/cmd/pyiceberg_receiver/__main__.py
index 3473653..0a6ea0a 100644
--- a/cmd/pyiceberg_receiver/__main__.py
+++ b/cmd/pyiceberg_receiver/__main__.py
@@ -1,17 +1,72 @@
 import grpc
+import argparse
 from concurrent import futures
 from iceberg_receiver import Receiver
 from pgwatch_pb2_grpc import add_ReceiverServicer_to_server
+from pyiceberg.catalog import load_catalog
+from pyiceberg.schema import Schema
+from pyiceberg.partitioning import PartitionSpec, PartitionField
+from pyiceberg.transforms import IdentityTransform
+from pyiceberg.types import (
+    StringType,
+    NestedField, 
+)
+
+parser = argparse.ArgumentParser()
+parser.add_argument(
+    "-p", "--port", 
+    type=int,
+    dest="port", 
+    required=True, 
+    action="store", 
+    help="The port number to use for the gRPC server."
+)
+
+parser.add_argument(
+    "-d", "--iceberg-data-dir",
+    type=str, 
+    dest="icebergDataDir",
+    metavar="DIR",
+    required=True,
+    action="store",
+    help="Directory to store iceberg tables in."
+)
+args = parser.parse_args()
+
+catalog = load_catalog("pgcatalog")
+catalog.create_namespace_if_not_exists("pgwatch")
+
+schema = Schema(
+    NestedField(field_id=1, name="DBName", field_type=StringType(), required=True),
+    NestedField(field_id=2, name="MetricName", field_type=StringType(), required=True),
+    NestedField(field_id=4, name="Data", field_type=StringType(), required=True),
+)
+
+partition_spec = PartitionSpec(
+    PartitionField(
+        source_id=2, field_id=1000, transform=IdentityTransform(), name="MetricName"
+    ),
+    PartitionField(
+        source_id=1, field_id=1001, transform=IdentityTransform(), name="DBName"
+    ),
+)
+
+tbl = catalog.create_table_if_not_exists(
+    identifier="pgwatch.metrics",
+    schema=schema,
+    location=args.icebergDataDir,
+    partition_spec=partition_spec
+)
 
 def serve(port: int):
     server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
     add_ReceiverServicer_to_server(
-        Receiver(),
+        Receiver(tbl),
         server,
     )
     server.add_insecure_port(f"0.0.0.0:{port}")
     server.start()
+    print(f"gRPC server started, listening on port {port}")
     server.wait_for_termination()
 
-if __name__ == "__main__":
-    serve(1234)
\ No newline at end of file
+serve(args.port)
\ No newline at end of file

From 12f9ad8215dc63f1656cf391a24ff64857faead9 Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Sun, 24 Aug 2025 03:01:49 +0300
Subject: [PATCH 04/19] add `.pyiceberg.yaml` to hold catalog configs

---
 cmd/pyiceberg_receiver/.pyiceberg.yaml | 7 +++++++
 1 file changed, 7 insertions(+)
 create mode 100644 cmd/pyiceberg_receiver/.pyiceberg.yaml

diff --git a/cmd/pyiceberg_receiver/.pyiceberg.yaml b/cmd/pyiceberg_receiver/.pyiceberg.yaml
new file mode 100644
index 0000000..c1143be
--- /dev/null
+++ b/cmd/pyiceberg_receiver/.pyiceberg.yaml
@@ -0,0 +1,7 @@
+catalog:
+  pgcatalog:
+    type: sql
+    uri: postgresql://username:password@localhost:5432/database
+    init_catalog_tables: true
+    echo: false
+    pool_pre_ping: false
\ No newline at end of file

From bc9620c945b00d806c445b69e42a11db831a7593 Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Sun, 24 Aug 2025 03:02:16 +0300
Subject: [PATCH 05/19] implement `UpdateMeasurements()` in `Receiver` class

---
 cmd/pyiceberg_receiver/iceberg_receiver.py | 26 +++++++++++++++++++++-
 1 file changed, 25 insertions(+), 1 deletion(-)

diff --git a/cmd/pyiceberg_receiver/iceberg_receiver.py b/cmd/pyiceberg_receiver/iceberg_receiver.py
index 80f168d..0454467 100644
--- a/cmd/pyiceberg_receiver/iceberg_receiver.py
+++ b/cmd/pyiceberg_receiver/iceberg_receiver.py
@@ -1,6 +1,30 @@
 from pgwatch_pb2_grpc import ReceiverServicer
 from pgwatch_pb2 import Reply
+import pyarrow as pa
+from pyiceberg.table import Table
+from google.protobuf import json_format
+import json
 
 class Receiver(ReceiverServicer):
+    def __init__(self, tbl: Table):
+        self.tbl = tbl
+        self.arrow_schema = pa.schema([
+            pa.field("DBName", pa.string(), nullable=False),
+            pa.field("MetricName", pa.string(), nullable=False),
+            pa.field("Data", pa.binary(), nullable=False),
+        ])
+
     def UpdateMeasurements(self, request, context):
-        return super().UpdateMeasurements(request, context)
\ No newline at end of file
+        dataRows = [json_format.MessageToDict(msg) for msg in request.Data]
+        jsonData = json.dumps(dataRows)
+
+        row = [{
+            "DBName": request.DBName,
+            "MetricName": request.MetricName,
+            "Data": jsonData,
+        }]
+
+        df = pa.Table.from_pylist(row, schema=self.arrow_schema)
+        self.tbl.append(df)
+
+        return Reply(logmsg="Metrics Inserted in iceberg.")
\ No newline at end of file

From 6b3219080c4a024a8f6ca2ccc6423faa2137e0ee Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Sun, 24 Aug 2025 03:02:38 +0300
Subject: [PATCH 06/19] add requirements.txt

---
 requirements.txt | 4 ++++
 1 file changed, 4 insertions(+)
 create mode 100644 requirements.txt

diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..1692d63
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,4 @@
+grpcio==1.74.0
+grpcio-tools==1.74.0
+protobuf==6.32.0
+setuptools==80.9.0

From 199fb3d6dbaed2ae9a5f2dc9b299513e015b5d31 Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Sun, 24 Aug 2025 05:54:10 +0300
Subject: [PATCH 07/19] [*] move catalog and table init logic to `Receiver`
 constructor

---
 cmd/pyiceberg_receiver/__main__.py         | 35 +--------------
 cmd/pyiceberg_receiver/iceberg_receiver.py | 51 ++++++++++++++++++----
 2 files changed, 43 insertions(+), 43 deletions(-)

diff --git a/cmd/pyiceberg_receiver/__main__.py b/cmd/pyiceberg_receiver/__main__.py
index 0a6ea0a..f8b039a 100644
--- a/cmd/pyiceberg_receiver/__main__.py
+++ b/cmd/pyiceberg_receiver/__main__.py
@@ -3,14 +3,6 @@
 from concurrent import futures
 from iceberg_receiver import Receiver
 from pgwatch_pb2_grpc import add_ReceiverServicer_to_server
-from pyiceberg.catalog import load_catalog
-from pyiceberg.schema import Schema
-from pyiceberg.partitioning import PartitionSpec, PartitionField
-from pyiceberg.transforms import IdentityTransform
-from pyiceberg.types import (
-    StringType,
-    NestedField, 
-)
 
 parser = argparse.ArgumentParser()
 parser.add_argument(
@@ -33,35 +25,10 @@
 )
 args = parser.parse_args()
 
-catalog = load_catalog("pgcatalog")
-catalog.create_namespace_if_not_exists("pgwatch")
-
-schema = Schema(
-    NestedField(field_id=1, name="DBName", field_type=StringType(), required=True),
-    NestedField(field_id=2, name="MetricName", field_type=StringType(), required=True),
-    NestedField(field_id=4, name="Data", field_type=StringType(), required=True),
-)
-
-partition_spec = PartitionSpec(
-    PartitionField(
-        source_id=2, field_id=1000, transform=IdentityTransform(), name="MetricName"
-    ),
-    PartitionField(
-        source_id=1, field_id=1001, transform=IdentityTransform(), name="DBName"
-    ),
-)
-
-tbl = catalog.create_table_if_not_exists(
-    identifier="pgwatch.metrics",
-    schema=schema,
-    location=args.icebergDataDir,
-    partition_spec=partition_spec
-)
-
 def serve(port: int):
     server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
     add_ReceiverServicer_to_server(
-        Receiver(tbl),
+        Receiver(args.icebergDataDir),
         server,
     )
     server.add_insecure_port(f"0.0.0.0:{port}")
diff --git a/cmd/pyiceberg_receiver/iceberg_receiver.py b/cmd/pyiceberg_receiver/iceberg_receiver.py
index 0454467..93d9030 100644
--- a/cmd/pyiceberg_receiver/iceberg_receiver.py
+++ b/cmd/pyiceberg_receiver/iceberg_receiver.py
@@ -1,12 +1,44 @@
+import pyarrow as pa
+import json
 from pgwatch_pb2_grpc import ReceiverServicer
 from pgwatch_pb2 import Reply
-import pyarrow as pa
-from pyiceberg.table import Table
 from google.protobuf import json_format
-import json
+from pyiceberg.catalog import load_catalog
+from pyiceberg.schema import Schema
+from pyiceberg.partitioning import PartitionSpec, PartitionField
+from pyiceberg.transforms import IdentityTransform
+from pyiceberg.types import (
+    NestedField,
+    StringType
+)
 
 class Receiver(ReceiverServicer):
-    def __init__(self, tbl: Table):
+    def __init__(self, icebergDataDir: str):
+        catalog = load_catalog("pgcatalog")
+        catalog.create_namespace_if_not_exists("pgwatch")
+
+        schema = Schema(
+            NestedField(field_id=1, name="DBName", field_type=StringType(), required=True),
+            NestedField(field_id=2, name="MetricName", field_type=StringType(), required=True),
+            NestedField(field_id=4, name="Data", field_type=StringType(), required=True),
+        )
+
+        partition_spec = PartitionSpec(
+            PartitionField(
+                source_id=2, field_id=1000, transform=IdentityTransform(), name="MetricName"
+            ),
+            PartitionField(
+                source_id=1, field_id=1001, transform=IdentityTransform(), name="DBName"
+            ),
+        )
+
+        tbl = catalog.create_table_if_not_exists(
+            identifier="pgwatch.metrics",
+            schema=schema,
+            location=icebergDataDir,
+            partition_spec=partition_spec
+        )        
+
         self.tbl = tbl
         self.arrow_schema = pa.schema([
             pa.field("DBName", pa.string(), nullable=False),
@@ -14,17 +46,18 @@ def __init__(self, tbl: Table):
             pa.field("Data", pa.binary(), nullable=False),
         ])
 
+
     def UpdateMeasurements(self, request, context):
-        dataRows = [json_format.MessageToDict(msg) for msg in request.Data]
-        jsonData = json.dumps(dataRows)
+        data = [json_format.MessageToDict(row) for row in request.Data]
+        dataJson = json.dumps(data)
 
-        row = [{
+        measurement = [{
             "DBName": request.DBName,
             "MetricName": request.MetricName,
-            "Data": jsonData,
+            "Data": dataJson,
         }]
 
-        df = pa.Table.from_pylist(row, schema=self.arrow_schema)
+        df = pa.Table.from_pylist(measurement, schema=self.arrow_schema)
         self.tbl.append(df)
 
         return Reply(logmsg="Metrics Inserted in iceberg.")
\ No newline at end of file

From d2cdf7d6fe65adb039151780bb08fd4fa5151b44 Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Mon, 25 Aug 2025 05:58:26 +0300
Subject: [PATCH 08/19] add tests for iceberg receiver

---
 cmd/pyiceberg_receiver/__main__.py            |  4 +-
 cmd/pyiceberg_receiver/iceberg_receiver.py    |  5 ++-
 .../iceberg_receiver_test.py                  | 43 +++++++++++++++++++
 3 files changed, 48 insertions(+), 4 deletions(-)
 create mode 100644 cmd/pyiceberg_receiver/iceberg_receiver_test.py

diff --git a/cmd/pyiceberg_receiver/__main__.py b/cmd/pyiceberg_receiver/__main__.py
index f8b039a..95aa042 100644
--- a/cmd/pyiceberg_receiver/__main__.py
+++ b/cmd/pyiceberg_receiver/__main__.py
@@ -1,7 +1,7 @@
 import grpc
 import argparse
 from concurrent import futures
-from iceberg_receiver import Receiver
+from iceberg_receiver import IcebergReceiver
 from pgwatch_pb2_grpc import add_ReceiverServicer_to_server
 
 parser = argparse.ArgumentParser()
@@ -28,7 +28,7 @@
 def serve(port: int):
     server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
     add_ReceiverServicer_to_server(
-        Receiver(args.icebergDataDir),
+        IcebergReceiver(args.icebergDataDir),
         server,
     )
     server.add_insecure_port(f"0.0.0.0:{port}")
diff --git a/cmd/pyiceberg_receiver/iceberg_receiver.py b/cmd/pyiceberg_receiver/iceberg_receiver.py
index 93d9030..4bf3e3b 100644
--- a/cmd/pyiceberg_receiver/iceberg_receiver.py
+++ b/cmd/pyiceberg_receiver/iceberg_receiver.py
@@ -12,7 +12,7 @@
     StringType
 )
 
-class Receiver(ReceiverServicer):
+class IcebergReceiver(ReceiverServicer):
     def __init__(self, icebergDataDir: str):
         catalog = load_catalog("pgcatalog")
         catalog.create_namespace_if_not_exists("pgwatch")
@@ -20,7 +20,7 @@ def __init__(self, icebergDataDir: str):
         schema = Schema(
             NestedField(field_id=1, name="DBName", field_type=StringType(), required=True),
             NestedField(field_id=2, name="MetricName", field_type=StringType(), required=True),
-            NestedField(field_id=4, name="Data", field_type=StringType(), required=True),
+            NestedField(field_id=3, name="Data", field_type=StringType(), required=True),
         )
 
         partition_spec = PartitionSpec(
@@ -39,6 +39,7 @@ def __init__(self, icebergDataDir: str):
             partition_spec=partition_spec
         )        
 
+        self.catalog = catalog
         self.tbl = tbl
         self.arrow_schema = pa.schema([
             pa.field("DBName", pa.string(), nullable=False),
diff --git a/cmd/pyiceberg_receiver/iceberg_receiver_test.py b/cmd/pyiceberg_receiver/iceberg_receiver_test.py
new file mode 100644
index 0000000..28a0b68
--- /dev/null
+++ b/cmd/pyiceberg_receiver/iceberg_receiver_test.py
@@ -0,0 +1,43 @@
+import yaml
+import os
+import pytest
+import shutil
+from testcontainers.postgres import PostgresContainer
+from iceberg_receiver import IcebergReceiver
+from pgwatch_pb2 import MeasurementEnvelope, Reply
+
+@pytest.fixture(scope="function", autouse=True)
+def setup_postgres_catalog():
+    with PostgresContainer("postgres:16") as postgres:
+        os.rename(".pyiceberg.yaml", ".pyiceberg2.yaml")
+        iceberg_yaml = {
+            "catalog": {
+                "pgcatalog": {
+                    "uri": postgres.get_connection_url(),
+                    "type": "sql",
+                    "init_catalog_tables": True
+                }
+            }
+        }
+        with open('.pyiceberg.yaml', 'w') as file:
+            yaml.dump(iceberg_yaml, file, default_flow_style=False, allow_unicode=True)
+
+        yield 
+
+        os.rename(".pyiceberg2.yaml", ".pyiceberg.yaml")
+        shutil.rmtree("./data/")
+
+
+def test_IcebergReceiver():
+    recv = IcebergReceiver("./data/") 
+    msg = MeasurementEnvelope(DBName="test",MetricName="test")
+    reply = recv.UpdateMeasurements(msg, None)
+    assert reply == Reply(logmsg="Metrics Inserted in iceberg.")
+
+    paTable = recv.tbl.scan().to_arrow()
+    pyList = paTable.to_pylist()
+
+    assert pyList[0]["DBName"] == "test"
+    assert pyList[0]["MetricName"] == "test"
+
+    recv.catalog.drop_table("pgwatch.metrics")
\ No newline at end of file

From b6cded96f9357fc858371ae94e8e9b76f8c535b0 Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Mon, 25 Aug 2025 06:12:24 +0300
Subject: [PATCH 09/19] extract arrow schema directly from table

---
 cmd/pyiceberg_receiver/iceberg_receiver.py | 6 +-----
 1 file changed, 1 insertion(+), 5 deletions(-)

diff --git a/cmd/pyiceberg_receiver/iceberg_receiver.py b/cmd/pyiceberg_receiver/iceberg_receiver.py
index 4bf3e3b..8236dff 100644
--- a/cmd/pyiceberg_receiver/iceberg_receiver.py
+++ b/cmd/pyiceberg_receiver/iceberg_receiver.py
@@ -41,11 +41,7 @@ def __init__(self, icebergDataDir: str):
 
         self.catalog = catalog
         self.tbl = tbl
-        self.arrow_schema = pa.schema([
-            pa.field("DBName", pa.string(), nullable=False),
-            pa.field("MetricName", pa.string(), nullable=False),
-            pa.field("Data", pa.binary(), nullable=False),
-        ])
+        self.arrow_schema = tbl.schema().as_arrow()
 
 
     def UpdateMeasurements(self, request, context):

From e4483244e315f07d5672d683b4c797f08ba5770d Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Mon, 25 Aug 2025 06:36:31 +0300
Subject: [PATCH 10/19] add docstrings to methods

---
 cmd/pyiceberg_receiver/__main__.py         |  2 ++
 cmd/pyiceberg_receiver/iceberg_receiver.py | 11 ++++++++++-
 2 files changed, 12 insertions(+), 1 deletion(-)

diff --git a/cmd/pyiceberg_receiver/__main__.py b/cmd/pyiceberg_receiver/__main__.py
index 95aa042..3329270 100644
--- a/cmd/pyiceberg_receiver/__main__.py
+++ b/cmd/pyiceberg_receiver/__main__.py
@@ -26,6 +26,8 @@
 args = parser.parse_args()
 
 def serve(port: int):
+    """Starts gRPC server listening on port"""
+
     server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
     add_ReceiverServicer_to_server(
         IcebergReceiver(args.icebergDataDir),
diff --git a/cmd/pyiceberg_receiver/iceberg_receiver.py b/cmd/pyiceberg_receiver/iceberg_receiver.py
index 8236dff..6834074 100644
--- a/cmd/pyiceberg_receiver/iceberg_receiver.py
+++ b/cmd/pyiceberg_receiver/iceberg_receiver.py
@@ -14,6 +14,15 @@
 
 class IcebergReceiver(ReceiverServicer):
     def __init__(self, icebergDataDir: str):
+        """
+        Creates pgwatch.metrics table in pgcatalog if it doesn't exist.
+
+        The table is partitioned by DBName and MetricName fields.
+
+        Args:
+            icebergDataDir (str): Local file system dir path to store data at.
+        """
+
         catalog = load_catalog("pgcatalog")
         catalog.create_namespace_if_not_exists("pgwatch")
 
@@ -33,7 +42,7 @@ def __init__(self, icebergDataDir: str):
         )
 
         tbl = catalog.create_table_if_not_exists(
-            identifier="pgwatch.metrics",
+            identifier=("pgwatch", "metrics"),
             schema=schema,
             location=icebergDataDir,
             partition_spec=partition_spec

From 66be6fc1453d3a2bf4756d1e2975b2e3f0375734 Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Mon, 25 Aug 2025 07:30:10 +0300
Subject: [PATCH 11/19] fix temp `.pyiceberg.yaml` creation logic in tests

---
 .../iceberg_receiver_test.py                  | 30 ++++++++++---------
 1 file changed, 16 insertions(+), 14 deletions(-)

diff --git a/cmd/pyiceberg_receiver/iceberg_receiver_test.py b/cmd/pyiceberg_receiver/iceberg_receiver_test.py
index 28a0b68..3d9ef44 100644
--- a/cmd/pyiceberg_receiver/iceberg_receiver_test.py
+++ b/cmd/pyiceberg_receiver/iceberg_receiver_test.py
@@ -1,16 +1,16 @@
 import yaml
 import os
 import pytest
-import shutil
 from testcontainers.postgres import PostgresContainer
-from iceberg_receiver import IcebergReceiver
 from pgwatch_pb2 import MeasurementEnvelope, Reply
 
-@pytest.fixture(scope="function", autouse=True)
-def setup_postgres_catalog():
+@pytest.fixture(scope="module", autouse=True)
+def setup_catalog(tmp_path_factory):
     with PostgresContainer("postgres:16") as postgres:
-        os.rename(".pyiceberg.yaml", ".pyiceberg2.yaml")
-        iceberg_yaml = {
+        tmp_path = tmp_path_factory.getbasetemp()
+        os.environ["PYICEBERG_HOME"] = tmp_path.as_posix()
+
+        test_iceberg_yaml = {
             "catalog": {
                 "pgcatalog": {
                     "uri": postgres.get_connection_url(),
@@ -19,17 +19,21 @@ def setup_postgres_catalog():
                 }
             }
         }
-        with open('.pyiceberg.yaml', 'w') as file:
-            yaml.dump(iceberg_yaml, file, default_flow_style=False, allow_unicode=True)
+
+        test_pyiceberg_file = tmp_path / ".pyiceberg.yaml"
+        with open(test_pyiceberg_file.as_posix(), 'w') as file:
+            yaml.dump(test_iceberg_yaml, file, default_flow_style=False, allow_unicode=True)
 
         yield 
 
-        os.rename(".pyiceberg2.yaml", ".pyiceberg.yaml")
-        shutil.rmtree("./data/")
 
+def test_IcebergReceiver(tmp_path):
+    # Late import to allow `PYICEERG_HOME` env to be
+    # set by `setup_catalog` fixture before 
+    # pyiceberg reads it on init
+    from iceberg_receiver import IcebergReceiver
 
-def test_IcebergReceiver():
-    recv = IcebergReceiver("./data/") 
+    recv = IcebergReceiver(tmp_path.as_posix()) 
     msg = MeasurementEnvelope(DBName="test",MetricName="test")
     reply = recv.UpdateMeasurements(msg, None)
     assert reply == Reply(logmsg="Metrics Inserted in iceberg.")
@@ -39,5 +43,3 @@ def test_IcebergReceiver():
 
     assert pyList[0]["DBName"] == "test"
     assert pyList[0]["MetricName"] == "test"
-
-    recv.catalog.drop_table("pgwatch.metrics")
\ No newline at end of file

From 9605dd86fc9893e18179411e5341ddb411183db3 Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Mon, 25 Aug 2025 07:31:22 +0300
Subject: [PATCH 12/19] ignore pytest cache

---
 .gitignore | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/.gitignore b/.gitignore
index 404625b..3c10904 100644
--- a/.gitignore
+++ b/.gitignore
@@ -2,4 +2,5 @@
 .vscode/
 *pb.go
 *pb2*.py
-__pycache__/
\ No newline at end of file
+__pycache__/
+.pytest_cache/
\ No newline at end of file

From b60ed9b5c50d4058c9766a6237194677d7c8e457 Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Mon, 25 Aug 2025 08:37:54 +0300
Subject: [PATCH 13/19] run python tests in ci

---
 .github/workflows/test.yml | 17 ++++++++++++++++-
 1 file changed, 16 insertions(+), 1 deletion(-)

diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index fc059bd..071b5f0 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -47,11 +47,26 @@ jobs:
     - name: Setup Protobuf
       uses: ./.github/actions/setup-protobuf
 
-    - name: Test
+    - name: Set up Python
+      uses: actions/setup-python@v5
+      with:
+        python-version: '3.12' 
+
+    - name: Install dependencies
+      run: |
+        python3 -m pip install --upgrade pip
+        pip install pytest
+        pip install -r requirements.txt
+
+    - name: Golang Tests
       run: |
         go generate ./sinks/pb
         go test -timeout 10m -failfast -v -coverprofile=profile.cov ./...
 
+    - name: Python Tests
+      run: |
+        pytest
+
     - name: Coveralls
       uses: coverallsapp/github-action@v2
       with:

From 17645e597d9110db589ccc9279a7e4ea165269d2 Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Mon, 25 Aug 2025 09:15:42 +0300
Subject: [PATCH 14/19] add README for iceberg receiver

---
 cmd/pyiceberg_receiver/README.md | 45 ++++++++++++++++++++++++++++++++
 1 file changed, 45 insertions(+)
 create mode 100644 cmd/pyiceberg_receiver/README.md

diff --git a/cmd/pyiceberg_receiver/README.md b/cmd/pyiceberg_receiver/README.md
new file mode 100644
index 0000000..c4bddf4
--- /dev/null
+++ b/cmd/pyiceberg_receiver/README.md
@@ -0,0 +1,45 @@
+# Iceberg Receiver
+
+A gRPC server that writes metrics received from pgwatch in Iceberg Table format.
+
+- The server assumes a PostgreSQL catalog is used and creates `pgwatch` namespace and `pgwatch.metrics` table within it if they don't exist.
+- The table is partitioned by `MetricName` and `DBName` (in order).
+- Metrics are written in the local file system as Apache Arrow records with the following schema:
+    ```python
+    Schema(
+        NestedField(field_id=1, name="DBName", field_type=StringType(), required=True),
+        NestedField(field_id=2, name="MetricName", field_type=StringType(), required=True),
+        NestedField(field_id=3, name="Data", field_type=StringType(), required=True),
+    )
+    ```
+- Catalog configurations should be provided in [.pyiceberg.yaml](./.pyiceberg.yaml) file under `pgcatalog` see [PyIceberg SQL Catalog](https://py.iceberg.apache.org/configuration/#sql-catalog) for details.
+
+## Flags
+
+```bash
+usage: pyiceberg_receiver [-h] -p PORT -d DIR
+
+options:
+  -h, --help            show this help message and exit
+  -p PORT, --port PORT  The port number to use for the gRPC server.
+  -d DIR, --iceberg-data-dir DIR
+                        Directory to store iceberg tables in.
+```
+
+## Usage example
+
+```bash
+# generate python gRPC code from protobuf
+python3 -m grpc_tools.protoc -I sinks/pb --python_out=cmd/pyiceberg_receiver --grpc_python_out=cmd/pyiceberg_receiver sinks/pb/pgwatch.proto
+# tell PyIceberg about the dir to look for .pyiceberg.yaml in
+export PYICEBERG_HOME="cmd/pyiceberg_receiver"
+# run the server
+python3 cmd/pyiceberg_receiver -p <grpc-server-port> -d <dir-path>
+```
+
+## TODO
+
+- [ ] Use object storage instead of the local file system.
+- [ ] Support TLS over the gRPC connection.
+- [ ] Add authentication interceptor.
+- [ ] Cache measurements to minimize the number of Parquet files written.
\ No newline at end of file

From 96104f0d18c06707422db140e5cc3d0c74dc6e19 Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Mon, 25 Aug 2025 09:34:13 +0300
Subject: [PATCH 15/19] move iceberg sink `requirements.txt` to
 `pyiceberg_receiver/`

---
 requirements.txt => cmd/pyiceberg_receiver/requirements.txt | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 rename requirements.txt => cmd/pyiceberg_receiver/requirements.txt (100%)

diff --git a/requirements.txt b/cmd/pyiceberg_receiver/requirements.txt
similarity index 100%
rename from requirements.txt
rename to cmd/pyiceberg_receiver/requirements.txt

From 5019067a8b729adab946db46ed86d6ff67d42db6 Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Mon, 25 Aug 2025 09:34:41 +0300
Subject: [PATCH 16/19] add install `requirements.txt` step to README

---
 cmd/pyiceberg_receiver/README.md | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/cmd/pyiceberg_receiver/README.md b/cmd/pyiceberg_receiver/README.md
index c4bddf4..dbe2fe0 100644
--- a/cmd/pyiceberg_receiver/README.md
+++ b/cmd/pyiceberg_receiver/README.md
@@ -31,6 +31,8 @@ options:
 ```bash
 # generate python gRPC code from protobuf
 python3 -m grpc_tools.protoc -I sinks/pb --python_out=cmd/pyiceberg_receiver --grpc_python_out=cmd/pyiceberg_receiver sinks/pb/pgwatch.proto
+# install dependencies
+pip install -r requirements.txt
 # tell PyIceberg about the dir to look for .pyiceberg.yaml in
 export PYICEBERG_HOME="cmd/pyiceberg_receiver"
 # run the server

From e615ba4bc55ab2d2fa7697e18d5fc92fbb6f1ae5 Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Mon, 25 Aug 2025 09:36:53 +0300
Subject: [PATCH 17/19] increase golang tests timeout

---
 .github/workflows/test.yml | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 071b5f0..8f9b6b0 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -56,12 +56,12 @@ jobs:
       run: |
         python3 -m pip install --upgrade pip
         pip install pytest
-        pip install -r requirements.txt
+        pip install -r cmd/pyiceberg_receiver/requirements.txt
 
     - name: Golang Tests
       run: |
         go generate ./sinks/pb
-        go test -timeout 10m -failfast -v -coverprofile=profile.cov ./...
+        go test -timeout 20m -failfast -v -coverprofile=profile.cov ./...
 
     - name: Python Tests
       run: |

From b7ec3e11cc74bfcbfe2ce3a1cb5334bc638070bf Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Mon, 25 Aug 2025 09:53:48 +0300
Subject: [PATCH 18/19] update iceberg sink `requirements.txt`

---
 cmd/pyiceberg_receiver/requirements.txt | 38 +++++++++++++++++++++++++
 1 file changed, 38 insertions(+)

diff --git a/cmd/pyiceberg_receiver/requirements.txt b/cmd/pyiceberg_receiver/requirements.txt
index 1692d63..3e4b484 100644
--- a/cmd/pyiceberg_receiver/requirements.txt
+++ b/cmd/pyiceberg_receiver/requirements.txt
@@ -1,4 +1,42 @@
+annotated-types==0.7.0
+cachetools==5.5.2
+certifi==2025.8.3
+charset-normalizer==3.4.3
+click==8.2.1
+docker==7.1.0
+fsspec==2025.7.0
+greenlet==3.2.4
 grpcio==1.74.0
 grpcio-tools==1.74.0
+idna==3.10
+iniconfig==2.1.0
+markdown-it-py==4.0.0
+mdurl==0.1.2
+mmh3==5.2.0
+packaging==25.0
+pluggy==1.6.0
 protobuf==6.32.0
+psycopg2-binary==2.9.10
+pyarrow==21.0.0
+pydantic==2.11.7
+pydantic_core==2.33.2
+Pygments==2.19.2
+pyiceberg==0.9.1
+pyparsing==3.2.3
+pytest==8.4.1
+python-dateutil==2.9.0.post0
+python-dotenv==1.1.1
+PyYAML==6.0.2
+requests==2.32.5
+rich==13.9.4
 setuptools==80.9.0
+six==1.17.0
+sortedcontainers==2.4.0
+SQLAlchemy==2.0.43
+strictyaml==1.7.3
+tenacity==9.1.2
+testcontainers==4.12.0
+typing-inspection==0.4.1
+typing_extensions==4.14.1
+urllib3==2.5.0
+wrapt==1.17.3

From da8bbe793ba5da9eaf78f76b84887537efc000e6 Mon Sep 17 00:00:00 2001
From: 0xgouda <gouda0x@gmail.com>
Date: Mon, 25 Aug 2025 10:08:42 +0300
Subject: [PATCH 19/19] generate python grpc code in before python tests

---
 .github/workflows/test.yml | 1 +
 1 file changed, 1 insertion(+)

diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 8f9b6b0..01555ac 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -65,6 +65,7 @@ jobs:
 
     - name: Python Tests
       run: |
+        python3 -m grpc_tools.protoc -I sinks/pb --python_out=cmd/pyiceberg_receiver --grpc_python_out=cmd/pyiceberg_receiver sinks/pb/pgwatch.proto
         pytest
 
     - name: Coveralls