Databricks: support INSERT BY NAME

src/ast/dml.rs

@@ -62,6 +62,20 @@ pub struct Insert {
     pub table_alias: Option<TableAliasWithoutColumns>,
     /// COLUMNS
     pub columns: Vec<ObjectName>,
+    /// `BY NAME` clause used by Databricks SQL.
+    ///
+    /// When present, columns from the source query are matched to columns in
+    /// the target table by name instead of by position. The syntax is:
+    ///
+    /// ```sql
+    /// INSERT INTO [TABLE] table_name
+    ///     [PARTITION (...)]
+    ///     [(column_name [, ...]) | BY NAME]
+    ///     query
+    /// ```
+    ///
+    /// See <https://docs.databricks.com/gcp/en/sql/language-manual/sql-ref-syntax-dml-insert-into>.
+    pub by_name: bool,
     /// Overwrite (Hive)
     pub overwrite: bool,
     /// A SQL query that specifies what to insert
@@ -201,6 +215,11 @@ impl Display for Insert {
             }
         }
 
+        if self.by_name {
+            write!(f, "BY NAME")?;
+            SpaceOrNewline.fmt(f)?;
+        }
+
         if !self.after_columns.is_empty() {
             write!(f, "({})", display_comma_separated(&self.after_columns))?;
             SpaceOrNewline.fmt(f)?;

src/ast/spans.rs

@@ -1354,6 +1354,7 @@ impl Spanned for Insert {
             table,
             table_alias,
             columns,
+            by_name: _,   // bool
             overwrite: _, // bool
             source,
             partitioned,

src/dialect/snowflake.rs

@@ -1871,6 +1871,7 @@ fn parse_multi_table_insert(
         table: TableObject::TableName(ObjectName(vec![])), // Not used for multi-table insert
         table_alias: None,
         columns: vec![],
+        by_name: false,
         overwrite,
         source: Some(source),
         assignments: vec![],

src/parser/mod.rs

@@ -18437,7 +18437,9 @@ impl<'a> Parser<'a> {
             let table = self.parse_keyword(Keyword::TABLE);
             let table_object = self.parse_table_object()?;
 
+            // `BY NAME` is an INSERT clause, not a table alias.
             let table_alias = if self.dialect.supports_insert_table_alias()
+                && !self.peek_keywords(&[Keyword::BY, Keyword::NAME])
                 && !self.peek_sub_query()
                 && self
                     .peek_one_of_keywords(&[Keyword::DEFAULT, Keyword::VALUES])
@@ -18461,6 +18463,7 @@ impl<'a> Parser<'a> {
 
             let is_mysql = dialect_of!(self is MySqlDialect);
 
+            let mut by_name = false;
             let (columns, partitioned, after_columns, output, source, assignments) = if self
                 .parse_keywords(&[Keyword::DEFAULT, Keyword::VALUES])
             {
@@ -18471,6 +18474,7 @@ impl<'a> Parser<'a> {
                         self.parse_parenthesized_qualified_column_list(Optional, is_mysql)?;
 
                     let partitioned = self.parse_insert_partition()?;
+                    by_name = self.parse_keywords(&[Keyword::BY, Keyword::NAME]);
                     // Hive allows you to specify columns after partitions as well if you want.
                     let after_columns = if dialect_of!(self is HiveDialect) {
                         self.parse_parenthesized_column_list(Optional, false)?
@@ -18595,6 +18599,7 @@ impl<'a> Parser<'a> {
                 ignore,
                 into,
                 overwrite,
+                by_name,
                 partitioned,
                 columns,
                 after_columns,

tests/sqlparser_common.rs

@@ -19679,3 +19679,48 @@ fn parse_function_arg_call_chain_no_exponential_blowup() {
     rx.recv_timeout(Duration::from_secs(5))
         .expect("parser should reject this quickly, not loop exponentially");
 }
+
+#[test]
+fn parse_insert_by_name() {
+    verified_stmt("INSERT INTO target BY NAME SELECT 1 AS a");
+
+    match verified_stmt("INSERT INTO target (a) BY NAME SELECT 1 AS a") {
+        Statement::Insert(Insert {
+            by_name, columns, ..
+        }) => {
+            assert!(by_name);
+            assert_eq!(columns.len(), 1);
+        }
+        _ => unreachable!(),
+    }
+
+    let dialects = all_dialects_where(|d| !d.supports_insert_table_alias());
+    match dialects.verified_stmt("INSERT INTO TABLE target PARTITION (p = 1) BY NAME SELECT 1 AS a")
+    {
+        Statement::Insert(Insert {
+            by_name,
+            has_table_keyword,
+            partitioned,
+            ..
+        }) => {
+            assert!(by_name);
+            assert!(has_table_keyword);
+            assert_eq!(partitioned.unwrap().len(), 1);
+        }
+        _ => unreachable!(),
+    }
+
+    // `BY NAME` does not shadow a table alias in dialects supporting one.
+    let dialects = all_dialects_where(|d| d.supports_insert_table_alias());
+    match dialects.verified_stmt("INSERT INTO target AS t BY NAME SELECT 1 AS a") {
+        Statement::Insert(Insert {
+            by_name,
+            table_alias,
+            ..
+        }) => {
+            assert!(by_name);
+            assert_eq!(table_alias.unwrap().alias.value, "t");
+        }
+        _ => unreachable!(),
+    }
+}

tests/sqlparser_databricks.rs

@@ -737,3 +737,11 @@ fn parse_cte_without_as() {
         .parse_sql_statements("WITH cte (SELECT 1) SELECT * FROM cte")
         .is_err());
 }
+
+#[test]
+fn test_databricks_insert_by_name() {
+    databricks_and_generic().verified_stmt("INSERT INTO target BY NAME SELECT 1 AS a");
+    databricks_and_generic().verified_stmt(
+        "INSERT INTO TABLE lakehouse.dwd.dwd_event_quality_sla_metric_di BY NAME WITH day AS (SELECT 1 AS event_data_id) SELECT event_data_id FROM day",
+    );
+}

tests/sqlparser_postgres.rs

@@ -6093,6 +6093,7 @@ fn test_simple_postgres_insert_with_alias() {
                     span: Span::empty(),
                 })
             ],
+            by_name: false,
             overwrite: false,
             source: Some(Box::new(Query {
                 with: None,
@@ -6173,6 +6174,7 @@ fn test_simple_postgres_insert_with_alias() {
                     span: Span::empty(),
                 })
             ],
+            by_name: false,
             overwrite: false,
             source: Some(Box::new(Query {
                 with: None,
@@ -6255,6 +6257,7 @@ fn test_simple_insert_with_quoted_alias() {
                     span: Span::empty(),
                 })
             ],
+            by_name: false,
             overwrite: false,
             source: Some(Box::new(Query {
                 with: None,
@@ -9663,3 +9666,22 @@ fn parse_right_deep_join_chain() {
     // NATURAL JOIN followed by a constrained join must stay left-associative.
     pg().verified_stmt("SELECT * FROM t0 NATURAL JOIN t1 INNER JOIN t2 ON true");
 }
+
+#[test]
+fn parse_insert_by_name_keywords_as_table_and_alias() {
+    // Without a table name, `BY NAME` is not an INSERT BY NAME clause. PostgreSQL
+    // treats `BY` as the table name and `NAME` as its implicit table alias.
+    match pg().verified_stmt("INSERT INTO BY NAME SELECT 1 AS a") {
+        Statement::Insert(Insert {
+            table: TableObject::TableName(table),
+            table_alias: Some(table_alias),
+            by_name,
+            ..
+        }) => {
+            assert_eq!(table.to_string(), "BY");
+            assert_eq!(table_alias.alias.value, "NAME");
+            assert!(!by_name);
+        }
+        statement => panic!("Expected INSERT statement, got: {statement:?}"),
+    }
+}