Repository: ibis-project/ibis

Issue:
bug: Postgres: UndefinedFunction extracting elements from JSONB
### What happened?

I'm attempting to extract an element from a JSONB column, but `JSON_EXTRACT_PATH` is used, throwing an error (should be `JSONB_EXTRACT_PATH`?)

### What version of ibis are you using?

9.1.0

### What backend(s) are you using, if any?

Postgres

### Relevant log output

```sh
t = cls.client.con.table("controller_outputs")

res = t.select((_['controller_terms'])['mval'].name("mval")).limit(1)

res
Out[20]: 
r0 := DatabaseTable: controller_outputs
  controller_output_id     !int64
  controller_output_run_id int32
  segment_id               !int64
  controller_terms         !json # note this is actually JSONB
  value                    !float64

r1 := Project[r0]
  mval: JSONGetItem(r0.controller_terms, index='mval')

Limit[r1, n=1]

res.to_pandas()
Traceback (most recent call last):

  Cell In[21], line 1
    res.to_pandas()

  File ~/anaconda3/envs/.../lib/python3.12/site-packages/ibis/expr/types/relations.py:3545 in to_pandas
    return self.execute(**kwargs)

  File ~/anaconda3/envs/.../lib/python3.12/site-packages/ibis/expr/types/core.py:393 in execute
    return self._find_backend(use_default=True).execute(

  File ~/anaconda3/envs/.../lib/python3.12/site-packages/ibis/backends/sql/__init__.py:301 in execute
    with self._safe_raw_sql(sql) as cur:

  File ~/anaconda3/envs/.../lib/python3.12/contextlib.py:137 in __enter__
    return next(self.gen)

  File ~/anaconda3/envs/.../lib/python3.12/site-packages/ibis/backends/postgres/__init__.py:786 in _safe_raw_sql
    with contextlib.closing(self.raw_sql(*args, **kwargs)) as result:

  File ~/anaconda3/envs/.../lib/python3.12/site-packages/ibis/backends/postgres/__init__.py:812 in raw_sql
    cursor.execute(query, **kwargs)

UndefinedFunction: function json_extract_path(jsonb, unknown) does not exist
LINE 1: SELECT JSON_EXTRACT_PATH("t0"."controller_terms", 'mval...
               ^
HINT:  No function matches the given name and argument types. You might need to add explicit type casts.
```


### Code of Conduct

- [X] I agree to follow this project's Code of Conduct

Discussion:
cpcloud: @p-a-a-a-trick Can you give this PR a try?

Patch:
diff --git a/ci/schema/postgres.sql b/ci/schema/postgres.sql
index e94eec27a..540735125 100644
--- a/ci/schema/postgres.sql
+++ b/ci/schema/postgres.sql
@@ -268,6 +268,26 @@ INSERT INTO json_t VALUES
     (13, '42'),
     (14, '37.37');
 
+DROP TABLE IF EXISTS jsonb_t CASCADE;
+
+CREATE TABLE IF NOT EXISTS jsonb_t (rowid BIGINT, js JSONB);
+
+INSERT INTO jsonb_t VALUES
+    (1, '{"a": [1,2,3,4], "b": 1}'),
+    (2, '{"a":null,"b":2}'),
+    (3, '{"a":"foo", "c":null}'),
+    (4, 'null'),
+    (5, '[42,47,55]'),
+    (6, '[]'),
+    (7, '"a"'),
+    (8, '""'),
+    (9, '"b"'),
+    (10, NULL),
+    (11, 'true'),
+    (12, 'false'),
+    (13, '42'),
+    (14, '37.37');
+
 DROP TABLE IF EXISTS win CASCADE;
 CREATE TABLE win (g TEXT, x BIGINT NOT NULL, y BIGINT);
 INSERT INTO win VALUES
diff --git a/ibis/backends/pyspark/__init__.py b/ibis/backends/pyspark/__init__.py
index f5b9ac2b0..0d6df76b8 100644
--- a/ibis/backends/pyspark/__init__.py
+++ b/ibis/backends/pyspark/__init__.py
@@ -129,7 +129,10 @@ class Backend(SQLBackend, CanListCatalog, CanCreateDatabase):
         self._cached_dataframes = {}
 
     def do_connect(
-        self, session: SparkSession | None = None, mode: ConnectionMode | None = None
+        self,
+        session: SparkSession | None = None,
+        mode: ConnectionMode = "batch",
+        **kwargs,
     ) -> None:
         """Create a PySpark `Backend` for use with Ibis.
 
@@ -142,6 +145,8 @@ class Backend(SQLBackend, CanListCatalog, CanCreateDatabase):
             query executed within this connection will be interpreted as a batch
             workload. If "streaming", every source, sink, and query executed within
             this connection will be interpreted as a streaming workload.
+        kwargs
+            Additional keyword arguments used to configure the SparkSession.
 
         Examples
         --------
@@ -157,7 +162,6 @@ class Backend(SQLBackend, CanListCatalog, CanCreateDatabase):
 
             session = SparkSession.builder.getOrCreate()
 
-        mode = mode or "batch"
         if mode not in ("batch", "streaming"):
             raise com.IbisInputError(
                 f"Invalid connection mode: {mode}, must be `streaming` or `batch`"
@@ -173,6 +177,9 @@ class Backend(SQLBackend, CanListCatalog, CanCreateDatabase):
         self._session.conf.set("spark.sql.session.timeZone", "UTC")
         self._session.conf.set("spark.sql.mapKeyDedupPolicy", "LAST_WIN")
 
+        for key, value in kwargs.items():
+            self._session.conf.set(key, value)
+
     def disconnect(self) -> None:
         self._session.stop()
 
@@ -870,14 +877,15 @@ class Backend(SQLBackend, CanListCatalog, CanCreateDatabase):
             An integer to effect a specific row limit. A value of `None` means
             "no limit". The default is in `ibis/config.py`.
         **kwargs
-            PySpark Delta Lake table write arguments.
-            https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameWriter.save.html
+            Additional keyword arguments passed to
+            [pyspark.sql.DataFrameWriter](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameWriter.html).
 
         """
         if self.mode == "streaming":
             raise NotImplementedError(
                 "Writing to a Delta Lake table in streaming mode is not supported"
             )
+        self._run_pre_execute_hooks(expr)
         df = self._session.sql(self.compile(expr, params=params, limit=limit))
         df.write.format("delta").save(os.fspath(path), **kwargs)
 
diff --git a/ibis/backends/sql/compilers/postgres.py b/ibis/backends/sql/compilers/postgres.py
index e074f9f89..9f8050c4a 100644
--- a/ibis/backends/sql/compilers/postgres.py
+++ b/ibis/backends/sql/compilers/postgres.py
@@ -315,66 +315,70 @@ class PostgresCompiler(SQLGlotCompiler):
         #
         # but also postgres should really support anonymous structs
         return self.cast(
-            self.f.jsonb_extract_path(self.f.to_jsonb(arg), sge.convert(f"f{idx:d}")),
-            op.dtype,
+            self.f.jsonb_extract_path(self.f.to_jsonb(arg), f"f{idx:d}"), op.dtype
+        )
+
+    def json_typeof(self, op, arg):
+        b = "b" * op.arg.dtype.binary
+        return self.f[f"json{b}_typeof"](arg)
+
+    def json_extract_path_text(self, op, arg, *rest):
+        b = "b" * op.arg.dtype.binary
+        return self.f[f"json{b}_extract_path_text"](
+            arg,
+            *rest,
+            # this is apparently how you pass in no additional arguments to
+            # a variadic function, see the "Variadic Function Resolution"
+            # section in
+            # https://www.postgresql.org/docs/current/typeconv-func.html
+            sge.Var(this="VARIADIC ARRAY[]::TEXT[]"),
         )
 
     def visit_UnwrapJSONString(self, op, *, arg):
         return self.if_(
-            self.f.json_typeof(arg).eq(sge.convert("string")),
-            self.f.json_extract_path_text(
-                arg,
-                # this is apparently how you pass in no additional arguments to
-                # a variadic function, see the "Variadic Function Resolution"
-                # section in
-                # https://www.postgresql.org/docs/current/typeconv-func.html
-                sge.Var(this="VARIADIC ARRAY[]::TEXT[]"),
-            ),
+            self.json_typeof(op, arg).eq(sge.convert("string")),
+            self.json_extract_path_text(op, arg),
             NULL,
         )
 
     def visit_UnwrapJSONInt64(self, op, *, arg):
-        text = self.f.json_extract_path_text(
-            arg, sge.Var(this="VARIADIC ARRAY[]::TEXT[]")
-        )
+        text = self.json_extract_path_text(op, arg)
         return self.if_(
-            self.f.json_typeof(arg).eq(sge.convert("number")),
+            self.json_typeof(op, arg).eq(sge.convert("number")),
             self.cast(
-                self.if_(self.f.regexp_like(text, r"^\d+$", "g"), text, NULL),
-                op.dtype,
+                self.if_(self.f.regexp_like(text, r"^\d+$", "g"), text, NULL), op.dtype
             ),
             NULL,
         )
 
     def visit_UnwrapJSONFloat64(self, op, *, arg):
-        text = self.f.json_extract_path_text(
-            arg, sge.Var(this="VARIADIC ARRAY[]::TEXT[]")
-        )
+        text = self.json_extract_path_text(op, arg)
         return self.if_(
-            self.f.json_typeof(arg).eq(sge.convert("number")),
+            self.json_typeof(op, arg).eq(sge.convert("number")),
             self.cast(text, op.dtype),
             NULL,
         )
 
     def visit_UnwrapJSONBoolean(self, op, *, arg):
         return self.if_(
-            self.f.json_typeof(arg).eq(sge.convert("boolean")),
-            self.cast(
-                self.f.json_extract_path_text(
-                    arg, sge.Var(this="VARIADIC ARRAY[]::TEXT[]")
-                ),
-                op.dtype,
-            ),
+            self.json_typeof(op, arg).eq(sge.convert("boolean")),
+            self.cast(self.json_extract_path_text(op, arg), op.dtype),
             NULL,
         )
 
+    def visit_JSONGetItem(self, op, *, arg, index):
+        if op.arg.dtype.binary:
+            return self.f.jsonb_extract_path(arg, self.cast(index, dt.string))
+        return super().visit_JSONGetItem(op, arg=arg, index=index)
+
     def visit_StructColumn(self, op, *, names, values):
         return self.f.row(*map(self.cast, values, op.dtype.types))
 
     def visit_ToJSONArray(self, op, *, arg):
+        b = "b" * op.arg.dtype.binary
         return self.if_(
-            self.f.json_typeof(arg).eq(sge.convert("array")),
-            self.f.array(sg.select(STAR).from_(self.f.json_array_elements(arg))),
+            self.json_typeof(op, arg).eq(sge.convert("array")),
+            self.f.array(sg.select(STAR).from_(self.f[f"json{b}_array_elements"](arg))),
             NULL,
         )
 
diff --git a/ibis/backends/sql/datatypes.py b/ibis/backends/sql/datatypes.py
index 62f3a24ed..c18c97472 100644
--- a/ibis/backends/sql/datatypes.py
+++ b/ibis/backends/sql/datatypes.py
@@ -34,7 +34,7 @@ _from_sqlglot_types = {
     typecode.INT: dt.Int32,
     typecode.IPADDRESS: dt.INET,
     typecode.JSON: dt.JSON,
-    typecode.JSONB: dt.JSON,
+    typecode.JSONB: partial(dt.JSON, binary=True),
     typecode.LONGBLOB: dt.Binary,
     typecode.LONGTEXT: dt.String,
     typecode.MEDIUMBLOB: dt.Binary,
@@ -115,7 +115,6 @@ _to_sqlglot_types = {
     dt.Float64: typecode.DOUBLE,
     dt.String: typecode.VARCHAR,
     dt.Binary: typecode.VARBINARY,
-    dt.JSON: typecode.JSON,
     dt.INET: typecode.INET,
     dt.UUID: typecode.UUID,
     dt.MACADDR: typecode.VARCHAR,
@@ -325,6 +324,10 @@ class SqlglotType(TypeMapper):
             srid = int(srid.this.this)
         return typeclass(geotype="geography", nullable=cls.default_nullable, srid=srid)
 
+    @classmethod
+    def _from_ibis_JSON(cls, dtype: dt.JSON) -> sge.DataType:
+        return sge.DataType(this=typecode.JSONB if dtype.binary else typecode.JSON)
+
     @classmethod
     def _from_ibis_Interval(cls, dtype: dt.Interval) -> sge.DataType:
         assert dtype.unit is not None, "interval unit cannot be None"
diff --git a/ibis/expr/datatypes/core.py b/ibis/expr/datatypes/core.py
index 2c8316de7..f7c4324ce 100644
--- a/ibis/expr/datatypes/core.py
+++ b/ibis/expr/datatypes/core.py
@@ -936,6 +936,13 @@ class JSON(Variadic):
     scalar = "JSONScalar"
     column = "JSONColumn"
 
+    binary: bool = False
+    """True if JSON is stored as binary, e.g., JSONB in PostgreSQL."""
+
+    @property
+    def _pretty_piece(self) -> str:
+        return "b" * self.binary
+
 
 @public
 class GeoSpatial(DataType):
@@ -1066,7 +1073,8 @@ multilinestring = MultiLineString()
 multipoint = MultiPoint()
 multipolygon = MultiPolygon()
 # json
-json = JSON()
+json = JSON(binary=False)
+jsonb = JSON(binary=True)
 # special string based data type
 uuid = UUID()
 macaddr = MACADDR()
@@ -1107,6 +1115,7 @@ public(
     multipoint=multipoint,
     multipolygon=multipolygon,
     json=json,
+    jsonb=jsonb,
     uuid=uuid,
     macaddr=macaddr,
     inet=inet,
diff --git a/ibis/expr/datatypes/parse.py b/ibis/expr/datatypes/parse.py
index ac802e7f8..22ec35533 100644
--- a/ibis/expr/datatypes/parse.py
+++ b/ibis/expr/datatypes/parse.py
@@ -200,7 +200,7 @@ def parse(
         | array
         | map
         | struct
-        | spaceless_string("json", "uuid", "macaddr", "inet").map(
+        | spaceless_string("jsonb", "json", "uuid", "macaddr", "inet").map(
             functools.partial(getattr, dt)
         )
         | spaceless_string("int").result(dt.int64)
diff --git a/ibis/expr/operations/json.py b/ibis/expr/operations/json.py
index bf5675fad..0f009af51 100644
--- a/ibis/expr/operations/json.py
+++ b/ibis/expr/operations/json.py
@@ -6,6 +6,7 @@ from public import public
 
 import ibis.expr.datatypes as dt
 import ibis.expr.rules as rlz
+from ibis.common.annotations import attribute
 from ibis.expr.operations import Value
 
 
@@ -16,7 +17,7 @@ class JSONGetItem(Value):
     arg: Value[dt.JSON]
     index: Value[dt.String | dt.Integer]
 
-    dtype = dt.json
+    dtype = rlz.dtype_like("arg")
     shape = rlz.shape_like("args")
 
 
@@ -26,9 +27,12 @@ class ToJSONArray(Value):
 
     arg: Value[dt.JSON]
 
-    dtype = dt.Array(dt.json)
     shape = rlz.shape_like("arg")
 
+    @attribute
+    def dtype(self) -> dt.DataType:
+        return dt.Array(self.arg.dtype)
+
 
 @public
 class ToJSONMap(Value):
@@ -36,9 +40,12 @@ class ToJSONMap(Value):
 
     arg: Value[dt.JSON]
 
-    dtype = dt.Map(dt.string, dt.json)
     shape = rlz.shape_like("arg")
 
+    @attribute
+    def dtype(self) -> dt.DataType:
+        return dt.Map(dt.string, self.arg.dtype)
+
 
 @public
 class UnwrapJSONString(Value):


Test patch:
diff --git a/ibis/backends/postgres/tests/test_client.py b/ibis/backends/postgres/tests/test_client.py
index bde3637cf..496ae0190 100644
--- a/ibis/backends/postgres/tests/test_client.py
+++ b/ibis/backends/postgres/tests/test_client.py
@@ -173,7 +173,7 @@ def test_create_and_drop_table(con, temp_table, params):
             ("numeric", dt.decimal),
             ("numeric(3, 2)", dt.Decimal(3, 2)),
             ("uuid", dt.uuid),
-            ("jsonb", dt.json),
+            ("jsonb", dt.jsonb),
             ("geometry", dt.geometry),
             ("geography", dt.geography),
         ]
diff --git a/ibis/backends/postgres/tests/test_json.py b/ibis/backends/postgres/tests/test_json.py
index 6f6d04b58..219562b81 100644
--- a/ibis/backends/postgres/tests/test_json.py
+++ b/ibis/backends/postgres/tests/test_json.py
@@ -4,10 +4,20 @@ from __future__ import annotations
 
 import json
 
+import numpy as np
+import pandas as pd
+import pandas.testing as tm
 import pytest
 from pytest import param
 
 import ibis
+import ibis.expr.datatypes as dt
+import ibis.expr.types as ir
+
+
+@pytest.fixture(scope="module")
+def jsonb_t(con):
+    return con.table("jsonb_t")
 
 
 @pytest.mark.parametrize("data", [param({"status": True}, id="status")])
@@ -16,3 +26,71 @@ def test_json(data, alltypes):
     expr = alltypes[[alltypes.id, lit]].head(1)
     df = expr.execute()
     assert df["tmp"].iloc[0] == data
+
+
+def test_jsonb_extract_path(con):
+    json_t = con.table("json_t")
+    jsonb_t = con.table("jsonb_t")
+
+    assert json_t.js.type() == dt.JSON(binary=False)
+    assert jsonb_t.js.type() == dt.JSON(binary=True)
+
+    tm.assert_series_equal(jsonb_t.js["a"].execute(), json_t.js["a"].execute())
+
+
+def test_json_getitem_object(jsonb_t):
+    expr_fn = lambda t: t.js["a"].name("res")
+    expected = frozenset([(1, 2, 3, 4), None, "foo"] + [None] * 3)
+    expr = expr_fn(jsonb_t)
+    result = frozenset(
+        expr.execute()
+        .map(lambda o: tuple(o) if isinstance(o, list) else o)
+        .replace({np.nan: None})
+    )
+    assert result == expected
+
+
+def test_json_getitem_array(jsonb_t):
+    expr_fn = lambda t: t.js[1].name("res")
+    expected = frozenset([None] * 4 + [47, None])
+    expr = expr_fn(jsonb_t)
+    result = frozenset(expr.execute().replace({np.nan: None}))
+    assert result == expected
+
+
+def test_json_array(jsonb_t):
+    expr = jsonb_t.mutate("rowid", res=jsonb_t.js.array).order_by("rowid")
+    result = expr.execute().res
+    expected = pd.Series(
+        [None, None, None, None, [42, 47, 55], []] + [None] * 8,
+        name="res",
+        dtype="object",
+    )
+    tm.assert_series_equal(result, expected)
+
+
+@pytest.mark.parametrize(
+    ("typ", "expected_data"),
+    [
+        ("str", [None] * 6 + ["a", "", "b"] + [None] * 5),
+        ("int", [None] * 12 + [42, None]),
+        ("float", [None] * 12 + [42.0, 37.37]),
+        ("bool", [None] * 10 + [True, False, None, None]),
+    ],
+    ids=["str", "int", "float", "bool"],
+)
+@pytest.mark.parametrize(
+    "expr_fn", [getattr, ir.JSONValue.unwrap_as], ids=["getattr", "unwrap_as"]
+)
+def test_json_unwrap(jsonb_t, typ, expected_data, expr_fn):
+    expr = expr_fn(jsonb_t.js, typ).name("res")
+    result = expr.execute()
+    expected = pd.Series(expected_data, name="res", dtype="object")
+    tm.assert_series_equal(
+        result.replace(np.nan, None).fillna(pd.NA).sort_values().reset_index(drop=True),
+        expected.replace(np.nan, None)
+        .fillna(pd.NA)
+        .sort_values()
+        .reset_index(drop=True),
+        check_dtype=False,
+    )
diff --git a/ibis/backends/risingwave/tests/test_client.py b/ibis/backends/risingwave/tests/test_client.py
index 255a6264e..1d2ce7612 100644
--- a/ibis/backends/risingwave/tests/test_client.py
+++ b/ibis/backends/risingwave/tests/test_client.py
@@ -110,7 +110,7 @@ def test_create_and_drop_table(con, temp_table):
             ("timestamp with time zone", dt.Timestamp("UTC", scale=6)),
             ("interval", dt.Interval("s")),
             ("numeric", dt.decimal),
-            ("jsonb", dt.json),
+            ("jsonb", dt.JSON(binary=True)),
         ]
     ],
 )
diff --git a/ibis/backends/snowflake/tests/test_client.py b/ibis/backends/snowflake/tests/test_client.py
index e50b4b716..ea68fa3a9 100644
--- a/ibis/backends/snowflake/tests/test_client.py
+++ b/ibis/backends/snowflake/tests/test_client.py
@@ -438,3 +438,18 @@ def test_table_unnest_with_empty_strings(con):
     expr = t.unnest(t.x)["x"]
     result = con.execute(expr)
     assert Counter(result.values) == expected
+
+
+def test_insert_dict_variants(con):
+    name = gen_name("test_insert_dict_variants")
+
+    t = con.create_table(name, schema=ibis.schema({"a": "int", "b": "str"}), temp=True)
+    assert len(t.execute()) == 0
+
+    data = [{"a": 1, "b": "a"}, {"a": 2, "b": "b"}]
+
+    con.insert(name, data)
+    assert len(t.execute()) == 2
+
+    con.insert(name, ibis.memtable(data))
+    assert len(t.execute()) == 4
diff --git a/ibis/backends/tests/test_client.py b/ibis/backends/tests/test_client.py
index 1d7ea55c5..6c1398c4c 100644
--- a/ibis/backends/tests/test_client.py
+++ b/ibis/backends/tests/test_client.py
@@ -44,7 +44,6 @@ if TYPE_CHECKING:
 
 pa = pytest.importorskip("pyarrow")
 ds = pytest.importorskip("pyarrow.dataset")
-pl = pytest.importorskip("polars", reason="Polars is not installed")
 
 
 @pytest.fixture
@@ -978,12 +977,12 @@ def test_self_join_memory_table(backend, con, monkeypatch):
         ),
         param(lambda: pd.DataFrame({"a": ["a"], "b": [1]}), "df_pandas", id="pandas"),
         param(
-            lambda: pl.DataFrame({"a": ["a"], "b": [1]}),
+            lambda: pytest.importorskip("polars").DataFrame({"a": ["a"], "b": [1]}),
             "df_polars_eager",
             id="polars dataframe",
         ),
         param(
-            lambda: pl.LazyFrame({"a": ["a"], "b": [1]}),
+            lambda: pytest.importorskip("polars").LazyFrame({"a": ["a"], "b": [1]}),
             "df_polars_lazy",
             id="polars lazyframe",
         ),
@@ -1003,12 +1002,16 @@ def test_self_join_memory_table(backend, con, monkeypatch):
             id="memtable pandas",
         ),
         param(
-            lambda: ibis.memtable(pl.DataFrame({"a": ["a"], "b": [1]})),
+            lambda: ibis.memtable(
+                pytest.importorskip("polars").DataFrame({"a": ["a"], "b": [1]})
+            ),
             "memtable_polars_eager",
             id="memtable polars dataframe",
         ),
         param(
-            lambda: ibis.memtable(pl.LazyFrame({"a": ["a"], "b": [1]})),
+            lambda: ibis.memtable(
+                pytest.importorskip("polars").LazyFrame({"a": ["a"], "b": [1]})
+            ),
             "memtable_polars_lazy",
             id="memtable polars lazyframe",
         ),
@@ -1333,162 +1336,6 @@ def test_create_table_timestamp(con, temp_table):
     assert result == schema
 
 
-@mark.notimpl(["datafusion", "flink", "impala", "trino", "druid"])
-@mark.never(
-    ["mssql"],
-    reason="mssql supports support temporary tables through naming conventions",
-)
-@mark.notimpl(["exasol"], reason="Exasol does not support temporary tables")
-@pytest.mark.never(
-    ["risingwave"],
-    raises=com.UnsupportedOperationError,
-    reason="Feature is not yet implemented: CREATE TEMPORARY TABLE",
-)
-def test_persist_expression(backend, alltypes):
-    non_persisted_table = alltypes.mutate(
-        test_column=ibis.literal("calculation"), other_calc=ibis.literal("xyz")
-    )
-    persisted_table = non_persisted_table.cache()
-    backend.assert_frame_equal(
-        non_persisted_table.to_pandas(), persisted_table.to_pandas()
-    )
-
-
-@mark.notimpl(["datafusion", "flink", "impala", "trino", "druid"])
-@mark.never(
-    ["mssql"],
-    reason="mssql supports support temporary tables through naming conventions",
-)
-@mark.notimpl(["exasol"], reason="Exasol does not support temporary tables")
-@pytest.mark.never(
-    ["risingwave"],
-    raises=com.UnsupportedOperationError,
-    reason="Feature is not yet implemented: CREATE TEMPORARY TABLE",
-)
-def test_persist_expression_contextmanager(backend, con, alltypes):
-    non_cached_table = alltypes.mutate(
-        test_column=ibis.literal("calculation"), other_column=ibis.literal("big calc")
-    )
-    with non_cached_table.cache() as cached_table:
-        backend.assert_frame_equal(
-            non_cached_table.to_pandas(), cached_table.to_pandas()
-        )
-    assert non_cached_table.op() not in con._query_cache.cache
-
-
-@mark.notimpl(["datafusion", "flink", "impala", "trino", "druid"])
-@mark.never(
-    ["mssql"],
-    reason="mssql supports support temporary tables through naming conventions",
-)
-@mark.notimpl(["exasol"], reason="Exasol does not support temporary tables")
-@pytest.mark.never(
-    ["risingwave"],
-    raises=com.UnsupportedOperationError,
-    reason="Feature is not yet implemented: CREATE TEMPORARY TABLE",
-)
-def test_persist_expression_contextmanager_ref_count(backend, con, alltypes):
-    non_cached_table = alltypes.mutate(
-        test_column=ibis.literal("calculation"), other_column=ibis.literal("big calc 2")
-    )
-    op = non_cached_table.op()
-    cached_table = non_cached_table.cache()
-    backend.assert_frame_equal(non_cached_table.to_pandas(), cached_table.to_pandas())
-
-    assert op in con._query_cache.cache
-    del cached_table
-    assert op not in con._query_cache.cache
-
-
-@mark.notimpl(["datafusion", "flink", "impala", "trino", "druid"])
-@mark.never(
-    ["mssql"],
-    reason="mssql supports support temporary tables through naming conventions",
-)
-@pytest.mark.never(
-    ["risingwave"],
-    raises=com.UnsupportedOperationError,
-    reason="Feature is not yet implemented: CREATE TEMPORARY TABLE",
-)
-@mark.notimpl(["exasol"], reason="Exasol does not support temporary tables")
-def test_persist_expression_multiple_refs(backend, con, alltypes):
-    non_cached_table = alltypes.mutate(
-        test_column=ibis.literal("calculation"), other_column=ibis.literal("big calc 2")
-    )
-    op = non_cached_table.op()
-    cached_table = non_cached_table.cache()
-
-    backend.assert_frame_equal(non_cached_table.to_pandas(), cached_table.to_pandas())
-
-    name = cached_table.op().name
-    nested_cached_table = non_cached_table.cache()
-
-    # cached tables are identical and reusing the same op
-    assert cached_table.op() is nested_cached_table.op()
-    # table is cached
-    assert op in con._query_cache.cache
-
-    # deleting the first reference, leaves table in cache
-    del nested_cached_table
-    assert op in con._query_cache.cache
-
-    # deleting the last reference, releases table from cache
-    del cached_table
-    assert op not in con._query_cache.cache
-
-    # assert that table has been dropped
-    assert name not in con.list_tables()
-
-
-@mark.notimpl(["datafusion", "flink", "impala", "trino", "druid"])
-@mark.never(
-    ["mssql"],
-    reason="mssql supports support temporary tables through naming conventions",
-)
-@mark.notimpl(["exasol"], reason="Exasol does not support temporary tables")
-@pytest.mark.never(
-    ["risingwave"],
-    raises=com.UnsupportedOperationError,
-    reason="Feature is not yet implemented: CREATE TEMPORARY TABLE",
-)
-def test_persist_expression_repeated_cache(alltypes, con):
-    non_cached_table = alltypes.mutate(
-        test_column=ibis.literal("calculation"), other_column=ibis.literal("big calc 2")
-    )
-    cached_table = non_cached_table.cache()
-    nested_cached_table = cached_table.cache()
-    name = cached_table.op().name
-
-    assert not nested_cached_table.to_pandas().empty
-
-    del nested_cached_table, cached_table
-
-    assert name not in con.list_tables()
-
-
-@mark.notimpl(["datafusion", "flink", "impala", "trino", "druid"])
-@mark.notimpl(["exasol"], reason="Exasol does not support temporary tables")
-@pytest.mark.never(
-    ["risingwave"],
-    raises=com.UnsupportedOperationError,
-    reason="Feature is not yet implemented: CREATE TEMPORARY TABLE",
-)
-def test_persist_expression_release(con, alltypes):
-    non_cached_table = alltypes.mutate(
-        test_column=ibis.literal("calculation"), other_column=ibis.literal("big calc 3")
-    )
-    cached_table = non_cached_table.cache()
-    cached_table.release()
-
-    assert non_cached_table.op() not in con._query_cache.cache
-
-    # a second release does not hurt
-    cached_table.release()
-
-    with pytest.raises(Exception, match=cached_table.op().name):
-        cached_table.execute()
-
-
 @contextlib.contextmanager
 def gen_test_name(con: BaseBackend):
     name = gen_name("test_table")
diff --git a/ibis/backends/tests/test_expr_caching.py b/ibis/backends/tests/test_expr_caching.py
new file mode 100644
index 000000000..d2623b0a5
--- /dev/null
+++ b/ibis/backends/tests/test_expr_caching.py
@@ -0,0 +1,152 @@
+from __future__ import annotations
+
+import pytest
+from pytest import mark
+
+import ibis
+import ibis.common.exceptions as com
+
+pa = pytest.importorskip("pyarrow")
+ds = pytest.importorskip("pyarrow.dataset")
+
+
+@mark.notimpl(["datafusion", "flink", "impala", "trino", "druid"])
+@mark.never(
+    ["mssql"],
+    reason="mssql supports support temporary tables through naming conventions",
+)
+@mark.notimpl(["exasol"], reason="Exasol does not support temporary tables")
+@pytest.mark.never(
+    ["risingwave"],
+    raises=com.UnsupportedOperationError,
+    reason="Feature is not yet implemented: CREATE TEMPORARY TABLE",
+)
+def test_persist_expression(backend, alltypes):
+    non_persisted_table = alltypes.mutate(
+        test_column=ibis.literal("calculation"), other_calc=ibis.literal("xyz")
+    )
+    persisted_table = non_persisted_table.cache()
+    backend.assert_frame_equal(
+        non_persisted_table.to_pandas(), persisted_table.to_pandas()
+    )
+
+
+@mark.notimpl(["datafusion", "flink", "impala", "trino", "druid"])
+@mark.never(
+    ["mssql"],
+    reason="mssql supports support temporary tables through naming conventions",
+)
+@mark.notimpl(["exasol"], reason="Exasol does not support temporary tables")
+@pytest.mark.never(
+    ["risingwave"],
+    raises=com.UnsupportedOperationError,
+    reason="Feature is not yet implemented: CREATE TEMPORARY TABLE",
+)
+def test_persist_expression_contextmanager(backend, con, alltypes):
+    non_cached_table = alltypes.mutate(
+        test_column=ibis.literal("calculation"), other_column=ibis.literal("big calc")
+    )
+    with non_cached_table.cache() as cached_table:
+        backend.assert_frame_equal(
+            non_cached_table.to_pandas(), cached_table.to_pandas()
+        )
+    assert non_cached_table.op() not in con._query_cache.cache
+
+
+@mark.notimpl(["datafusion", "flink", "impala", "trino", "druid"])
+@mark.never(
+    ["mssql"],
+    reason="mssql supports support temporary tables through naming conventions",
+)
+@pytest.mark.never(
+    ["risingwave"],
+    raises=com.UnsupportedOperationError,
+    reason="Feature is not yet implemented: CREATE TEMPORARY TABLE",
+)
+@mark.notimpl(["exasol"], reason="Exasol does not support temporary tables")
+@mark.notyet(
+    ["pyspark"],
+    raises=AssertionError,
+    reason=(
+        "PySpark holds on to `cached_table` in the stack frame of an internal function. "
+        "Caching works, but this test is probably too strict for PySpark. "
+        "On the other hand, it's hard to write a test that makes assertions after weakref "
+        "finalizers are called."
+    ),
+)
+def test_persist_expression_multiple_refs(backend, con, alltypes):
+    non_cached_table = alltypes.mutate(
+        test_column=ibis.literal("calculation"), other_column=ibis.literal("big calc 2")
+    )
+    op = non_cached_table.op()
+    cached_table = non_cached_table.cache()
+
+    backend.assert_frame_equal(non_cached_table.to_pandas(), cached_table.to_pandas())
+
+    name = cached_table.op().name
+    nested_cached_table = non_cached_table.cache()
+
+    # cached tables are identical and reusing the same op
+    assert cached_table.op() is nested_cached_table.op()
+    # table is cached
+    assert op in con._query_cache.cache
+
+    # deleting the first reference, leaves table in cache
+    del nested_cached_table
+    assert op in con._query_cache.cache
+
+    # deleting the last reference, releases table from cache
+    del cached_table
+    assert op not in con._query_cache.cache
+
+    # assert that table has been dropped
+    assert name not in con.list_tables()
+
+
+@mark.notimpl(["datafusion", "flink", "impala", "trino", "druid"])
+@mark.never(
+    ["mssql"],
+    reason="mssql supports support temporary tables through naming conventions",
+)
+@mark.notimpl(["exasol"], reason="Exasol does not support temporary tables")
+@pytest.mark.never(
+    ["risingwave"],
+    raises=com.UnsupportedOperationError,
+    reason="Feature is not yet implemented: CREATE TEMPORARY TABLE",
+)
+def test_persist_expression_repeated_cache(alltypes, con):
+    non_cached_table = alltypes.mutate(
+        test_column=ibis.literal("calculation"), other_column=ibis.literal("big calc 2")
+    )
+    cached_table = non_cached_table.cache()
+    nested_cached_table = cached_table.cache()
+    name = cached_table.op().name
+
+    assert not nested_cached_table.to_pandas().empty
+
+    del nested_cached_table, cached_table
+
+    assert name not in con.list_tables()
+
+
+@mark.notimpl(["datafusion", "flink", "impala", "trino", "druid"])
+@mark.notimpl(["exasol"], reason="Exasol does not support temporary tables")
+@pytest.mark.never(
+    ["risingwave"],
+    raises=com.UnsupportedOperationError,
+    reason="Feature is not yet implemented: CREATE TEMPORARY TABLE",
+)
+def test_persist_expression_release(con, alltypes):
+    non_cached_table = alltypes.mutate(
+        test_column=ibis.literal("calculation"), other_column=ibis.literal("big calc 3")
+    )
+    cached_table = non_cached_table.cache()
+    cached_table.release()
+
+    assert non_cached_table.op() not in con._query_cache.cache
+
+    # a second release does not hurt
+    cached_table.release()
+
+    with pytest.raises(Exception, match=cached_table.op().name):
+        cached_table.execute()
diff --git a/ibis/tests/strategies.py b/ibis/tests/strategies.py
index d31e1ca03..8d2e7cdb1 100644
--- a/ibis/tests/strategies.py
+++ b/ibis/tests/strategies.py
@@ -81,7 +81,7 @@ def binary_dtype(nullable=_nullable):
 
 
 def json_dtype(nullable=_nullable):
-    return st.builds(dt.JSON, nullable=nullable)
+    return st.builds(dt.JSON, binary=st.booleans(), nullable=nullable)
 
 
 def inet_dtype(nullable=_nullable):