Add support for CREATE/ALTER/DROP CONNECTOR syntax (#1701)

src/ast/ddl.rs

@@ -30,10 +30,10 @@ use sqlparser_derive::{Visit, VisitMut};
 
 use crate::ast::value::escape_single_quote_string;
 use crate::ast::{
-    display_comma_separated, display_separated, CreateFunctionBody, CreateFunctionUsing, DataType,
+    display_comma_separated, display_separated, CommentDef, CreateFunctionBody,
-    Expr, FunctionBehavior, FunctionCalledOnNull, FunctionDeterminismSpecifier, FunctionParallel,
+    CreateFunctionUsing, DataType, Expr, FunctionBehavior, FunctionCalledOnNull,
-    Ident, MySQLColumnPosition, ObjectName, OperateFunctionArg, OrderByExpr, ProjectionSelect,
+    FunctionDeterminismSpecifier, FunctionParallel, Ident, MySQLColumnPosition, ObjectName,
-    SequenceOptions, SqlOption, Tag, Value,
+    OperateFunctionArg, OrderByExpr, ProjectionSelect, SequenceOptions, SqlOption, Tag, Value,
 };
 use crate::keywords::Keyword;
 use crate::tokenizer::Token;
@@ -338,6 +338,23 @@ impl fmt::Display for Owner {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub enum AlterConnectorOwner {
+    User(Ident),
+    Role(Ident),
+}
+
+impl fmt::Display for AlterConnectorOwner {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            AlterConnectorOwner::User(ident) => write!(f, "USER {ident}"),
+            AlterConnectorOwner::Role(ident) => write!(f, "ROLE {ident}"),
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 #[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
@@ -2055,3 +2072,61 @@ impl fmt::Display for CreateFunction {
         Ok(())
     }
 }
+
+/// ```sql
+/// CREATE CONNECTOR [IF NOT EXISTS] connector_name
+/// [TYPE datasource_type]
+/// [URL datasource_url]
+/// [COMMENT connector_comment]
+/// [WITH DCPROPERTIES(property_name=property_value, ...)]
+/// ```
+///
+/// [Hive](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=27362034#LanguageManualDDL-CreateDataConnectorCreateConnector)
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub struct CreateConnector {
+    pub name: Ident,
+    pub if_not_exists: bool,
+    pub connector_type: Option<String>,
+    pub url: Option<String>,
+    pub comment: Option<CommentDef>,
+    pub with_dcproperties: Option<Vec<SqlOption>>,
+}
+
+impl fmt::Display for CreateConnector {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(
+            f,
+            "CREATE CONNECTOR {if_not_exists}{name}",
+            if_not_exists = if self.if_not_exists {
+                "IF NOT EXISTS "
+            } else {
+                ""
+            },
+            name = self.name,
+        )?;
+
+        if let Some(connector_type) = &self.connector_type {
+            write!(f, " TYPE '{connector_type}'")?;
+        }
+
+        if let Some(url) = &self.url {
+            write!(f, " URL '{url}'")?;
+        }
+
+        if let Some(comment) = &self.comment {
+            write!(f, " COMMENT = '{comment}'")?;
+        }
+
+        if let Some(with_dcproperties) = &self.with_dcproperties {
+            write!(
+                f,
+                " WITH DCPROPERTIES({})",
+                display_comma_separated(with_dcproperties)
+            )?;
+        }
+
+        Ok(())
+    }
+}

src/ast/mod.rs

@@ -47,14 +47,14 @@ pub use self::dcl::{
     AlterRoleOperation, ResetConfig, RoleOption, SecondaryRoles, SetConfigValue, Use,
 };
 pub use self::ddl::{
-    AlterColumnOperation, AlterIndexOperation, AlterPolicyOperation, AlterTableOperation,
+    AlterColumnOperation, AlterConnectorOwner, AlterIndexOperation, AlterPolicyOperation,
-    ClusteredBy, ColumnDef, ColumnOption, ColumnOptionDef, ColumnPolicy, ColumnPolicyProperty,
+    AlterTableOperation, ClusteredBy, ColumnDef, ColumnOption, ColumnOptionDef, ColumnPolicy,
-    ConstraintCharacteristics, CreateFunction, Deduplicate, DeferrableInitial, DropBehavior,
+    ColumnPolicyProperty, ConstraintCharacteristics, CreateConnector, CreateFunction, Deduplicate,
-    GeneratedAs, GeneratedExpressionMode, IdentityParameters, IdentityProperty,
+    DeferrableInitial, DropBehavior, GeneratedAs, GeneratedExpressionMode, IdentityParameters,
-    IdentityPropertyFormatKind, IdentityPropertyKind, IdentityPropertyOrder, IndexOption,
+    IdentityProperty, IdentityPropertyFormatKind, IdentityPropertyKind, IdentityPropertyOrder,
-    IndexType, KeyOrIndexDisplay, NullsDistinctOption, Owner, Partition, ProcedureParam,
+    IndexOption, IndexType, KeyOrIndexDisplay, NullsDistinctOption, Owner, Partition,
-    ReferentialAction, TableConstraint, TagsColumnOption, UserDefinedTypeCompositeAttributeDef,
+    ProcedureParam, ReferentialAction, TableConstraint, TagsColumnOption,
-    UserDefinedTypeRepresentation, ViewColumnDef,
+    UserDefinedTypeCompositeAttributeDef, UserDefinedTypeRepresentation, ViewColumnDef,
 };
 pub use self::dml::{CreateIndex, CreateTable, Delete, Insert};
 pub use self::operator::{BinaryOperator, UnaryOperator};
@@ -2646,6 +2646,11 @@ pub enum Statement {
         with_check: Option<Expr>,
     },
     /// ```sql
+    /// CREATE CONNECTOR
+    /// ```
+    /// See [Hive](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=27362034#LanguageManualDDL-CreateDataConnectorCreateConnector)
+    CreateConnector(CreateConnector),
+    /// ```sql
     /// ALTER TABLE
     /// ```
     AlterTable {
@@ -2697,6 +2702,20 @@ pub enum Statement {
         operation: AlterPolicyOperation,
     },
     /// ```sql
+    /// ALTER CONNECTOR connector_name SET DCPROPERTIES(property_name=property_value, ...);
+    /// or
+    /// ALTER CONNECTOR connector_name SET URL new_url;
+    /// or
+    /// ALTER CONNECTOR connector_name SET OWNER [USER|ROLE] user_or_role;
+    /// ```
+    /// (Hive-specific)
+    AlterConnector {
+        name: Ident,
+        properties: Option<Vec<SqlOption>>,
+        url: Option<String>,
+        owner: Option<ddl::AlterConnectorOwner>,
+    },
+    /// ```sql
     /// ATTACH DATABASE 'path/to/file' AS alias
     /// ```
     /// (SQLite-specific)
@@ -2795,6 +2814,11 @@ pub enum Statement {
         drop_behavior: Option<DropBehavior>,
     },
     /// ```sql
+    /// DROP CONNECTOR
+    /// ```
+    /// See [Hive](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=27362034#LanguageManualDDL-DropConnector)
+    DropConnector { if_exists: bool, name: Ident },
+    /// ```sql
     /// DECLARE
     /// ```
     /// Declare Cursor Variables
@@ -4354,6 +4378,7 @@ impl fmt::Display for Statement {
 
                 Ok(())
             }
+            Statement::CreateConnector(create_connector) => create_connector.fmt(f),
             Statement::AlterTable {
                 name,
                 if_exists,
@@ -4411,6 +4436,28 @@ impl fmt::Display for Statement {
             } => {
                 write!(f, "ALTER POLICY {name} ON {table_name}{operation}")
             }
+            Statement::AlterConnector {
+                name,
+                properties,
+                url,
+                owner,
+            } => {
+                write!(f, "ALTER CONNECTOR {name}")?;
+                if let Some(properties) = properties {
+                    write!(
+                        f,
+                        " SET DCPROPERTIES({})",
+                        display_comma_separated(properties)
+                    )?;
+                }
+                if let Some(url) = url {
+                    write!(f, " SET URL '{url}'")?;
+                }
+                if let Some(owner) = owner {
+                    write!(f, " SET OWNER {owner}")?;
+                }
+                Ok(())
+            }
             Statement::Drop {
                 object_type,
                 if_exists,
@@ -4498,6 +4545,14 @@ impl fmt::Display for Statement {
                 }
                 Ok(())
             }
+            Statement::DropConnector { if_exists, name } => {
+                write!(
+                    f,
+                    "DROP CONNECTOR {if_exists}{name}",
+                    if_exists = if *if_exists { "IF EXISTS " } else { "" }
+                )?;
+                Ok(())
+            }
             Statement::Discard { object_type } => {
                 write!(f, "DISCARD {object_type}")?;
                 Ok(())

src/ast/spans.rs

@@ -398,6 +398,7 @@ impl Spanned for Statement {
             Statement::CreateIndex(create_index) => create_index.span(),
             Statement::CreateRole { .. } => Span::empty(),
             Statement::CreateSecret { .. } => Span::empty(),
+            Statement::CreateConnector { .. } => Span::empty(),
             Statement::AlterTable {
                 name,
                 if_exists: _,
@@ -487,7 +488,9 @@ impl Spanned for Statement {
             Statement::OptimizeTable { .. } => Span::empty(),
             Statement::CreatePolicy { .. } => Span::empty(),
             Statement::AlterPolicy { .. } => Span::empty(),
+            Statement::AlterConnector { .. } => Span::empty(),
             Statement::DropPolicy { .. } => Span::empty(),
+            Statement::DropConnector { .. } => Span::empty(),
             Statement::ShowDatabases { .. } => Span::empty(),
             Statement::ShowSchemas { .. } => Span::empty(),
             Statement::ShowViews { .. } => Span::empty(),

src/dialect/mod.rs

@@ -876,6 +876,7 @@ pub trait Dialect: Debug + Any {
     fn supports_string_escape_constant(&self) -> bool {
         false
     }
+
     /// Returns true if the dialect supports the table hints in the `FROM` clause.
     fn supports_table_hints(&self) -> bool {
         false

src/keywords.rs

@@ -201,6 +201,7 @@ define_keywords!(
     CONFLICT,
     CONNECT,
     CONNECTION,
+    CONNECTOR,
     CONSTRAINT,
     CONTAINS,
     CONTINUE,
@@ -246,6 +247,7 @@ define_keywords!(
     DAYOFWEEK,
     DAYOFYEAR,
     DAYS,
+    DCPROPERTIES,
     DEALLOCATE,
     DEC,
     DECADE,

src/parser/alter.rs

@@ -18,8 +18,8 @@ use alloc::vec;
 use super::{Parser, ParserError};
 use crate::{
     ast::{
-        AlterPolicyOperation, AlterRoleOperation, Expr, Password, ResetConfig, RoleOption,
+        AlterConnectorOwner, AlterPolicyOperation, AlterRoleOperation, Expr, Password, ResetConfig,
-        SetConfigValue, Statement,
+        RoleOption, SetConfigValue, Statement,
     },
     dialect::{MsSqlDialect, PostgreSqlDialect},
     keywords::Keyword,
@@ -99,6 +99,47 @@ impl Parser<'_> {
         }
     }
 
+    /// Parse an `ALTER CONNECTOR` statement
+    /// ```sql
+    /// ALTER CONNECTOR connector_name SET DCPROPERTIES(property_name=property_value, ...);
+    ///
+    /// ALTER CONNECTOR connector_name SET URL new_url;
+    ///
+    /// ALTER CONNECTOR connector_name SET OWNER [USER|ROLE] user_or_role;
+    /// ```
+    pub fn parse_alter_connector(&mut self) -> Result<Statement, ParserError> {
+        let name = self.parse_identifier()?;
+        self.expect_keyword_is(Keyword::SET)?;
+
+        let properties = match self.parse_options_with_keywords(&[Keyword::DCPROPERTIES])? {
+            properties if !properties.is_empty() => Some(properties),
+            _ => None,
+        };
+
+        let url = if self.parse_keyword(Keyword::URL) {
+            Some(self.parse_literal_string()?)
+        } else {
+            None
+        };
+
+        let owner = if self.parse_keywords(&[Keyword::OWNER, Keyword::USER]) {
+            let owner = self.parse_identifier()?;
+            Some(AlterConnectorOwner::User(owner))
+        } else if self.parse_keywords(&[Keyword::OWNER, Keyword::ROLE]) {
+            let owner = self.parse_identifier()?;
+            Some(AlterConnectorOwner::Role(owner))
+        } else {
+            None
+        };
+
+        Ok(Statement::AlterConnector {
+            name,
+            properties,
+            url,
+            owner,
+        })
+    }
+
     fn parse_mssql_alter_role(&mut self) -> Result<Statement, ParserError> {
         let role_name = self.parse_identifier()?;
 

src/parser/mod.rs

@@ -4268,6 +4268,8 @@ impl<'a> Parser<'a> {
             self.parse_create_type()
         } else if self.parse_keyword(Keyword::PROCEDURE) {
             self.parse_create_procedure(or_alter)
+        } else if self.parse_keyword(Keyword::CONNECTOR) {
+            self.parse_create_connector()
         } else {
             self.expected("an object type after CREATE", self.peek_token())
         }
@@ -5580,6 +5582,49 @@ impl<'a> Parser<'a> {
         })
     }
 
+    /// ```sql
+    /// CREATE CONNECTOR [IF NOT EXISTS] connector_name
+    /// [TYPE datasource_type]
+    /// [URL datasource_url]
+    /// [COMMENT connector_comment]
+    /// [WITH DCPROPERTIES(property_name=property_value, ...)]
+    /// ```
+    ///
+    /// [Hive Documentation](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=27362034#LanguageManualDDL-CreateDataConnectorCreateConnector)
+    pub fn parse_create_connector(&mut self) -> Result<Statement, ParserError> {
+        let if_not_exists = self.parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]);
+        let name = self.parse_identifier()?;
+
+        let connector_type = if self.parse_keyword(Keyword::TYPE) {
+            Some(self.parse_literal_string()?)
+        } else {
+            None
+        };
+
+        let url = if self.parse_keyword(Keyword::URL) {
+            Some(self.parse_literal_string()?)
+        } else {
+            None
+        };
+
+        let comment = self.parse_optional_inline_comment()?;
+
+        let with_dcproperties =
+            match self.parse_options_with_keywords(&[Keyword::WITH, Keyword::DCPROPERTIES])? {
+                properties if !properties.is_empty() => Some(properties),
+                _ => None,
+            };
+
+        Ok(Statement::CreateConnector(CreateConnector {
+            name,
+            if_not_exists,
+            connector_type,
+            url,
+            comment,
+            with_dcproperties,
+        }))
+    }
+
     pub fn parse_drop(&mut self) -> Result<Statement, ParserError> {
         // MySQL dialect supports `TEMPORARY`
         let temporary = dialect_of!(self is MySqlDialect | GenericDialect | DuckDbDialect)
@@ -5609,6 +5654,8 @@ impl<'a> Parser<'a> {
             return self.parse_drop_function();
         } else if self.parse_keyword(Keyword::POLICY) {
             return self.parse_drop_policy();
+        } else if self.parse_keyword(Keyword::CONNECTOR) {
+            return self.parse_drop_connector();
         } else if self.parse_keyword(Keyword::PROCEDURE) {
             return self.parse_drop_procedure();
         } else if self.parse_keyword(Keyword::SECRET) {
@@ -5619,7 +5666,7 @@ impl<'a> Parser<'a> {
             return self.parse_drop_extension();
         } else {
             return self.expected(
-                "DATABASE, EXTENSION, FUNCTION, INDEX, POLICY, PROCEDURE, ROLE, SCHEMA, SECRET, SEQUENCE, STAGE, TABLE, TRIGGER, TYPE, or VIEW after DROP",
+                "CONNECTOR, DATABASE, EXTENSION, FUNCTION, INDEX, POLICY, PROCEDURE, ROLE, SCHEMA, SECRET, SEQUENCE, STAGE, TABLE, TRIGGER, TYPE, or VIEW after DROP",
                 self.peek_token(),
             );
         };
@@ -5693,6 +5740,16 @@ impl<'a> Parser<'a> {
             drop_behavior,
         })
     }
+    /// ```sql  
+    /// DROP CONNECTOR [IF EXISTS] name
+    /// ```
+    ///
+    /// See [Hive](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=27362034#LanguageManualDDL-DropConnector)
+    fn parse_drop_connector(&mut self) -> Result<Statement, ParserError> {
+        let if_exists = self.parse_keywords(&[Keyword::IF, Keyword::EXISTS]);
+        let name = self.parse_identifier()?;
+        Ok(Statement::DropConnector { if_exists, name })
+    }
 
     /// ```sql
     /// DROP PROCEDURE [ IF EXISTS ] name [ ( [ [ argmode ] [ argname ] argtype [, ...] ] ) ] [, ...]
@@ -7989,6 +8046,7 @@ impl<'a> Parser<'a> {
             Keyword::INDEX,
             Keyword::ROLE,
             Keyword::POLICY,
+            Keyword::CONNECTOR,
         ])?;
         match object_type {
             Keyword::VIEW => self.parse_alter_view(),
@@ -8041,6 +8099,7 @@ impl<'a> Parser<'a> {
             }
             Keyword::ROLE => self.parse_alter_role(),
             Keyword::POLICY => self.parse_alter_policy(),
+            Keyword::CONNECTOR => self.parse_alter_connector(),
             // unreachable because expect_one_of_keywords used above
             _ => unreachable!(),
         }

tests/sqlparser_common.rs

@@ -12289,6 +12289,175 @@ fn test_alter_policy() {
     );
 }
 
+#[test]
+fn test_create_connector() {
+    let sql = "CREATE CONNECTOR my_connector \
+               TYPE 'jdbc' \
+               URL 'jdbc:mysql://localhost:3306/mydb' \
+               WITH DCPROPERTIES('user' = 'root', 'password' = 'password')";
+    let dialects = all_dialects();
+    match dialects.verified_stmt(sql) {
+        Statement::CreateConnector(CreateConnector {
+            name,
+            connector_type,
+            url,
+            with_dcproperties,
+            ..
+        }) => {
+            assert_eq!(name.to_string(), "my_connector");
+            assert_eq!(connector_type, Some("jdbc".to_string()));
+            assert_eq!(url, Some("jdbc:mysql://localhost:3306/mydb".to_string()));
+            assert_eq!(
+                with_dcproperties,
+                Some(vec![
+                    SqlOption::KeyValue {
+                        key: Ident::with_quote('\'', "user"),
+                        value: Expr::Value(Value::SingleQuotedString("root".to_string()))
+                    },
+                    SqlOption::KeyValue {
+                        key: Ident::with_quote('\'', "password"),
+                        value: Expr::Value(Value::SingleQuotedString("password".to_string()))
+                    }
+                ])
+            );
+        }
+        _ => unreachable!(),
+    }
+
+    // omit IF NOT EXISTS/TYPE/URL/COMMENT/WITH DCPROPERTIES clauses is allowed
+    dialects.verified_stmt("CREATE CONNECTOR my_connector");
+
+    // missing connector name
+    assert_eq!(
+        dialects
+            .parse_sql_statements("CREATE CONNECTOR")
+            .unwrap_err()
+            .to_string(),
+        "sql parser error: Expected: identifier, found: EOF"
+    );
+}
+
+#[test]
+fn test_drop_connector() {
+    let dialects = all_dialects();
+    match dialects.verified_stmt("DROP CONNECTOR IF EXISTS my_connector") {
+        Statement::DropConnector { if_exists, name } => {
+            assert_eq!(if_exists, true);
+            assert_eq!(name.to_string(), "my_connector");
+        }
+        _ => unreachable!(),
+    }
+
+    // omit IF EXISTS is allowed
+    dialects.verified_stmt("DROP CONNECTOR my_connector");
+
+    // missing connector name
+    assert_eq!(
+        dialects
+            .parse_sql_statements("DROP CONNECTOR")
+            .unwrap_err()
+            .to_string(),
+        "sql parser error: Expected: identifier, found: EOF"
+    );
+}
+
+#[test]
+fn test_alter_connector() {
+    let dialects = all_dialects();
+    match dialects.verified_stmt(
+        "ALTER CONNECTOR my_connector SET DCPROPERTIES('user' = 'root', 'password' = 'password')",
+    ) {
+        Statement::AlterConnector {
+            name,
+            properties,
+            url,
+            owner,
+        } => {
+            assert_eq!(name.to_string(), "my_connector");
+            assert_eq!(
+                properties,
+                Some(vec![
+                    SqlOption::KeyValue {
+                        key: Ident::with_quote('\'', "user"),
+                        value: Expr::Value(Value::SingleQuotedString("root".to_string()))
+                    },
+                    SqlOption::KeyValue {
+                        key: Ident::with_quote('\'', "password"),
+                        value: Expr::Value(Value::SingleQuotedString("password".to_string()))
+                    }
+                ])
+            );
+            assert_eq!(url, None);
+            assert_eq!(owner, None);
+        }
+        _ => unreachable!(),
+    }
+
+    match dialects
+        .verified_stmt("ALTER CONNECTOR my_connector SET URL 'jdbc:mysql://localhost:3306/mydb'")
+    {
+        Statement::AlterConnector {
+            name,
+            properties,
+            url,
+            owner,
+        } => {
+            assert_eq!(name.to_string(), "my_connector");
+            assert_eq!(properties, None);
+            assert_eq!(url, Some("jdbc:mysql://localhost:3306/mydb".to_string()));
+            assert_eq!(owner, None);
+        }
+        _ => unreachable!(),
+    }
+
+    match dialects.verified_stmt("ALTER CONNECTOR my_connector SET OWNER USER 'root'") {
+        Statement::AlterConnector {
+            name,
+            properties,
+            url,
+            owner,
+        } => {
+            assert_eq!(name.to_string(), "my_connector");
+            assert_eq!(properties, None);
+            assert_eq!(url, None);
+            assert_eq!(
+                owner,
+                Some(AlterConnectorOwner::User(Ident::with_quote('\'', "root")))
+            );
+        }
+        _ => unreachable!(),
+    }
+
+    match dialects.verified_stmt("ALTER CONNECTOR my_connector SET OWNER ROLE 'admin'") {
+        Statement::AlterConnector {
+            name,
+            properties,
+            url,
+            owner,
+        } => {
+            assert_eq!(name.to_string(), "my_connector");
+            assert_eq!(properties, None);
+            assert_eq!(url, None);
+            assert_eq!(
+                owner,
+                Some(AlterConnectorOwner::Role(Ident::with_quote('\'', "admin")))
+            );
+        }
+        _ => unreachable!(),
+    }
+
+    // Wrong option name
+    assert_eq!(
+        dialects
+            .parse_sql_statements(
+                "ALTER CONNECTOR my_connector SET WRONG 'jdbc:mysql://localhost:3306/mydb'"
+            )
+            .unwrap_err()
+            .to_string(),
+        "sql parser error: Expected: end of statement, found: WRONG"
+    );
+}
+
 #[test]
 fn test_select_where_with_like_or_ilike_any() {
     verified_stmt(r#"SELECT * FROM x WHERE a ILIKE ANY '%abc%'"#);