BigQuery: Parse WITH CONNECTION on CREATE EXTERNAL TABLE

src/ast/ddl.rs

@@ -3021,6 +3021,9 @@ pub struct CreateTable {
     /// Snowflake "EXTERNAL_VOLUME" clause for Iceberg tables
     /// <https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table>
     pub external_volume: Option<String>,
+    /// BigQuery `WITH CONNECTION` clause for external tables
+    /// <https://cloud.google.com/bigquery/docs/reference/standard-sql/data-definition-language#create_external_table_statement>
+    pub with_connection: Option<ObjectName>,
     /// Snowflake "BASE_LOCATION" clause for Iceberg tables
     /// <https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table>
     pub base_location: Option<String>,
@@ -3250,6 +3253,9 @@ impl fmt::Display for CreateTable {
         if let Some(cluster_by) = self.cluster_by.as_ref() {
             write!(f, " CLUSTER BY {cluster_by}")?;
         }
+        if let Some(with_connection) = &self.with_connection {
+            write!(f, " WITH CONNECTION {with_connection}")?;
+        }
         if let options @ CreateTableOptions::Options(_) = &self.table_options {
             write!(f, " {options}")?;
         }

src/ast/helpers/stmt_create_table.rs

@@ -157,6 +157,9 @@ pub struct CreateTableBuilder {
     pub base_location: Option<String>,
     /// Optional external volume identifier.
     pub external_volume: Option<String>,
+    /// BigQuery `WITH CONNECTION` clause for external tables.
+    /// <https://cloud.google.com/bigquery/docs/reference/standard-sql/data-definition-language#create_external_table_statement>
+    pub with_connection: Option<ObjectName>,
     /// Optional catalog name.
     pub catalog: Option<String>,
     /// Optional catalog synchronization option.
@@ -235,6 +238,7 @@ impl CreateTableBuilder {
             with_tags: None,
             base_location: None,
             external_volume: None,
+            with_connection: None,
             catalog: None,
             catalog_sync: None,
             storage_serialization_policy: None,
@@ -488,6 +492,11 @@ impl CreateTableBuilder {
         self.external_volume = external_volume;
         self
     }
+    /// Set the BigQuery `WITH CONNECTION` clause for external tables.
+    pub fn with_connection(mut self, with_connection: Option<ObjectName>) -> Self {
+        self.with_connection = with_connection;
+        self
+    }
     /// Set the catalog name for the table.
     pub fn catalog(mut self, catalog: Option<String>) -> Self {
         self.catalog = catalog;
@@ -605,6 +614,7 @@ impl CreateTableBuilder {
             with_tags: self.with_tags,
             base_location: self.base_location,
             external_volume: self.external_volume,
+            with_connection: self.with_connection,
             catalog: self.catalog,
             catalog_sync: self.catalog_sync,
             storage_serialization_policy: self.storage_serialization_policy,
@@ -686,6 +696,7 @@ impl From<CreateTable> for CreateTableBuilder {
             with_tags: table.with_tags,
             base_location: table.base_location,
             external_volume: table.external_volume,
+            with_connection: table.with_connection,
             catalog: table.catalog,
             catalog_sync: table.catalog_sync,
             storage_serialization_policy: table.storage_serialization_policy,

src/ast/spans.rs

@@ -589,6 +589,7 @@ impl Spanned for CreateTable {
             with_storage_lifecycle_policy: _,   // todo, Snowflake specific
             with_tags: _,                       // todo, Snowflake specific
             external_volume: _,                 // todo, Snowflake specific
+            with_connection: _,                 // todo, BigQuery external table connection
             base_location: _,                   // todo, Snowflake specific
             catalog: _,                         // todo, Snowflake specific
             catalog_sync: _,                    // todo, Snowflake specific

src/parser/mod.rs

@@ -6416,8 +6416,28 @@ impl<'a> Parser<'a> {
             None
         };
         let location = hive_formats.as_ref().and_then(|hf| hf.location.clone());
-        let table_properties = self.parse_options(Keyword::TBLPROPERTIES)?;
-        let table_options = if !table_properties.is_empty() {
+
+        // BigQuery external tables support `WITH CONNECTION <name>` and `OPTIONS(...)`
+        // clauses after the (optional) column list.
+        let with_connection = if self.parse_keywords(&[Keyword::WITH, Keyword::CONNECTION]) {
+            Some(self.parse_object_name(false)?)
+        } else {
+            None
+        };
+        // BigQuery uses OPTIONS(...); Hive uses TBLPROPERTIES(...). They are
+        // mutually exclusive in practice, and `parse_options` returns an empty
+        // vec when the keyword isn't present, so trying OPTIONS first and
+        // falling back to TBLPROPERTIES preserves the existing Hive path
+        // without accepting both in the same statement.
+        let options = self.parse_options(Keyword::OPTIONS)?;
+        let table_properties = if options.is_empty() {
+            self.parse_options(Keyword::TBLPROPERTIES)?
+        } else {
+            vec![]
+        };
+        let table_options = if !options.is_empty() {
+            CreateTableOptions::Options(options)
+        } else if !table_properties.is_empty() {
             CreateTableOptions::TableProperties(table_properties)
         } else if let Some(options) = self.maybe_parse_options(Keyword::OPTIONS)? {
             CreateTableOptions::Options(options)
@@ -6430,6 +6450,7 @@ impl<'a> Parser<'a> {
             .hive_distribution(hive_distribution)
             .hive_formats(hive_formats)
             .table_options(table_options)
+            .with_connection(with_connection)
             .or_replace(or_replace)
             .if_not_exists(if_not_exists)
             .external(true)

tests/sqlparser_bigquery.rs

@@ -2203,6 +2203,90 @@ fn parse_big_query_declare() {
     );
 }
 
+#[test]
+fn parse_bigquery_create_external_table_with_connection_and_options() {
+    let sql = concat!(
+        "CREATE OR REPLACE EXTERNAL TABLE `proj.ds.tbl` ",
+        "WITH CONNECTION `projects/proj/locations/us/connections/c` ",
+        r#"OPTIONS(format = "ICEBERG", uris = ["gs://b/m.json"])"#,
+    );
+    let stmts = bigquery().parse_sql_statements(sql).unwrap();
+    assert_eq!(stmts.len(), 1);
+    let Statement::CreateTable(ct) = &stmts[0] else {
+        panic!("expected CreateTable, got {:?}", stmts[0]);
+    };
+    assert!(ct.external);
+    assert!(ct.or_replace);
+    assert!(ct.columns.is_empty());
+    let with_connection = ct.with_connection.as_ref().expect("with_connection set");
+    assert_eq!(
+        with_connection.to_string(),
+        "`projects/proj/locations/us/connections/c`"
+    );
+    let CreateTableOptions::Options(opts) = &ct.table_options else {
+        panic!("expected OPTIONS, got {:?}", ct.table_options);
+    };
+    assert_eq!(opts.len(), 2);
+}
+
+#[test]
+fn parse_bigquery_create_external_table_with_connection_variants() {
+    // WITH CONNECTION alone — no OPTIONS, no explicit column list.
+    // Display normalizes the bare form by adding an empty column list, so
+    // use `one_statement_parses_to` to assert the normalized output.
+    let stmt = bigquery().one_statement_parses_to(
+        "CREATE EXTERNAL TABLE t WITH CONNECTION c",
+        "CREATE EXTERNAL TABLE t () WITH CONNECTION c",
+    );
+    let Statement::CreateTable(ct) = stmt else {
+        panic!("expected CreateTable");
+    };
+    assert!(ct.external);
+    assert!(!ct.or_replace);
+    assert!(ct.columns.is_empty());
+    assert_eq!(
+        ct.with_connection
+            .as_ref()
+            .expect("with_connection set")
+            .to_string(),
+        "c"
+    );
+    assert!(matches!(ct.table_options, CreateTableOptions::None));
+
+    // With explicit columns + OPTIONS. Exercises the Display round-trip of
+    // the `(columns) WITH CONNECTION <name> OPTIONS(...)` ordering so a
+    // future refactor that reshuffles clause order will fail this test.
+    let sql = concat!(
+        "CREATE EXTERNAL TABLE t (a INT64, b STRING) ",
+        r#"WITH CONNECTION c OPTIONS(uris = ["gs://x"])"#,
+    );
+    let stmt = bigquery().verified_stmt(sql);
+    let Statement::CreateTable(ct) = stmt else {
+        panic!("expected CreateTable");
+    };
+    assert_eq!(ct.columns.len(), 2);
+    assert_eq!(
+        ct.with_connection
+            .as_ref()
+            .expect("with_connection set")
+            .to_string(),
+        "c"
+    );
+    let CreateTableOptions::Options(opts) = &ct.table_options else {
+        panic!("expected OPTIONS, got {:?}", ct.table_options);
+    };
+    assert_eq!(opts.len(), 1);
+
+    // Baseline: no WITH CONNECTION. The new parser branch must not produce
+    // a spurious with_connection when the keyword pair is absent.
+    let stmt =
+        bigquery().verified_stmt("CREATE EXTERNAL TABLE t (a INT64) OPTIONS(uris = [\"gs://x\"])");
+    let Statement::CreateTable(ct) = stmt else {
+        panic!("expected CreateTable");
+    };
+    assert!(ct.with_connection.is_none());
+}
+
 fn bigquery() -> TestedDialects {
     TestedDialects::new(vec![Box::new(BigQueryDialect {})])
 }

tests/sqlparser_duckdb.rs

@@ -780,6 +780,7 @@ fn test_duckdb_union_datatype() {
             with_tags: Default::default(),
             base_location: Default::default(),
             external_volume: Default::default(),
+            with_connection: Default::default(),
             catalog: Default::default(),
             catalog_sync: Default::default(),
             storage_serialization_policy: Default::default(),

tests/sqlparser_mssql.rs

@@ -1999,6 +1999,7 @@ fn parse_create_table_with_valid_options() {
                 with_tags: None,
                 base_location: None,
                 external_volume: None,
+                with_connection: None,
                 catalog: None,
                 catalog_sync: None,
                 storage_serialization_policy: None,
@@ -2173,6 +2174,7 @@ fn parse_create_table_with_identity_column() {
                 with_tags: None,
                 base_location: None,
                 external_volume: None,
+                with_connection: None,
                 catalog: None,
                 catalog_sync: None,
                 storage_serialization_policy: None,

tests/sqlparser_postgres.rs

@@ -6702,6 +6702,7 @@ fn parse_trigger_related_functions() {
             with_tags: None,
             base_location: None,
             external_volume: None,
+            with_connection: None,
             catalog: None,
             catalog_sync: None,
             storage_serialization_policy: None,