feat(ast): add CreatePublication and CreateSubscription AST nodes

Add PublicationTarget enum (AllTables, Tables, TablesInSchema),
CreatePublication struct, and CreateSubscription struct to ddl.rs.
Add Statement variants, Display arms, and re-exports in mod.rs.
Add Span::empty() arms in spans.rs.
Add PUBLICATION and SUBSCRIPTION to keywords.rs.

Author: fmguerreiro

src/ast/ddl.rs

@@ -6136,3 +6136,111 @@ impl From<CreateTextSearchTemplate> for crate::ast::Statement {
         crate::ast::Statement::CreateTextSearchTemplate(v)
     }
 }
+
+/// The target of a `CREATE PUBLICATION` statement: which rows to publish.
+///
+/// See <https://www.postgresql.org/docs/current/sql-createpublication.html>
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub enum PublicationTarget {
+    /// `FOR ALL TABLES`
+    AllTables,
+    /// `FOR TABLE table [, ...]`
+    Tables(Vec<ObjectName>),
+    /// `FOR TABLES IN SCHEMA schema [, ...]`
+    TablesInSchema(Vec<Ident>),
+}
+
+impl fmt::Display for PublicationTarget {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            PublicationTarget::AllTables => write!(f, "FOR ALL TABLES"),
+            PublicationTarget::Tables(tables) => {
+                write!(f, "FOR TABLE {}", display_comma_separated(tables))
+            }
+            PublicationTarget::TablesInSchema(schemas) => {
+                write!(
+                    f,
+                    "FOR TABLES IN SCHEMA {}",
+                    display_comma_separated(schemas)
+                )
+            }
+        }
+    }
+}
+
+/// A `CREATE PUBLICATION` statement.
+///
+/// Note: this is a PostgreSQL-specific statement.
+/// <https://www.postgresql.org/docs/current/sql-createpublication.html>
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub struct CreatePublication {
+    /// The publication name.
+    pub name: Ident,
+    /// Optional target specification (`FOR ALL TABLES`, `FOR TABLE ...`, or `FOR TABLES IN SCHEMA ...`).
+    pub target: Option<PublicationTarget>,
+    /// Optional `WITH (key = value, ...)` clause.
+    pub with_options: Vec<SqlOption>,
+}
+
+impl fmt::Display for CreatePublication {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "CREATE PUBLICATION {}", self.name)?;
+        if let Some(target) = &self.target {
+            write!(f, " {target}")?;
+        }
+        if !self.with_options.is_empty() {
+            write!(f, " WITH ({})", display_comma_separated(&self.with_options))?;
+        }
+        Ok(())
+    }
+}
+
+impl From<CreatePublication> for crate::ast::Statement {
+    fn from(v: CreatePublication) -> Self {
+        crate::ast::Statement::CreatePublication(v)
+    }
+}
+
+/// A `CREATE SUBSCRIPTION` statement.
+///
+/// Note: this is a PostgreSQL-specific statement.
+/// <https://www.postgresql.org/docs/current/sql-createsubscription.html>
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub struct CreateSubscription {
+    /// The subscription name.
+    pub name: Ident,
+    /// The `CONNECTION 'conninfo'` string.
+    pub connection: Value,
+    /// The `PUBLICATION publication_name [, ...]` list.
+    pub publications: Vec<Ident>,
+    /// Optional `WITH (key = value, ...)` clause.
+    pub with_options: Vec<SqlOption>,
+}
+
+impl fmt::Display for CreateSubscription {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(
+            f,
+            "CREATE SUBSCRIPTION {name} CONNECTION {connection} PUBLICATION {publications}",
+            name = self.name,
+            connection = self.connection,
+            publications = display_comma_separated(&self.publications),
+        )?;
+        if !self.with_options.is_empty() {
+            write!(f, " WITH ({})", display_comma_separated(&self.with_options))?;
+        }
+        Ok(())
+    }
+}
+
+impl From<CreateSubscription> for crate::ast::Statement {
+    fn from(v: CreateSubscription) -> Self {
+        crate::ast::Statement::CreateSubscription(v)
+    }
+}

src/ast/mod.rs

@@ -72,8 +72,9 @@ pub use self::ddl::{
     CreateAggregateOption, CreateCollation, CreateCollationDefinition, CreateConnector,
     CreateDomain, CreateExtension, CreateForeignDataWrapper, CreateForeignTable, CreateFunction,
     CreateIndex, CreateOperator, CreateOperatorClass, CreateOperatorFamily, CreatePolicy,
-    CreatePolicyCommand, CreatePolicyType, CreateTable, CreateTextSearchConfiguration,
-    CreateTextSearchDictionary, CreateTextSearchParser, CreateTextSearchTemplate, CreateTrigger,
+    CreatePolicyCommand, CreatePolicyType, CreatePublication, CreateSubscription, CreateTable,
+    CreateTextSearchConfiguration, CreateTextSearchDictionary, CreateTextSearchParser,
+    CreateTextSearchTemplate, CreateTrigger, PublicationTarget,
     CreateView, Deduplicate, DeferrableInitial, DistStyle, DropBehavior, DropExtension,
     DropFunction, DropOperator, DropOperatorClass, DropOperatorFamily, DropOperatorSignature,
     DropPolicy, DropTrigger, FdwRoutineClause, ForValues, FunctionReturnType, GeneratedAs,
@@ -4017,6 +4018,18 @@ pub enum Statement {
     /// <https://www.postgresql.org/docs/current/sql-createtstemplate.html>
     CreateTextSearchTemplate(CreateTextSearchTemplate),
     /// ```sql
+    /// CREATE PUBLICATION name [ FOR ALL TABLES | FOR TABLE table [, ...] | FOR TABLES IN SCHEMA schema [, ...] ] [ WITH ( option = value [, ...] ) ]
+    /// ```
+    /// Note: this is a PostgreSQL-specific statement.
+    /// <https://www.postgresql.org/docs/current/sql-createpublication.html>
+    CreatePublication(CreatePublication),
+    /// ```sql
+    /// CREATE SUBSCRIPTION name CONNECTION 'conninfo' PUBLICATION publication_name [, ...] [ WITH ( option = value [, ...] ) ]
+    /// ```
+    /// Note: this is a PostgreSQL-specific statement.
+    /// <https://www.postgresql.org/docs/current/sql-createsubscription.html>
+    CreateSubscription(CreateSubscription),
+    /// ```sql
     /// DROP EXTENSION [ IF EXISTS ] name [, ...] [ CASCADE | RESTRICT ]
     /// ```
     /// Note: this is a PostgreSQL-specific statement.
@@ -5503,6 +5516,8 @@ impl fmt::Display for Statement {
             Statement::CreateTextSearchDictionary(v) => write!(f, "{v}"),
             Statement::CreateTextSearchParser(v) => write!(f, "{v}"),
             Statement::CreateTextSearchTemplate(v) => write!(f, "{v}"),
+            Statement::CreatePublication(v) => write!(f, "{v}"),
+            Statement::CreateSubscription(v) => write!(f, "{v}"),
             Statement::DropExtension(drop_extension) => write!(f, "{drop_extension}"),
             Statement::DropOperator(drop_operator) => write!(f, "{drop_operator}"),
             Statement::DropOperatorFamily(drop_operator_family) => {

src/ast/spans.rs

@@ -388,6 +388,8 @@ impl Spanned for Statement {
             Statement::CreateTextSearchDictionary(_) => Span::empty(),
             Statement::CreateTextSearchParser(_) => Span::empty(),
             Statement::CreateTextSearchTemplate(_) => Span::empty(),
+            Statement::CreatePublication(_) => Span::empty(),
+            Statement::CreateSubscription(_) => Span::empty(),
             Statement::DropExtension(drop_extension) => drop_extension.span(),
             Statement::DropOperator(drop_operator) => drop_operator.span(),
             Statement::DropOperatorFamily(drop_operator_family) => drop_operator_family.span(),

src/keywords.rs

@@ -822,6 +822,7 @@ define_keywords!(
     PROGRAM,
     PROJECTION,
     PUBLIC,
+    PUBLICATION,
     PURCHASE,
     PURGE,
     QUALIFY,
@@ -1010,6 +1011,7 @@ define_keywords!(
     STRUCT,
     SUBMULTISET,
     SUBSCRIPT,
+    SUBSCRIPTION,
     SUBSTR,
     SUBSTRING,
     SUBSTRING_REGEX,

src/parser/mod.rs

@@ -5228,6 +5228,10 @@ impl<'a> Parser<'a> {
             }
         } else if self.parse_keywords(&[Keyword::TEXT, Keyword::SEARCH]) {
             self.parse_create_text_search()
+        } else if self.parse_keyword(Keyword::PUBLICATION) {
+            self.parse_create_publication().map(Into::into)
+        } else if self.parse_keyword(Keyword::SUBSCRIPTION) {
+            self.parse_create_subscription().map(Into::into)
         } else {
             self.expected_ref("an object type after CREATE", self.peek_token_ref())
         }
@@ -20083,6 +20087,59 @@ impl<'a> Parser<'a> {
         })
     }
 
+    /// Parse a `CREATE PUBLICATION` statement.
+    ///
+    /// See <https://www.postgresql.org/docs/current/sql-createpublication.html>
+    pub fn parse_create_publication(&mut self) -> Result<CreatePublication, ParserError> {
+        let name = self.parse_identifier()?;
+
+        let target = if self.parse_keyword(Keyword::FOR) {
+            if self.parse_keywords(&[Keyword::ALL, Keyword::TABLES]) {
+                Some(PublicationTarget::AllTables)
+            } else if self.parse_keyword(Keyword::TABLE) {
+                let tables = self.parse_comma_separated(|p| p.parse_object_name(false))?;
+                Some(PublicationTarget::Tables(tables))
+            } else if self.parse_keywords(&[Keyword::TABLES, Keyword::IN, Keyword::SCHEMA]) {
+                let schemas = self.parse_comma_separated(|p| p.parse_identifier())?;
+                Some(PublicationTarget::TablesInSchema(schemas))
+            } else {
+                return self.expected_ref(
+                    "ALL TABLES, TABLE, or TABLES IN SCHEMA after FOR",
+                    self.peek_token_ref(),
+                );
+            }
+        } else {
+            None
+        };
+
+        let with_options = self.parse_options(Keyword::WITH)?;
+
+        Ok(CreatePublication {
+            name,
+            target,
+            with_options,
+        })
+    }
+
+    /// Parse a `CREATE SUBSCRIPTION` statement.
+    ///
+    /// See <https://www.postgresql.org/docs/current/sql-createsubscription.html>
+    pub fn parse_create_subscription(&mut self) -> Result<CreateSubscription, ParserError> {
+        let name = self.parse_identifier()?;
+        self.expect_keyword_is(Keyword::CONNECTION)?;
+        let connection = self.parse_value()?.value;
+        self.expect_keyword_is(Keyword::PUBLICATION)?;
+        let publications = self.parse_comma_separated(|p| p.parse_identifier())?;
+        let with_options = self.parse_options(Keyword::WITH)?;
+
+        Ok(CreateSubscription {
+            name,
+            connection,
+            publications,
+            with_options,
+        })
+    }
+
     /// The index of the first unprocessed token.
     pub fn index(&self) -> usize {
         self.index

tests/sqlparser_postgres.rs

@@ -9615,3 +9615,109 @@ fn parse_create_aggregate_with_moving_aggregate_options() {
         _ => panic!("Expected CreateAggregate, got: {stmt:?}"),
     }
 }
+
+#[test]
+fn parse_create_publication_basic() {
+    let sql = "CREATE PUBLICATION mypub FOR TABLE public.t";
+    let Statement::CreatePublication(stmt) = pg().verified_stmt(sql) else {
+        unreachable!()
+    };
+    assert_eq!(stmt.name.value, "mypub");
+    assert!(stmt.with_options.is_empty());
+    match stmt.target.unwrap() {
+        PublicationTarget::Tables(tables) => {
+            assert_eq!(tables.len(), 1);
+            assert_eq!(tables[0].to_string(), "public.t");
+        }
+        other => panic!("unexpected target: {other:?}"),
+    }
+}
+
+#[test]
+fn parse_create_publication_for_all_tables() {
+    let sql = "CREATE PUBLICATION mypub FOR ALL TABLES";
+    let Statement::CreatePublication(stmt) = pg().verified_stmt(sql) else {
+        unreachable!()
+    };
+    assert_eq!(stmt.name.value, "mypub");
+    assert!(matches!(stmt.target, Some(PublicationTarget::AllTables)));
+    assert!(stmt.with_options.is_empty());
+}
+
+#[test]
+fn parse_create_publication_for_tables_in_schema() {
+    let sql = "CREATE PUBLICATION mypub FOR TABLES IN SCHEMA myschema";
+    let Statement::CreatePublication(stmt) = pg().verified_stmt(sql) else {
+        unreachable!()
+    };
+    assert_eq!(stmt.name.value, "mypub");
+    match stmt.target.unwrap() {
+        PublicationTarget::TablesInSchema(schemas) => {
+            assert_eq!(schemas.len(), 1);
+            assert_eq!(schemas[0].value, "myschema");
+        }
+        other => panic!("unexpected target: {other:?}"),
+    }
+}
+
+#[test]
+fn parse_create_publication_with_options() {
+    let sql = "CREATE PUBLICATION mypub FOR ALL TABLES WITH (publish = 'insert, update')";
+    let Statement::CreatePublication(stmt) = pg().verified_stmt(sql) else {
+        unreachable!()
+    };
+    assert_eq!(stmt.name.value, "mypub");
+    assert!(matches!(stmt.target, Some(PublicationTarget::AllTables)));
+    assert_eq!(stmt.with_options.len(), 1);
+    match &stmt.with_options[0] {
+        SqlOption::KeyValue { key, value } => {
+            assert_eq!(key.value, "publish");
+            assert_eq!(value.to_string(), "'insert, update'");
+        }
+        other => panic!("unexpected option: {other:?}"),
+    }
+}
+
+#[test]
+fn parse_create_subscription_basic() {
+    let sql = "CREATE SUBSCRIPTION mysub CONNECTION 'host=localhost' PUBLICATION mypub";
+    let Statement::CreateSubscription(stmt) = pg().verified_stmt(sql) else {
+        unreachable!()
+    };
+    assert_eq!(stmt.name.value, "mysub");
+    assert_eq!(stmt.connection.to_string(), "'host=localhost'");
+    assert_eq!(stmt.publications.len(), 1);
+    assert_eq!(stmt.publications[0].value, "mypub");
+    assert!(stmt.with_options.is_empty());
+}
+
+#[test]
+fn parse_create_subscription_with_options() {
+    let sql = "CREATE SUBSCRIPTION mysub CONNECTION 'host=localhost dbname=mydb' PUBLICATION mypub, otherpub WITH (copy_data = true, slot_name = 'myslot')";
+    let Statement::CreateSubscription(stmt) = pg().verified_stmt(sql) else {
+        unreachable!()
+    };
+    assert_eq!(stmt.name.value, "mysub");
+    assert_eq!(
+        stmt.connection.to_string(),
+        "'host=localhost dbname=mydb'"
+    );
+    assert_eq!(stmt.publications.len(), 2);
+    assert_eq!(stmt.publications[0].value, "mypub");
+    assert_eq!(stmt.publications[1].value, "otherpub");
+    assert_eq!(stmt.with_options.len(), 2);
+    match &stmt.with_options[0] {
+        SqlOption::KeyValue { key, value } => {
+            assert_eq!(key.value, "copy_data");
+            assert_eq!(value.to_string(), "true");
+        }
+        other => panic!("unexpected option: {other:?}"),
+    }
+    match &stmt.with_options[1] {
+        SqlOption::KeyValue { key, value } => {
+            assert_eq!(key.value, "slot_name");
+            assert_eq!(value.to_string(), "'myslot'");
+        }
+        other => panic!("unexpected option: {other:?}"),
+    }
+}