Add support of parsing ON CLUSTER in ALTER TABLE for ClickHouse

ClickHouse dialect supports `ON CLUSTER` clause for ALTER TABLE:

```sql
ALTER TABLE table_name ON CLUSTER cluster_name ADD COLUMN c UInt32
```

Please refer to: https://clickhouse.com/docs/en/sql-reference/statements/alter

Author: git-hulk

src/ast/dml.rs

@@ -206,11 +206,7 @@ impl Display for CreateTable {
             name = self.name,
         )?;
         if let Some(on_cluster) = &self.on_cluster {
-            write!(
-                f,
-                " ON CLUSTER {}",
-                on_cluster.replace('{', "'{").replace('}', "}'")
-            )?;
+            write!(f, " ON CLUSTER {}", on_cluster)?;
         }
         if !self.columns.is_empty() || !self.constraints.is_empty() {
             write!(f, " ({}", display_comma_separated(&self.columns))?;

src/ast/mod.rs

@@ -2162,6 +2162,10 @@ pub enum Statement {
         only: bool,
         operations: Vec<AlterTableOperation>,
         location: Option<HiveSetLocation>,
+        /// ClickHouse dialect supports `ON CLUSTER` clause for ALTER TABLE
+        /// For example: `ALTER TABLE table_name ON CLUSTER cluster_name ADD COLUMN c UInt32`
+        /// [ClickHouse](https://clickhouse.com/docs/en/sql-reference/statements/alter)
+        on_cluster: Option<String>,
     },
     /// ```sql
     /// ALTER INDEX
@@ -3623,6 +3627,7 @@ impl fmt::Display for Statement {
                 only,
                 operations,
                 location,
+                on_cluster,
             } => {
                 write!(f, "ALTER TABLE ")?;
                 if *if_exists {
@@ -3631,9 +3636,13 @@ impl fmt::Display for Statement {
                 if *only {
                     write!(f, "ONLY ")?;
                 }
+                write!(f, "{name} ", name = name)?;
+                if let Some(cluster) = on_cluster {
+                    write!(f, "ON CLUSTER {cluster} ")?;
+                }
                 write!(
                     f,
-                    "{name} {operations}",
+                    "{operations}",
                     operations = display_comma_separated(operations)
                 )?;
                 if let Some(loc) = location {

src/parser/mod.rs

@@ -5372,6 +5372,19 @@ impl<'a> Parser<'a> {
         }
     }
 
+    fn parse_optional_on_cluster(&mut self) -> Result<Option<String>, ParserError> {
+        if self.parse_keywords(&[Keyword::ON, Keyword::CLUSTER]) {
+            let next_token = self.next_token();
+            match next_token.token {
+                Token::SingleQuotedString(s) => Ok(Some(format!("'{}'", s))),
+                Token::Word(s) => Ok(Some(s.to_string())),
+                _ => self.expected("identifier or cluster literal", next_token)?,
+            }
+        } else {
+            Ok(None)
+        }
+    }
+
     pub fn parse_create_table(
         &mut self,
         or_replace: bool,
@@ -5384,16 +5397,7 @@ impl<'a> Parser<'a> {
         let table_name = self.parse_object_name(allow_unquoted_hyphen)?;
 
         // Clickhouse has `ON CLUSTER 'cluster'` syntax for DDLs
-        let on_cluster = if self.parse_keywords(&[Keyword::ON, Keyword::CLUSTER]) {
-            let next_token = self.next_token();
-            match next_token.token {
-                Token::SingleQuotedString(s) => Some(s),
-                Token::Word(s) => Some(s.to_string()),
-                _ => self.expected("identifier or cluster literal", next_token)?,
-            }
-        } else {
-            None
-        };
+        let on_cluster = self.parse_optional_on_cluster()?;
 
         let like = if self.parse_keyword(Keyword::LIKE) || self.parse_keyword(Keyword::ILIKE) {
             self.parse_object_name(allow_unquoted_hyphen).ok()
@@ -6576,6 +6580,7 @@ impl<'a> Parser<'a> {
                 let if_exists = self.parse_keywords(&[Keyword::IF, Keyword::EXISTS]);
                 let only = self.parse_keyword(Keyword::ONLY); // [ ONLY ]
                 let table_name = self.parse_object_name(false)?;
+                let on_cluster = self.parse_optional_on_cluster()?;
                 let operations = self.parse_comma_separated(Parser::parse_alter_table_operation)?;
 
                 let mut location = None;
@@ -6597,6 +6602,7 @@ impl<'a> Parser<'a> {
                     only,
                     operations,
                     location,
+                    on_cluster,
                 })
             }
             Keyword::INDEX => {

src/test_utils.rs

@@ -274,6 +274,7 @@ pub fn alter_table_op_with_name(stmt: Statement, expected_name: &str) -> AlterTa
             if_exists,
             only: is_only,
             operations,
+            on_cluster: _,
             location: _,
         } => {
             assert_eq!(name.to_string(), expected_name);

tests/sqlparser_clickhouse.rs

@@ -25,6 +25,7 @@ use sqlparser::ast::Value::Number;
 use sqlparser::ast::*;
 use sqlparser::dialect::ClickHouseDialect;
 use sqlparser::dialect::GenericDialect;
+use sqlparser::parser::ParserError::ParserError;
 
 #[test]
 fn parse_map_access_expr() {
@@ -1091,6 +1092,27 @@ fn parse_create_table_on_commit_and_as_query() {
     }
 }
 
+#[test]
+fn test_alter_table_with_on_cluster() {
+    let sql = "ALTER TABLE t ON CLUSTER 'cluster' ADD CONSTRAINT bar PRIMARY KEY (baz)";
+    match clickhouse_and_generic().verified_stmt(sql) {
+        Statement::AlterTable {
+            name, on_cluster, ..
+        } => {
+            assert_eq!(name.to_string(), "t");
+            assert_eq!(on_cluster, Some("'cluster'".to_string()));
+        }
+        _ => unreachable!(),
+    }
+
+    let res = clickhouse_and_generic()
+        .parse_sql_statements("ALTER TABLE t ON CLUSTER 123 ADD CONSTRAINT bar PRIMARY KEY (baz)");
+    assert_eq!(
+        res.unwrap_err(),
+        ParserError("Expected: identifier or cluster literal, found: 123".to_string())
+    )
+}
+
 fn clickhouse() -> TestedDialects {
     TestedDialects {
         dialects: vec![Box::new(ClickHouseDialect {})],

tests/sqlparser_common.rs

@@ -3506,7 +3506,7 @@ fn parse_create_table_on_cluster() {
     let sql = "CREATE TABLE t ON CLUSTER '{cluster}' (a INT, b INT)";
     match generic.verified_stmt(sql) {
         Statement::CreateTable(CreateTable { on_cluster, .. }) => {
-            assert_eq!(on_cluster.unwrap(), "{cluster}".to_string());
+            assert_eq!(on_cluster.unwrap(), "'{cluster}'".to_string());
         }
         _ => unreachable!(),
     }

tests/sqlparser_mysql.rs

@@ -1975,6 +1975,7 @@ fn parse_alter_table_add_column() {
             only,
             operations,
             location: _,
+            on_cluster: _,
         } => {
             assert_eq!(name.to_string(), "tab");
             assert!(!if_exists);
@@ -2004,6 +2005,7 @@ fn parse_alter_table_add_column() {
             only,
             operations,
             location: _,
+            on_cluster: _,
         } => {
             assert_eq!(name.to_string(), "tab");
             assert!(!if_exists);
@@ -2041,6 +2043,7 @@ fn parse_alter_table_add_columns() {
             only,
             operations,
             location: _,
+            on_cluster: _,
         } => {
             assert_eq!(name.to_string(), "tab");
             assert!(!if_exists);

tests/sqlparser_postgres.rs

@@ -677,6 +677,7 @@ fn parse_alter_table_add_columns() {
             only,
             operations,
             location: _,
+            on_cluster: _,
         } => {
             assert_eq!(name.to_string(), "tab");
             assert!(if_exists);
@@ -759,6 +760,7 @@ fn parse_alter_table_owner_to() {
                 only: _,
                 operations,
                 location: _,
+                on_cluster: _,
             } => {
                 assert_eq!(name.to_string(), "tab");
                 assert_eq!(