sql: improve PG subsource name resolution errors

Include a few improvements to resolving table names when creating
PG sources, including:

Duplicate subsource names are easy to encounter, but the error they
produced didn't provide any context around how to fix the issue.
We can instead produce an error that gives users more guidance.

FOR SCHEMAS (..) didn't error if a referenced schema didn't
identify any tables in the publication; this seems like a potential
source of confusion for users.

Other misc. simplifications.

Author: Sean Loiselle

src/sql/src/plan/error.rs

@@ -20,6 +20,7 @@ use mz_expr::EvalError;
 use mz_ore::error::ErrorExt;
 use mz_ore::stack::RecursionLimitError;
 use mz_ore::str::{separated, StrExt};
+use mz_postgres_util::PostgresError;
 use mz_repr::adt::char::InvalidCharLengthError;
 use mz_repr::adt::numeric::InvalidNumericMaxScaleError;
 use mz_repr::adt::system::Oid;
@@ -107,7 +108,7 @@ pub enum PlanError {
         schema_lookup: String,
         cause: Arc<dyn Error + Send + Sync>,
     },
-    FetchingPostgresPublicationInfoFailed {
+    PostgresConnectionErr {
         cause: Arc<mz_postgres_util::PostgresError>,
     },
     InvalidProtobufSchema {
@@ -154,6 +155,14 @@ pub enum PlanError {
         cluster_name: String,
         linked_object_name: String,
     },
+    EmptyPublication(String),
+    DuplicateSubsourceReference {
+        name: UnresolvedItemName,
+        upstream_references: Vec<UnresolvedItemName>,
+    },
+    PostgresDatabaseMissingFilteredSchemas {
+        schemas: Vec<String>,
+    },
     // TODO(benesch): eventually all errors should be structured.
     Unstructured(String),
 }
@@ -169,9 +178,7 @@ impl PlanError {
     pub fn detail(&self) -> Option<String> {
         match self {
             Self::FetchingCsrSchemaFailed { cause, .. } => Some(cause.to_string_with_causes()),
-            Self::FetchingPostgresPublicationInfoFailed { cause } => {
-                Some(cause.to_string_with_causes())
-            }
+            Self::PostgresConnectionErr { cause } => Some(cause.to_string_with_causes()),
             Self::InvalidProtobufSchema { cause } => Some(cause.to_string_with_causes()),
             Self::InvalidOptionValue { err, .. } => err.detail(),
             _ => None,
@@ -205,7 +212,7 @@ impl PlanError {
                 as text."
                     .into(),
             ),
-            Self::FetchingPostgresPublicationInfoFailed { cause } => {
+            Self::PostgresConnectionErr { cause } => {
                 if let Some(cause) = cause.source() {
                     if let Some(cause) = cause.downcast_ref::<io::Error>() {
                         if cause.kind() == io::ErrorKind::TimedOut {
@@ -239,6 +246,9 @@ impl PlanError {
                 let supported_azs_str = supported_azs.iter().join("\n  ");
                 Some(format!("Did you supply an availability zone name instead of an ID? Known availability zone IDs:\n  {}", supported_azs_str))
             }
+            Self::DuplicateSubsourceReference { .. } => {
+                Some("Specify target table names using FOR TABLES (foo AS bar), or limit the upstream tables using FOR SCHEMAS (foo)".into())
+            }
             _ => None,
         }
     }
@@ -358,8 +368,8 @@ impl fmt::Display for PlanError {
             Self::FetchingCsrSchemaFailed { schema_lookup, .. } => {
                 write!(f, "failed to fetch schema {schema_lookup} from schema registry")
             }
-            Self::FetchingPostgresPublicationInfoFailed { .. } => {
-                write!(f, "failed to fetch publication information from PostgreSQL database")
+            Self::PostgresConnectionErr { .. } => {
+                write!(f, "failed to connect to PostgreSQL database")
             }
             Self::InvalidProtobufSchema { .. } => {
                 write!(f, "invalid protobuf schema")
@@ -401,6 +411,13 @@ impl fmt::Display for PlanError {
             Self::InvalidSchemaName => write!(f, "no schema has been selected to create in"),
             Self::ItemAlreadyExists { name, item_type } => write!(f, "{item_type} {} already exists", name.quoted()),
             Self::ModifyLinkedCluster {cluster_name, ..} => write!(f, "cannot modify linked cluster {}", cluster_name.quoted()),
+            Self::EmptyPublication(publication) => write!(f, "PostgreSQL PUBLICATION {publication} is empty"),
+            Self::DuplicateSubsourceReference { name, upstream_references } => {
+                write!(f, "multiple tables with name {}: {}", name.to_ast_string_stable(), itertools::join(upstream_references.iter().map(|n| n.to_ast_string_stable()), ", "))
+            },
+            Self::PostgresDatabaseMissingFilteredSchemas { schemas} => {
+                write!(f, "FOR SCHEMAS (..) included {}, but PostgreSQL database has no schema with that name", itertools::join(schemas.iter(), ", "))
+            }
         }
     }
 }
@@ -474,6 +491,12 @@ impl From<ParserError> for PlanError {
     }
 }
 
+impl From<PostgresError> for PlanError {
+    fn from(e: PostgresError) -> PlanError {
+        PlanError::PostgresConnectionErr { cause: Arc::new(e) }
+    }
+}
+
 struct ColumnDisplay<'a> {
     table: &'a Option<PartialItemName>,
     column: &'a ColumnName,

src/sql/src/pure.rs

@@ -17,6 +17,7 @@ use std::path::Path;
 use std::sync::Arc;
 
 use anyhow::anyhow;
+use itertools::Itertools;
 use mz_repr::adt::system::Oid;
 use prost::Message;
 use protobuf_native::compiler::{SourceTreeDescriptorDatabase, VirtualSourceTree};
@@ -291,11 +292,11 @@ pub async fn purify_create_source(
                 .config(&*connection_context.secrets_reader)
                 .await?;
             let publication_tables =
-                mz_postgres_util::publication_info(&config, &publication, None)
-                    .await
-                    .map_err(|cause| PlanError::FetchingPostgresPublicationInfoFailed {
-                        cause: Arc::new(cause),
-                    })?;
+                mz_postgres_util::publication_info(&config, &publication, None).await?;
+
+            if publication_tables.is_empty() {
+                return Err(PlanError::EmptyPublication(publication));
+            }
 
             // An index from table name -> schema name -> database name -> PostgresTableDesc
             let mut tables_by_name = BTreeMap::new();
@@ -316,9 +317,6 @@ pub async fn purify_create_source(
             let mut validated_requested_subsources = vec![];
             match referenced_subsources {
                 Some(ReferencedSubsources::All) => {
-                    if publication_tables.is_empty() {
-                        sql_bail!("FOR ALL TABLES is only valid for non-empty publications");
-                    }
                     for table in &publication_tables {
                         let upstream_name = UnresolvedItemName::qualified(&[
                             &connection.database,
@@ -330,26 +328,41 @@ pub async fn purify_create_source(
                     }
                 }
                 Some(ReferencedSubsources::SubsetSchemas(schemas)) => {
-                    let schemas: BTreeSet<_> = schemas.iter().map(|s| s.as_str()).collect();
+                    let available_schemas: BTreeSet<_> = mz_postgres_util::get_schemas(&config)
+                        .await?
+                        .into_iter()
+                        .map(|s| s.name)
+                        .collect();
+
+                    let requested_schemas: BTreeSet<_> =
+                        schemas.iter().map(|s| s.as_str().to_string()).collect();
+
+                    let missing_schemas: Vec<_> = requested_schemas
+                        .difference(&available_schemas)
+                        .map(|s| s.to_string())
+                        .collect();
+
+                    if !missing_schemas.is_empty() {
+                        return Err(PlanError::PostgresDatabaseMissingFilteredSchemas {
+                            schemas: missing_schemas,
+                        });
+                    }
+
                     for table in &publication_tables {
-                        if !schemas.contains(table.namespace.as_str()) {
+                        if !requested_schemas.contains(table.namespace.as_str()) {
                             continue;
                         }
 
-                        let upstream_name = UnresolvedObjectName::qualified(&[
+                        let upstream_name = UnresolvedItemName::qualified(&[
                             &connection.database,
                             &table.namespace,
                             &table.name,
                         ]);
-                        let subsource_name = UnresolvedObjectName::unqualified(&table.name);
+                        let subsource_name = UnresolvedItemName::unqualified(&table.name);
                         validated_requested_subsources.push((upstream_name, subsource_name, table));
                     }
                 }
                 Some(ReferencedSubsources::SubsetTables(subsources)) => {
-                    // TODO: can subsources be empty?
-                    if publication_tables.is_empty() {
-                        sql_bail!("FOR TABLES (..) is only valid for non-empty publications");
-                    }
                     // The user manually selected a subset of upstream tables so we need to
                     // validate that the names actually exist and are not ambiguous
 
@@ -366,11 +379,34 @@ pub async fn purify_create_source(
 
             if validated_requested_subsources.is_empty() {
                 sql_bail!(
-                    "Postgres source must ingest at least one table, but {} matched none",
+                    "[internal error]: Postgres source must ingest at least one table, but {} matched none",
                     referenced_subsources.as_ref().unwrap().to_ast_string()
                 );
             }
 
+            // This condition would get caught during the catalog transaction, but produces a
+            // vague, non-contextual error. Instead, error here so we can suggest to the user
+            // how to fix the problem.
+            if let Some(name) = validated_requested_subsources
+                .iter()
+                .map(|(_, subsource_name, _)| subsource_name)
+                .duplicates()
+                .next()
+                .cloned()
+            {
+                let mut upstream_references: Vec<_> = validated_requested_subsources
+                    .into_iter()
+                    .filter_map(|(u, t, _)| if t == name { Some(u) } else { None })
+                    .collect();
+
+                upstream_references.sort();
+
+                return Err(PlanError::DuplicateSubsourceReference {
+                    name,
+                    upstream_references,
+                });
+            }
+
             let mut text_cols_dict: BTreeMap<u32, BTreeSet<String>> = BTreeMap::new();
 
             for name in text_columns.iter_mut() {

test/pg-cdc/pg-cdc.td

@@ -153,7 +153,7 @@ $ postgres-verify-slot connection=postgres://postgres:postgres@postgres slot=mat
 ! CREATE SOURCE "no_such_host"
   FROM POSTGRES CONNECTION no_such_host (PUBLICATION 'mz_source');
 # TODO: assert on `detail` here.
-exact:failed to fetch publication information from PostgreSQL database
+exact:failed to connect to PostgreSQL database
 
 > CREATE CONNECTION no_such_port TO POSTGRES (
     HOST postgres,
@@ -165,7 +165,7 @@ exact:failed to fetch publication information from PostgreSQL database
 ! CREATE SOURCE "no_such_port"
   FROM POSTGRES CONNECTION no_such_port (PUBLICATION 'mz_source');
 # TODO: assert on `detail` here.
-exact:failed to fetch publication information from PostgreSQL database
+exact:failed to connect to PostgreSQL database
 
 > CREATE CONNECTION no_such_user TO POSTGRES (
     HOST postgres,
@@ -176,7 +176,7 @@ exact:failed to fetch publication information from PostgreSQL database
 ! CREATE SOURCE "no_such_user"
   FROM POSTGRES CONNECTION no_such_user (PUBLICATION 'mz_source');
 # TODO: assert on `detail` here.
-exact:failed to fetch publication information from PostgreSQL database
+exact:failed to connect to PostgreSQL database
 
 > CREATE SECRET badpass AS 'badpass'
 > CREATE CONNECTION no_such_password TO POSTGRES (
@@ -188,7 +188,7 @@ exact:failed to fetch publication information from PostgreSQL database
 ! CREATE SOURCE "no_such_password"
   FROM POSTGRES CONNECTION no_such_password (PUBLICATION 'mz_source');
 # TODO: assert on `detail` here.
-exact:failed to fetch publication information from PostgreSQL database
+exact:failed to connect to PostgreSQL database
 
 > CREATE CONNECTION no_such_dbname TO POSTGRES (
     HOST postgres,
@@ -199,12 +199,12 @@ exact:failed to fetch publication information from PostgreSQL database
 ! CREATE SOURCE "no_such_dbname"
   FROM POSTGRES CONNECTION no_such_dbname (PUBLICATION 'mz_source');
 # TODO: assert on `detail` here.
-exact:failed to fetch publication information from PostgreSQL database
+exact:failed to connect to PostgreSQL database
 
 ! CREATE SOURCE "no_such_publication"
   FROM POSTGRES CONNECTION pg (PUBLICATION 'no_such_publication');
 # TODO: assert on `detail` here.
-exact:failed to fetch publication information from PostgreSQL database
+exact:failed to connect to PostgreSQL database
 
 ! CREATE SOURCE "mz_source"
   FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source')
@@ -221,11 +221,6 @@ regex: the following columns contain unsupported types:\npostgres.public.enum_ta
   );
 regex: the following columns contain unsupported types:\npostgres\.public\.another_enum_table\."колона" \(OID \d+\)\npostgres\.public\.enum_table\.a \(OID \d+\)
 
-! CREATE SOURCE "mz_source"
-  FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source')
-  FOR SCHEMAS (dne);
-regex: Postgres source must ingest at least one table, but FOR SCHEMAS \(dne\) matched none
-
 #
 # Establish direct replication
 #
@@ -573,31 +568,34 @@ contains: invalid TEXT COLUMNS option value: table table_dne not found in source
     PUBLICATION 'mz_source',
     TEXT COLUMNS [another_enum_table."колона"]
   )
-  FOR ALL TABLES;
+  FOR TABLES(
+    conflict_schema.another_enum_table AS conflict_enum,
+    public.another_enum_table AS public_enum
+  );
 contains: invalid TEXT COLUMNS option value: table another_enum_table is ambiguous, consider specifying the schema
 
 ! CREATE SOURCE enum_source
   FROM POSTGRES CONNECTION pg (
     PUBLICATION 'mz_source',
     TEXT COLUMNS [foo]
   )
-  FOR ALL TABLES;
+  FOR SCHEMAS (public);
 contains: invalid TEXT COLUMNS option value: column name 'foo' must have at least a table qualification
 
 ! CREATE SOURCE enum_source
   FROM POSTGRES CONNECTION pg (
     PUBLICATION 'mz_source',
     TEXT COLUMNS [foo.bar.qux.qax.foo]
   )
-  FOR ALL TABLES;
+  FOR SCHEMAS (public);
 contains: invalid TEXT COLUMNS option value: qualified name did not have between 1 and 3 components: foo.bar.qux.qax
 
 ! CREATE SOURCE enum_source
   FROM POSTGRES CONNECTION pg (
     PUBLICATION 'mz_source',
     TEXT COLUMNS [enum_table.a, enum_table.a]
   )
-  FOR ALL TABLES;
+  FOR SCHEMAS (public);
 contains: invalid TEXT COLUMNS option value: unexpected multiple references to postgres.public.enum_table.a
 
 # utf8_table is not part of mz_source_narrow publication
@@ -606,7 +604,7 @@ contains: invalid TEXT COLUMNS option value: unexpected multiple references to p
     PUBLICATION 'mz_source_narrow',
     TEXT COLUMNS [enum_table.a, utf8_table.f1]
   )
-  FOR ALL TABLES;
+  FOR SCHEMAS (public);
 contains: invalid TEXT COLUMNS option value: table utf8_table not found in source
 
 # n.b includes a reference to pk_table, which is not a table that's part of the

test/pg-cdc/subsource-resolution-duplicates.td

@@ -0,0 +1,55 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# IMPORTANT: The Postgres server has a custom pg_hba.conf that only
+# accepts connections from specific users. You will have to update
+# pg_hba.conf if you modify the existing user names or add new ones.
+
+> CREATE SECRET pgpass AS 'postgres'
+> CREATE CONNECTION pg TO POSTGRES (
+    HOST postgres,
+    DATABASE postgres,
+    USER postgres,
+    PASSWORD SECRET pgpass
+  )
+
+$ postgres-execute connection=postgres://postgres:postgres@postgres
+ALTER USER postgres WITH replication;
+DROP SCHEMA IF EXISTS public CASCADE;
+CREATE SCHEMA public;
+DROP SCHEMA IF EXISTS other CASCADE;
+CREATE SCHEMA other;
+
+DROP PUBLICATION IF EXISTS mz_source;
+CREATE PUBLICATION mz_source FOR ALL TABLES;
+
+CREATE TABLE t (f1 INT);
+INSERT INTO t VALUES (1);
+ALTER TABLE t REPLICA IDENTITY FULL;
+
+CREATE TABLE other.t (f1 INT);
+INSERT INTO other.t VALUES (1);
+ALTER TABLE other.t REPLICA IDENTITY FULL;
+
+! CREATE SOURCE mz_source
+  FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source')
+  FOR ALL TABLES;
+contains:multiple tables with name "t": "postgres"."other"."t", "postgres"."public"."t"
+
+> CREATE SOURCE mz_source
+  FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source')
+  FOR SCHEMAS (other);
+
+> SHOW sources
+ mz_source          postgres  4
+ mz_source_progress subsource <null>
+ t                  subsource <null>
+
+$ postgres-execute connection=postgres://postgres:postgres@postgres
+DROP SCHEMA other CASCADE;