add a basic tls auth layer to the postgres connector (#9817)

GitOrigin-RevId: f723f2db2274a8841a62470e51b8d2ed5d190d0a

Author: zxqfd555

CHANGELOG.md

@@ -9,6 +9,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - `pw.io.kafka.read` and `pw.io.kafka.write` connectors now support OAUTHBEARER authentication.
 - `pw.io.mongodb.write` connector now supports an `output_table_type` parameter with two modes: `stream_of_changes` (default) and `snapshot`. In `snapshot` mode, the connector maintains the current state of the Pathway table in MongoDB using the `_id` field as the primary key, while `stream_of_changes` preserves the existing behavior by writing all events with `time` and `diff` flags to reflect transactional minibatches and the nature of each change.
 - Workers can now automatically scale up or down based on pipeline load, using a configurable monitoring window. This feature requires persistence to be enabled and can be configured via `worker_scaling_enabled` and `workload_tracking_window_ms` in `pw.persistence.Config`. Please refer to the tutorial for more details.
+- `pw.io.postgres.write` now properly supports TLS configuration via `sslmode` and `sslrootcert` connection string parameters.
 
 ## [0.29.0] - 2026-01-22
 

Cargo.lock

@@ -63,6 +63,7 @@ log = { version = "0.4.27", features = ["std"] }
 lz4_flex = "0.11.5"
 mongodb = { version = "3.2.2", features = ["sync"] }
 mysql = "26.0.1"
+native-tls = "0.2.14"
 ndarray = { version = "0.15.6", features = ["serde"] }
 nix = { version = "0.30.1", features = ["fs", "user", "resource"] }
 num-integer = "0.1.46"
@@ -75,6 +76,7 @@ opentelemetry_sdk = { version = "0.30.0", features = ["rt-tokio", "rt-tokio-curr
 ordered-float = { version = "4.6.0", features = ["serde"] }
 pgvector = { version = "0.4.1", features = ["postgres", "halfvec"] }
 postgres = { version = "0.19.10", features = ["with-chrono-0_4", "with-serde_json-1"] }
+postgres-native-tls = "0.5.2"
 prometheus-client = "0.23.1"
 pyo3 = { version = "0.25.0", features = ["abi3-py310", "multiple-pymethods"] }
 pyo3-async-runtimes = "0.25.0"

Cargo.toml

@@ -6,6 +6,7 @@
     MySQLContext,
     PgvectorContext,
     PostgresContext,
+    PostgresWithTlsContext,
     QuestDBContext,
 )
 
@@ -15,6 +16,11 @@ def postgres():
     return PostgresContext()
 
 
+@pytest.fixture
+def postgres_with_tls():
+    return PostgresWithTlsContext()
+
+
 @pytest.fixture
 def pgvector():
     return PgvectorContext()

integration_tests/db_connectors/conftest.py

@@ -4,13 +4,25 @@
 
 import pandas as pd
 import pytest
-from utils import MYSQL_CONNECTION_STRING, MYSQL_DB_NAME, SimpleObject
+from utils import (
+    MYSQL_CONNECTION_STRING,
+    MYSQL_DB_NAME,
+    SimpleObject,
+    is_mysql_reachable,
+)
 
 import pathway as pw
 from pathway.internals import api
 from pathway.internals.parse_graph import G
 
+# FIXME: the mysql Docker container used in the integration tests is unstable and
+# sometimes is unavailable, even though the healthcheck passes
+xfail_if_mysql_failed_to_start = pytest.mark.xfail(
+    not is_mysql_reachable(), reason="mysql has failed to start"
+)
 
+
+@xfail_if_mysql_failed_to_start
 @pytest.mark.parametrize("output_table_type", ["stream_of_changes", "snapshot"])
 @pytest.mark.parametrize("init_mode", ["default", "create_if_not_exists", "replace"])
 def test_outputs(output_table_type, init_mode, mysql):
@@ -98,6 +110,7 @@ def run(offset: int):
         ]
 
 
+@xfail_if_mysql_failed_to_start
 @pytest.mark.parametrize("are_types_optional", [False, True])
 @pytest.mark.parametrize("init_mode", ["create_if_not_exists", "replace"])
 def test_different_types_schema_and_serialization(init_mode, are_types_optional, mysql):
@@ -223,6 +236,7 @@ class InputSchema(pw.Schema):  # type:ignore
             assert column_props.is_nullable == are_types_optional, column_name
 
 
+@xfail_if_mysql_failed_to_start
 def test_output_snapshot_overwrite_by_key(mysql):
     words = ["one", "two", "three", "one", "four", "two", "one", "one", "two", "four"]
 
@@ -254,6 +268,7 @@ class InputSchema(pw.Schema):
     assert external_schema["count"].type_name == "bigint"
 
 
+@xfail_if_mysql_failed_to_start
 def test_mysql_overwrites_old_snapshot(mysql):
     table_name = mysql.random_table_name()
 
@@ -312,6 +327,7 @@ def run(offset: int):
     ]
 
 
+@xfail_if_mysql_failed_to_start
 def test_mysql_composite_snapshot_key(mysql):
     table_name = mysql.random_table_name()
 
@@ -357,6 +373,7 @@ def run(offset: int):
     ]
 
 
+@xfail_if_mysql_failed_to_start
 @pytest.mark.parametrize("malfomed_primary_key", [None, []])
 def test_mysql_no_snapshot_key(malfomed_primary_key, mysql):
     table_name = mysql.random_table_name()
@@ -390,6 +407,7 @@ class InputSchema(pw.Schema):
         pw.run()
 
 
+@xfail_if_mysql_failed_to_start
 def test_mysql_single_column_snapshot_mode(mysql):
     table_name = mysql.random_table_name()