fail fast in pw.io.postgres.read if the publication doesn't exist (#9955)

GitOrigin-RevId: d923a7039c3f4993cc1d06bfae2566f79585f940

Author: zxqfd555

integration_tests/db_connectors/test_postgres.py

@@ -2115,3 +2115,34 @@ def stream_target():
     assert VALID_STREAMING_ID in ids_out, "Valid streaming row must be present"
     for oor_id in OOR_IDS:
         assert oor_id not in ids_out, f"Out-of-range row {oor_id} must be skipped"
+
+
+def test_no_publication(tmp_path, postgres):
+    class InputSchema(pw.Schema):
+        value: str
+
+    output_path = tmp_path / "output.jsonl"
+    table_name = postgres.random_table_name()
+
+    postgres.execute_sql(
+        f"""
+        CREATE TABLE {table_name} (
+            value TEXT PRIMARY KEY
+        );
+        """
+    )
+
+    postgres.execute_sql(f"INSERT INTO {table_name} (value) VALUES ('hello');")
+    table = pw.io.postgres.read(
+        postgres_settings=POSTGRES_SETTINGS,
+        table_name=table_name,
+        schema=InputSchema,
+        mode="streaming",
+        publication_name=f"{table_name}_pub",
+        autocommit_duration_ms=10,
+    )
+    pw.io.jsonlines.write(table, output_path)
+    with pytest.raises(
+        RuntimeError, match=f"Publication '{table_name}_pub' does not exist"
+    ):
+        pw.run()

src/connectors/postgres.rs

@@ -1124,6 +1124,9 @@ pub enum ReplicationError {
     #[error(transparent)]
     WalReader(#[from] pg_walstream::error::ReplicationError),
 
+    #[error("Publication '{name}' does not exist")]
+    PublicationNotFound { name: String },
+
     #[error(transparent)]
     Query(#[from] postgres::Error),
 
@@ -2079,6 +2082,18 @@ impl PsqlReader {
         Self::validate_table_schema(&mut client, &table_ctx)?;
 
         let wal_reader = if let Some(replication_settings) = replication_settings {
+            let rows = client
+                .query(
+                    "SELECT 1 FROM pg_publication WHERE pubname = $1",
+                    &[&replication_settings.publication_name],
+                )
+                .map_err(ReplicationError::Query)?;
+            if rows.is_empty() {
+                return Err(ReplicationError::PublicationNotFound {
+                    name: replication_settings.publication_name.clone(),
+                }
+                .into());
+            }
             Some(WalReader::new(replication_settings, table_ctx.clone())?)
         } else {
             None