feat(sqlalchemy-bigquery): add JSON type support

BigQuery has natively supported the JSON type since 2022, but the
SQLAlchemy dialect was missing visit_JSON in BigQueryTypeCompiler
and the JSON entry in _type_map. This caused UnsupportedCompilationError
when creating tables with sa.JSON() columns via Alembic.

Changes:
- Add visit_JSON to BigQueryTypeCompiler returning "JSON"
- Add "JSON" to _type_map mapping to sqlalchemy.types.JSON
- Export JSON from the package __init__.py
- Add unit test for JSON column DDL compilation

Fixes: #16123

Author: geoHeil

packages/sqlalchemy-bigquery/sqlalchemy_bigquery/__init__.py

@@ -35,6 +35,7 @@
     FLOAT64,
     INT64,
     INTEGER,
+    JSON,
     NUMERIC,
     RECORD,
     STRING,
@@ -75,6 +76,7 @@
     "FLOAT64",
     "INT64",
     "INTEGER",
+    "JSON",
     "NUMERIC",
     "RECORD",
     "STRING",

packages/sqlalchemy-bigquery/sqlalchemy_bigquery/_types.py

@@ -46,6 +46,7 @@
     "STRUCT": STRUCT,
     "TIMESTAMP": sqlalchemy.types.TIMESTAMP,
     "TIME": sqlalchemy.types.TIME,
+    "JSON": sqlalchemy.types.JSON,
 }
 
 # By convention, dialect-provided types are spelled with all upper case.
@@ -65,6 +66,7 @@
 STRING = _type_map["STRING"]
 TIMESTAMP = _type_map["TIMESTAMP"]
 TIME = _type_map["TIME"]
+JSON = _type_map["JSON"]
 
 try:
     _type_map["GEOGRAPHY"] = GEOGRAPHY

packages/sqlalchemy-bigquery/sqlalchemy_bigquery/base.py

@@ -642,6 +642,9 @@ def visit_NUMERIC(self, type_, **kw):
 
     visit_DECIMAL = visit_NUMERIC
 
+    def visit_JSON(self, type_, **kw):
+        return "JSON"
+
 
 class BigQueryDDLCompiler(DDLCompiler):
     option_datatype_mapping = {

packages/sqlalchemy-bigquery/tests/unit/test_compiler.py

@@ -41,6 +41,19 @@ def table(faux_conn, metadata):
     table.drop(faux_conn)
 
 
+def test_compile_json_column(faux_conn, metadata):
+    sqlalchemy.Table(
+        "json_table",
+        metadata,
+        sqlalchemy.Column("id", sqlalchemy.Integer),
+        sqlalchemy.Column("data", sqlalchemy.JSON),
+    )
+    metadata.create_all(faux_conn.engine)
+    assert " ".join(faux_conn.test_data["execute"][-1][0].strip().split()) == (
+        "CREATE TABLE `json_table` ( `id` INT64, `data` JSON )"
+    )
+
+
 def test_constraints_are_ignored(faux_conn, metadata):
     sqlalchemy.Table(
         "ref",