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

@@ -40,6 +40,7 @@
     "FLOAT": sqlalchemy.types.Float,
     "INT64": sqlalchemy.types.Integer,
     "INTEGER": sqlalchemy.types.Integer,
+    "JSON": sqlalchemy.types.JSON,
     "NUMERIC": sqlalchemy.types.Numeric,
     "RECORD": STRUCT,
     "STRING": sqlalchemy.types.String,
@@ -60,6 +61,7 @@
 FLOAT = _type_map["FLOAT"]
 INT64 = _type_map["INT64"]
 INTEGER = _type_map["INTEGER"]
+JSON = _type_map["JSON"]
 NUMERIC = _type_map["NUMERIC"]
 RECORD = _type_map["RECORD"]
 STRING = _type_map["STRING"]

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",