diff --git a/alembic/ddl/sqlite.py b/alembic/ddl/sqlite.py
index c260d53f..1a38d22d 100644
--- a/alembic/ddl/sqlite.py
+++ b/alembic/ddl/sqlite.py
@@ -182,9 +182,12 @@ def cast_for_batch_migrate(
         existing_transfer: Dict[str, Union[TypeEngine, Cast]],
         new_type: TypeEngine,
     ) -> None:
+        # the JSON check uses the type affinity rather than isinstance() so
+        # that a TypeDecorator which augments JSON is also detected; CASTing
+        # to JSON in SQLite yields 0 and would destroy the data
         if (
             existing.type._type_affinity is not new_type._type_affinity
-            and not isinstance(new_type, JSON)
+            and new_type._type_affinity is not JSON
         ):
             existing_transfer["expr"] = cast(
                 existing_transfer["expr"], new_type
diff --git a/docs/build/unreleased/1120.rst b/docs/build/unreleased/1120.rst
new file mode 100644
index 00000000..79fbb80a
--- /dev/null
+++ b/docs/build/unreleased/1120.rst
@@ -0,0 +1,12 @@
+.. change::
+    :tags: bug, batch migrations
+    :tickets: 1120
+
+    Fixed data loss in SQLite batch migrations when altering a column to a
+    :class:`~sqlalchemy.types.TypeDecorator` that augments
+    :class:`~sqlalchemy.types.JSON`.  The data transfer step suppresses the
+    ``CAST`` for :class:`~sqlalchemy.types.JSON`, as CASTing to ``JSON`` in
+    SQLite yields ``0``, however the check used ``isinstance()`` and therefore
+    did not match a ``TypeDecorator`` wrapping ``JSON``; every value in the
+    column was replaced with ``0``.  The check now compares type affinity, so
+    augmented types are detected as well.
diff --git a/tests/test_batch.py b/tests/test_batch.py
index 0dfd273b..cb76bffc 100644
--- a/tests/test_batch.py
+++ b/tests/test_batch.py
@@ -21,6 +21,7 @@
 from sqlalchemy import String
 from sqlalchemy import Table
 from sqlalchemy import Text
+from sqlalchemy import TypeDecorator
 from sqlalchemy import UniqueConstraint
 from sqlalchemy.dialects import sqlite as sqlite_dialect
 from sqlalchemy.schema import CreateIndex
@@ -1154,6 +1155,33 @@ def test_change_type(self):
             "ALTER TABLE _alembic_tmp_foo RENAME TO foo",
         )
 
+    def test_change_type_json_typedecorator(self):
+        """a TypeDecorator that augments JSON must not be CAST either.
+
+        CASTing to JSON in SQLite yields 0, so emitting the CAST here
+        would destroy the column's data.
+
+        """
+
+        class CustomJson(TypeDecorator):
+            impl = JSON
+            cache_ok = True
+
+        context = self._fixture()
+        self.table.append_column(Column("toj", Text))
+        with self.op.batch_alter_table(
+            "foo", copy_from=self.table
+        ) as batch_op:
+            batch_op.alter_column("toj", type_=CustomJson)
+        context.assert_(
+            "CREATE TABLE _alembic_tmp_foo (id INTEGER NOT NULL, "
+            "data VARCHAR(50), x INTEGER, toj JSON, PRIMARY KEY (id))",
+            "INSERT INTO _alembic_tmp_foo (id, data, x, toj) "
+            "SELECT foo.id, foo.data, foo.x, foo.toj FROM foo",
+            "DROP TABLE foo",
+            "ALTER TABLE _alembic_tmp_foo RENAME TO foo",
+        )
+
     def test_change_type_from_schematype(self):
         context = self._fixture()
         self.table.append_column(