diff --git a/docs/reference/streamfield/blocks.rst b/docs/reference/streamfield/blocks.rst
index c27794b452..33b20de4b2 100644
--- a/docs/reference/streamfield/blocks.rst
+++ b/docs/reference/streamfield/blocks.rst
@@ -6,17 +6,21 @@ StreamField block reference
 This document details the block types provided by Wagtail for use in :ref:`StreamField <streamfield>`, and how they can be combined into new block types.
 
 
-.. class:: wagtail.fields.StreamField(blocks, blank=False, min_num=None, max_num=None, block_counts=None)
+.. class:: wagtail.fields.StreamField(blocks, use_json_field=None, blank=False, min_num=None, max_num=None, block_counts=None, collapsed=False)
 
    A model field for representing long-form content as a sequence of content blocks of various types. See :ref:`streamfield`.
 
    :param blocks: A list of block types, passed as either a list of ``(name, block_definition)`` tuples or a ``StreamBlock`` instance.
+   :param use_json_field: When true, the field uses :class:`~django.db.models.JSONField` as its internal type, allowing the use of ``JSONField`` lookups and transforms. When false, it uses :class:`~django.db.models.TextField` instead. This argument **must** be set to ``True``/``False``.
    :param blank: When false (the default), at least one block must be provided for the field to be considered valid.
    :param min_num: Minimum number of sub-blocks that the stream must have.
    :param max_num: Maximum number of sub-blocks that the stream may have.
    :param block_counts: Specifies the minimum and maximum number of each block type, as a dictionary mapping block names to dicts with (optional) ``min_num`` and ``max_num`` fields.
    :param collapsed: When true, all blocks are initially collapsed.
 
+   .. versionchanged:: 2.17
+     The required ``use_json_field`` argument is added.
+
    .. code-block:: python
 
       body = StreamField([
@@ -26,7 +30,7 @@ This document details the block types provided by Wagtail for use in :ref:`Strea
       ], block_counts={
           'heading': {'min_num': 1},
           'image': {'max_num': 5},
-      })
+      }, use_json_field=True)
 
 
 Block options
diff --git a/docs/releases/2.17.md b/docs/releases/2.17.md
index 490063cd86..03f16917ba 100644
--- a/docs/releases/2.17.md
+++ b/docs/releases/2.17.md
@@ -148,3 +148,9 @@ When overriding the `get_form_class` method of a ModelAdmin `CreateView` or `Edi
 
 -   The `size` argument was used to add a `length` parameter to the HTTP header.
 -   This was never part of the HTTP/1.0 and HTTP/1.1 specifications see [RFC7232](https://httpwg.org/specs/rfc7232.html#header.if-modified-since) and existed only as a an unofficial implementation in IE browsers.
+
+### `StreamField`s must explicitly set `use_json_field` argument to `True`/`False`
+
+`StreamField` now requires a `use_json_field` keyword argument that can be set to `True`/`False`. If set to `True`, the field will use `JSONField` as its internal type instead of `TextField`, which will change the data type used on the database and allow you to use `JSONField` lookups and transforms on the `StreamField`. If set to `False`, the field will keep its previous behaviour and no database changes will be made. If set to `None` (the default), the field will keep its previous behaviour and a warning (`RemovedInWagtail219Warning`) will appear.
+
+After setting the keyword argument, make sure to generate and run the migrations for the models.
diff --git a/docs/topics/streamfield.rst b/docs/topics/streamfield.rst
index 01d754cd2e..2486d565df 100644
--- a/docs/topics/streamfield.rst
+++ b/docs/topics/streamfield.rst
@@ -547,6 +547,9 @@ Migrating RichTextFields to StreamField
 
 If you change an existing RichTextField to a StreamField, the database migration will complete with no errors, since both fields use a text column within the database. However, StreamField uses a JSON representation for its data, so the existing text requires an extra conversion step in order to become accessible again. For this to work, the StreamField needs to include a RichTextBlock as one of the available block types. (When updating the model, don't forget to change ``FieldPanel`` to ``StreamFieldPanel`` too.) Create the migration as normal using ``./manage.py makemigrations``, then edit it as follows (in this example, the 'body' field of the ``demo.BlogPage`` model is being converted to a StreamField with a RichTextBlock named ``rich_text``):
 
+.. note::
+    This migration cannot be used if the StreamField has the ``use_json_field`` argument set to ``True``. To migrate, set the ``use_json_field`` argument to ``False`` first, migrate the data, then set it back to ``True``.
+
 .. code-block:: python
 
     # -*- coding: utf-8 -*-
diff --git a/wagtail/fields.py b/wagtail/fields.py
index f7b504d00b..a527d0df4e 100644
--- a/wagtail/fields.py
+++ b/wagtail/fields.py
@@ -1,11 +1,14 @@
 import json
+import warnings
 
 from django.core.serializers.json import DjangoJSONEncoder
 from django.db import models
+from django.db.models.fields.json import KeyTransform
 from django.utils.encoding import force_str
 
 from wagtail.blocks import Block, BlockField, StreamBlock, StreamValue
 from wagtail.rich_text import get_text_for_indexing
+from wagtail.utils.deprecation import RemovedInWagtail219Warning
 
 
 class RichTextField(models.TextField):
@@ -63,8 +66,7 @@ class Creator:
 
 
 class StreamField(models.Field):
-    def __init__(self, block_types, **kwargs):
-
+    def __init__(self, block_types, use_json_field=None, **kwargs):
         # extract kwargs that are to be passed on to the block, not handled by super
         block_opts = {}
         for arg in ["min_num", "max_num", "block_counts", "collapsed"]:
@@ -78,6 +80,9 @@ class StreamField(models.Field):
 
         super().__init__(**kwargs)
 
+        self.use_json_field = use_json_field
+        self._check_json_field()
+
         if isinstance(block_types, Block):
             # use the passed block as the top-level block
             self.stream_block = block_types
@@ -90,13 +95,36 @@ class StreamField(models.Field):
 
         self.stream_block.set_meta_options(block_opts)
 
+    @property
+    def json_field(self):
+        return models.JSONField(encoder=DjangoJSONEncoder)
+
+    def _check_json_field(self):
+        if type(self.use_json_field) is not bool:
+            warnings.warn(
+                f"StreamField must explicitly set use_json_field argument to True/False instead of {self.use_json_field}.",
+                RemovedInWagtail219Warning,
+                stacklevel=3,
+            )
+
     def get_internal_type(self):
-        return "TextField"
+        return "JSONField" if self.use_json_field else "TextField"
+
+    def get_lookup(self, lookup_name):
+        if self.use_json_field:
+            return self.json_field.get_lookup(lookup_name)
+        return super().get_lookup(lookup_name)
+
+    def get_transform(self, lookup_name):
+        if self.use_json_field:
+            return self.json_field.get_transform(lookup_name)
+        return super().get_transform(lookup_name)
 
     def deconstruct(self):
         name, path, _, kwargs = super().deconstruct()
         block_types = list(self.stream_block.child_blocks.items())
         args = [block_types]
+        kwargs["use_json_field"] = self.use_json_field
         return name, path, args, kwargs
 
     def to_python(self, value):
@@ -122,6 +150,17 @@ class StreamField(models.Field):
                 return StreamValue(self.stream_block, [])
 
             return self.stream_block.to_python(unpacked_value)
+        elif (
+            self.use_json_field
+            and value
+            and isinstance(value, list)
+            and isinstance(value[0], dict)
+        ):
+            # The value is already unpacked since JSONField-based StreamField should
+            # accept deserialised values (no need to call json.dumps() first).
+            # In addition, the value is not a list of (block_name, value) tuples
+            # handled in the `else` block.
+            return self.stream_block.to_python(value)
         else:
             # See if it looks like the standard non-smart representation of a
             # StreamField value: a list of (block_name, value) tuples
@@ -148,12 +187,29 @@ class StreamField(models.Field):
             # for reverse migrations that convert StreamField data back into plain text
             # fields.)
             return value.raw_text
-        else:
+        elif isinstance(value, StreamValue) or not self.use_json_field:
+            # StreamValue instances must be prepared first.
+            # Before use_json_field was implemented, this is also the value used in queries.
             return json.dumps(
                 self.stream_block.get_prep_value(value), cls=DjangoJSONEncoder
             )
+        else:
+            # When querying with JSONField features, the rhs might not be a StreamValue.
+            return self.json_field.get_prep_value(value)
 
     def from_db_value(self, value, expression, connection):
+        if self.use_json_field and isinstance(expression, KeyTransform):
+            # This could happen when using JSONField key transforms,
+            # e.g. Page.object.values('body__0').
+            try:
+                # We might be able to properly resolve to the appropriate StreamValue
+                # based on `expression` and `self.stream_block`, but it might be too
+                # complicated to do so. For now, just deserialise the value.
+                return json.loads(value)
+            except ValueError:
+                # Just in case the extracted value is not valid JSON.
+                return value
+
         return self.to_python(value)
 
     def formfield(self, **kwargs):
diff --git a/wagtail/test/testapp/migrations/0062_alter_addedstreamfieldwithemptylistdefaultpage_body_and_more.py b/wagtail/test/testapp/migrations/0062_alter_addedstreamfieldwithemptylistdefaultpage_body_and_more.py
new file mode 100644
index 0000000000..b7cd9714a8
--- /dev/null
+++ b/wagtail/test/testapp/migrations/0062_alter_addedstreamfieldwithemptylistdefaultpage_body_and_more.py
@@ -0,0 +1,83 @@
+# Generated by Django 4.0.3 on 2022-03-18 06:36
+
+from django.db import migrations
+import wagtail.blocks
+import wagtail.contrib.table_block.blocks
+import wagtail.fields
+import wagtail.images.blocks
+import wagtail.test.testapp.models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('tests', '0061_tag_fk_for_django_4'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='addedstreamfieldwithemptylistdefaultpage',
+            name='body',
+            field=wagtail.fields.StreamField([('title', wagtail.blocks.CharBlock())], default=[], use_json_field=False),
+        ),
+        migrations.AlterField(
+            model_name='addedstreamfieldwithemptystringdefaultpage',
+            name='body',
+            field=wagtail.fields.StreamField([('title', wagtail.blocks.CharBlock())], default='', use_json_field=False),
+        ),
+        migrations.AlterField(
+            model_name='addedstreamfieldwithoutdefaultpage',
+            name='body',
+            field=wagtail.fields.StreamField([('title', wagtail.blocks.CharBlock())], use_json_field=False),
+        ),
+        migrations.AlterField(
+            model_name='blockcountsstreammodel',
+            name='body',
+            field=wagtail.fields.StreamField([('text', wagtail.blocks.CharBlock()), ('rich_text', wagtail.blocks.RichTextBlock()), ('image', wagtail.images.blocks.ImageChooserBlock())], use_json_field=False),
+        ),
+        migrations.AlterField(
+            model_name='customrichblockfieldpage',
+            name='body',
+            field=wagtail.fields.StreamField([('rich_text', wagtail.blocks.RichTextBlock(editor='custom'))], use_json_field=False),
+        ),
+        migrations.AlterField(
+            model_name='deadlystreampage',
+            name='body',
+            field=wagtail.fields.StreamField([('title', wagtail.test.testapp.models.DeadlyCharBlock())], use_json_field=False),
+        ),
+        migrations.AlterField(
+            model_name='defaultrichblockfieldpage',
+            name='body',
+            field=wagtail.fields.StreamField([('rich_text', wagtail.blocks.RichTextBlock())], use_json_field=False),
+        ),
+        migrations.AlterField(
+            model_name='defaultstreampage',
+            name='body',
+            field=wagtail.fields.StreamField([('text', wagtail.blocks.CharBlock()), ('rich_text', wagtail.blocks.RichTextBlock()), ('image', wagtail.images.blocks.ImageChooserBlock())], default='', use_json_field=False),
+        ),
+        migrations.AlterField(
+            model_name='inlinestreampagesection',
+            name='body',
+            field=wagtail.fields.StreamField([('text', wagtail.blocks.CharBlock()), ('rich_text', wagtail.blocks.RichTextBlock()), ('image', wagtail.images.blocks.ImageChooserBlock())], use_json_field=False),
+        ),
+        migrations.AlterField(
+            model_name='minmaxcountstreammodel',
+            name='body',
+            field=wagtail.fields.StreamField([('text', wagtail.blocks.CharBlock()), ('rich_text', wagtail.blocks.RichTextBlock()), ('image', wagtail.images.blocks.ImageChooserBlock())], use_json_field=False),
+        ),
+        migrations.AlterField(
+            model_name='streammodel',
+            name='body',
+            field=wagtail.fields.StreamField([('text', wagtail.blocks.CharBlock()), ('rich_text', wagtail.blocks.RichTextBlock()), ('image', wagtail.images.blocks.ImageChooserBlock())], use_json_field=False),
+        ),
+        migrations.AlterField(
+            model_name='streampage',
+            name='body',
+            field=wagtail.fields.StreamField([('text', wagtail.blocks.CharBlock()), ('rich_text', wagtail.blocks.RichTextBlock()), ('image', wagtail.test.testapp.models.ExtendedImageChooserBlock()), ('product', wagtail.blocks.StructBlock([('name', wagtail.blocks.CharBlock()), ('price', wagtail.blocks.CharBlock())])), ('raw_html', wagtail.blocks.RawHTMLBlock()), ('books', wagtail.blocks.StreamBlock([('title', wagtail.blocks.CharBlock()), ('author', wagtail.blocks.CharBlock())]))], use_json_field=False),
+        ),
+        migrations.AlterField(
+            model_name='tableblockstreampage',
+            name='table',
+            field=wagtail.fields.StreamField([('table', wagtail.contrib.table_block.blocks.TableBlock())], use_json_field=False),
+        ),
+    ]
diff --git a/wagtail/test/testapp/migrations/0063_jsonblockcountsstreammodel_and_more.py b/wagtail/test/testapp/migrations/0063_jsonblockcountsstreammodel_and_more.py
new file mode 100644
index 0000000000..1d98bebd9a
--- /dev/null
+++ b/wagtail/test/testapp/migrations/0063_jsonblockcountsstreammodel_and_more.py
@@ -0,0 +1,37 @@
+# Generated by Django 4.0.3 on 2022-03-18 06:37
+
+from django.db import migrations, models
+import wagtail.blocks
+import wagtail.fields
+import wagtail.images.blocks
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('tests', '0062_alter_addedstreamfieldwithemptylistdefaultpage_body_and_more'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='JSONBlockCountsStreamModel',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('body', wagtail.fields.StreamField([('text', wagtail.blocks.CharBlock()), ('rich_text', wagtail.blocks.RichTextBlock()), ('image', wagtail.images.blocks.ImageChooserBlock())], use_json_field=True)),
+            ],
+        ),
+        migrations.CreateModel(
+            name='JSONMinMaxCountStreamModel',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('body', wagtail.fields.StreamField([('text', wagtail.blocks.CharBlock()), ('rich_text', wagtail.blocks.RichTextBlock()), ('image', wagtail.images.blocks.ImageChooserBlock())], use_json_field=True)),
+            ],
+        ),
+        migrations.CreateModel(
+            name='JSONStreamModel',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('body', wagtail.fields.StreamField([('text', wagtail.blocks.CharBlock()), ('rich_text', wagtail.blocks.RichTextBlock()), ('image', wagtail.images.blocks.ImageChooserBlock())], use_json_field=True)),
+            ],
+        ),
+    ]
diff --git a/wagtail/test/testapp/models.py b/wagtail/test/testapp/models.py
index 77ce508fcf..37572b55ac 100644
--- a/wagtail/test/testapp/models.py
+++ b/wagtail/test/testapp/models.py
@@ -1103,7 +1103,19 @@ class StreamModel(models.Model):
             ("text", CharBlock()),
             ("rich_text", RichTextBlock()),
             ("image", ImageChooserBlock()),
-        ]
+        ],
+        use_json_field=False,
+    )
+
+
+class JSONStreamModel(models.Model):
+    body = StreamField(
+        [
+            ("text", CharBlock()),
+            ("rich_text", RichTextBlock()),
+            ("image", ImageChooserBlock()),
+        ],
+        use_json_field=True,
     )
 
 
@@ -1116,6 +1128,20 @@ class MinMaxCountStreamModel(models.Model):
         ],
         min_num=2,
         max_num=5,
+        use_json_field=False,
+    )
+
+
+class JSONMinMaxCountStreamModel(models.Model):
+    body = StreamField(
+        [
+            ("text", CharBlock()),
+            ("rich_text", RichTextBlock()),
+            ("image", ImageChooserBlock()),
+        ],
+        min_num=2,
+        max_num=5,
+        use_json_field=True,
     )
 
 
@@ -1131,6 +1157,23 @@ class BlockCountsStreamModel(models.Model):
             "rich_text": {"max_num": 1},
             "image": {"min_num": 1, "max_num": 1},
         },
+        use_json_field=False,
+    )
+
+
+class JSONBlockCountsStreamModel(models.Model):
+    body = StreamField(
+        [
+            ("text", CharBlock()),
+            ("rich_text", RichTextBlock()),
+            ("image", ImageChooserBlock()),
+        ],
+        block_counts={
+            "text": {"min_num": 1},
+            "rich_text": {"max_num": 1},
+            "image": {"min_num": 1, "max_num": 1},
+        },
+        use_json_field=True,
     )
 
 
@@ -1175,7 +1218,8 @@ class StreamPage(Page):
                     ]
                 ),
             ),
-        ]
+        ],
+        use_json_field=False,
     )
 
     api_fields = ("body",)
@@ -1196,6 +1240,7 @@ class DefaultStreamPage(Page):
             ("image", ImageChooserBlock()),
         ],
         default="",
+        use_json_field=False,
     )
 
     content_panels = [
@@ -1395,7 +1440,8 @@ class DefaultRichBlockFieldPage(Page):
     body = StreamField(
         [
             ("rich_text", RichTextBlock()),
-        ]
+        ],
+        use_json_field=False,
     )
 
     content_panels = Page.content_panels + [FieldPanel("body")]
@@ -1414,7 +1460,8 @@ class CustomRichBlockFieldPage(Page):
     body = StreamField(
         [
             ("rich_text", RichTextBlock(editor="custom")),
-        ]
+        ],
+        use_json_field=False,
     )
 
     content_panels = [
@@ -1459,7 +1506,8 @@ class InlineStreamPageSection(Orderable):
             ("text", CharBlock()),
             ("rich_text", RichTextBlock()),
             ("image", ImageChooserBlock()),
-        ]
+        ],
+        use_json_field=False,
     )
     panels = [FieldPanel("body")]
 
@@ -1472,7 +1520,7 @@ class InlineStreamPage(Page):
 
 
 class TableBlockStreamPage(Page):
-    table = StreamField([("table", TableBlock())])
+    table = StreamField([("table", TableBlock())], use_json_field=False)
 
     content_panels = [FieldPanel("table")]
 
@@ -1502,15 +1550,15 @@ class AlwaysShowInMenusPage(Page):
 
 # test for AddField migrations on StreamFields using various default values
 class AddedStreamFieldWithoutDefaultPage(Page):
-    body = StreamField([("title", CharBlock())])
+    body = StreamField([("title", CharBlock())], use_json_field=False)
 
 
 class AddedStreamFieldWithEmptyStringDefaultPage(Page):
-    body = StreamField([("title", CharBlock())], default="")
+    body = StreamField([("title", CharBlock())], default="", use_json_field=False)
 
 
 class AddedStreamFieldWithEmptyListDefaultPage(Page):
-    body = StreamField([("title", CharBlock())], default=[])
+    body = StreamField([("title", CharBlock())], default=[], use_json_field=False)
 
 
 class SecretPage(Page):
@@ -1639,7 +1687,8 @@ class DeadlyStreamPage(Page):
     body = StreamField(
         [
             ("title", DeadlyCharBlock()),
-        ]
+        ],
+        use_json_field=False,
     )
     content_panels = Page.content_panels + [
         FieldPanel("body"),
diff --git a/wagtail/tests/test_streamfield.py b/wagtail/tests/test_streamfield.py
index 91c81a08d3..1d44efc72c 100644
--- a/wagtail/tests/test_streamfield.py
+++ b/wagtail/tests/test_streamfield.py
@@ -1,10 +1,11 @@
 # -*- coding: utf-8 -*
 import json
+from unittest import skip
 
 from django.apps import apps
-from django.db import models
+from django.db import connection, models
 from django.template import Context, Template, engines
-from django.test import TestCase
+from django.test import TestCase, skipUnlessDBFeature
 from django.utils.safestring import SafeString
 
 from wagtail import blocks
@@ -15,17 +16,23 @@ from wagtail.images.tests.utils import get_test_image_file
 from wagtail.rich_text import RichText
 from wagtail.test.testapp.models import (
     BlockCountsStreamModel,
+    JSONBlockCountsStreamModel,
+    JSONMinMaxCountStreamModel,
+    JSONStreamModel,
     MinMaxCountStreamModel,
     StreamModel,
 )
+from wagtail.utils.deprecation import RemovedInWagtail219Warning
 
 
 class TestLazyStreamField(TestCase):
+    model = StreamModel
+
     def setUp(self):
         self.image = Image.objects.create(
             title="Test image", file=get_test_image_file()
         )
-        self.with_image = StreamModel.objects.create(
+        self.with_image = self.model.objects.create(
             body=json.dumps(
                 [
                     {"type": "image", "value": self.image.pk},
@@ -33,11 +40,10 @@ class TestLazyStreamField(TestCase):
                 ]
             )
         )
-        self.no_image = StreamModel.objects.create(
+        self.no_image = self.model.objects.create(
             body=json.dumps([{"type": "text", "value": "foo"}])
         )
-        self.nonjson_body = StreamModel.objects.create(body="<h1>hello world</h1>")
-        self.three_items = StreamModel.objects.create(
+        self.three_items = self.model.objects.create(
             body=json.dumps(
                 [
                     {"type": "text", "value": "foo"},
@@ -54,7 +60,7 @@ class TestLazyStreamField(TestCase):
         """
         with self.assertNumQueries(1):
             # Get the instance. The StreamField should *not* load the image yet
-            instance = StreamModel.objects.get(pk=self.with_image.pk)
+            instance = self.model.objects.get(pk=self.with_image.pk)
 
         with self.assertNumQueries(0):
             # Access the body. The StreamField should still not get the image.
@@ -71,7 +77,7 @@ class TestLazyStreamField(TestCase):
 
     def test_slice(self):
         with self.assertNumQueries(1):
-            instance = StreamModel.objects.get(pk=self.three_items.pk)
+            instance = self.model.objects.get(pk=self.three_items.pk)
 
         with self.assertNumQueries(1):
             # Access the image item from the stream. The image is fetched now
@@ -99,7 +105,7 @@ class TestLazyStreamField(TestCase):
         """
         with self.assertNumQueries(1):
             # Get the instance, nothing else
-            instance = StreamModel.objects.get(pk=self.no_image.pk)
+            instance = self.model.objects.get(pk=self.no_image.pk)
 
         with self.assertNumQueries(0):
             # Access the body. The StreamField has no images, so nothing should
@@ -113,7 +119,7 @@ class TestLazyStreamField(TestCase):
         queryset list
         """
         with self.assertNumQueries(1):
-            instances = StreamModel.objects.filter(
+            instances = self.model.objects.filter(
                 pk__in=[self.with_image.pk, self.no_image.pk]
             )
             instances_lookup = {instance.pk: instance for instance in instances}
@@ -133,7 +139,7 @@ class TestLazyStreamField(TestCase):
         image_1 = Image.objects.create(title="Test image 1", file=file_obj)
         image_3 = Image.objects.create(title="Test image 3", file=file_obj)
 
-        with_image = StreamModel.objects.create(
+        with_image = self.model.objects.create(
             body=json.dumps(
                 [
                     {"type": "image", "value": image_1.pk},
@@ -145,7 +151,7 @@ class TestLazyStreamField(TestCase):
         )
 
         with self.assertNumQueries(1):
-            instance = StreamModel.objects.get(pk=with_image.pk)
+            instance = self.model.objects.get(pk=with_image.pk)
 
         # Prefetch all image blocks
         with self.assertNumQueries(1):
@@ -167,7 +173,7 @@ class TestLazyStreamField(TestCase):
         blocks that have not been accessed.
         """
         with self.assertNumQueries(1):
-            instance = StreamModel.objects.get(pk=self.with_image.pk)
+            instance = self.model.objects.get(pk=self.with_image.pk)
 
         # Expect a single UPDATE to update the model, without any additional
         # SELECT related to the image block that has not been accessed.
@@ -175,7 +181,13 @@ class TestLazyStreamField(TestCase):
             instance.save()
 
 
+class TestJSONLazyStreamField(TestLazyStreamField):
+    model = JSONStreamModel
+
+
 class TestSystemCheck(TestCase):
+    use_json_field = False
+
     def tearDown(self):
         # unregister InvalidStreamModel from the overall model registry
         # so that it doesn't break tests elsewhere
@@ -192,7 +204,8 @@ class TestSystemCheck(TestCase):
                 [
                     ("heading", blocks.CharBlock()),
                     ("rich text", blocks.RichTextBlock()),
-                ]
+                ],
+                use_json_field=self.use_json_field,
             )
 
         errors = InvalidStreamModel.check()
@@ -202,27 +215,33 @@ class TestSystemCheck(TestCase):
         self.assertEqual(errors[0].obj, InvalidStreamModel._meta.get_field("body"))
 
 
+class TestJSONSystemCheck(TestSystemCheck):
+    use_json_field = True
+
+
 class TestStreamValueAccess(TestCase):
+    model = StreamModel
+
     def setUp(self):
-        self.json_body = StreamModel.objects.create(
+        self.json_body = self.model.objects.create(
             body=json.dumps([{"type": "text", "value": "foo"}])
         )
-        self.nonjson_body = StreamModel.objects.create(body="<h1>hello world</h1>")
 
     def test_can_read_non_json_content(self):
         """StreamField columns should handle non-JSON database content gracefully"""
-        self.assertIsInstance(self.nonjson_body.body, StreamValue)
+        nonjson_body = self.model.objects.create(body="<h1>hello world</h1>")
+        self.assertIsInstance(nonjson_body.body, StreamValue)
         # the main list-like content of the StreamValue should be blank
-        self.assertFalse(self.nonjson_body.body)
+        self.assertFalse(nonjson_body.body)
         # the unparsed text content should be available in raw_text
-        self.assertEqual(self.nonjson_body.body.raw_text, "<h1>hello world</h1>")
+        self.assertEqual(nonjson_body.body.raw_text, "<h1>hello world</h1>")
 
     def test_can_assign_as_list(self):
         self.json_body.body = [("rich_text", RichText("<h2>hello world</h2>"))]
         self.json_body.save()
 
         # the body should now be a stream consisting of a single rich_text block
-        fetched_body = StreamModel.objects.get(id=self.json_body.id).body
+        fetched_body = self.model.objects.get(id=self.json_body.id).body
         self.assertIsInstance(fetched_body, StreamValue)
         self.assertEqual(len(fetched_body), 1)
         self.assertIsInstance(fetched_body[0].value, RichText)
@@ -232,7 +251,7 @@ class TestStreamValueAccess(TestCase):
         self.json_body.body.append(("text", "bar"))
         self.json_body.save()
 
-        fetched_body = StreamModel.objects.get(id=self.json_body.id).body
+        fetched_body = self.model.objects.get(id=self.json_body.id).body
         self.assertIsInstance(fetched_body, StreamValue)
         self.assertEqual(len(fetched_body), 2)
         self.assertEqual(fetched_body[0].block_type, "text")
@@ -241,13 +260,23 @@ class TestStreamValueAccess(TestCase):
         self.assertEqual(fetched_body[1].value, "bar")
 
 
+class TestJSONStreamValueAccess(TestStreamValueAccess):
+    model = JSONStreamModel
+
+    @skip("JSONField-based StreamField does not support storing non-json content.")
+    def test_can_read_non_json_content(self):
+        pass
+
+
 class TestStreamFieldRenderingBase(TestCase):
+    model = StreamModel
+
     def setUp(self):
         self.image = Image.objects.create(
             title="Test image", file=get_test_image_file()
         )
 
-        self.instance = StreamModel.objects.create(
+        self.instance = self.model.objects.create(
             body=json.dumps(
                 [
                     {"type": "rich_text", "value": "<p>Rich text</p>"},
@@ -281,6 +310,10 @@ class TestStreamFieldRendering(TestStreamFieldRenderingBase):
         self.assertIsInstance(rendered, SafeString)
 
 
+class TestJSONStreamFieldRendering(TestStreamFieldRendering):
+    model = JSONStreamModel
+
+
 class TestStreamFieldDjangoRendering(TestStreamFieldRenderingBase):
     def render(self, string, context):
         return Template(string).render(Context(context))
@@ -290,6 +323,10 @@ class TestStreamFieldDjangoRendering(TestStreamFieldRenderingBase):
         self.assertHTMLEqual(rendered, self.expected)
 
 
+class TestJSONStreamFieldDjangoRendering(TestStreamFieldDjangoRendering):
+    model = JSONStreamModel
+
+
 class TestStreamFieldJinjaRendering(TestStreamFieldRenderingBase):
     def setUp(self):
         super().setUp()
@@ -303,10 +340,20 @@ class TestStreamFieldJinjaRendering(TestStreamFieldRenderingBase):
         self.assertHTMLEqual(rendered, self.expected)
 
 
+class TestJSONStreamFieldJinjaRendering(TestStreamFieldJinjaRendering):
+    model = JSONStreamModel
+
+
 class TestRequiredStreamField(TestCase):
+    use_json_field = False
+
     def test_non_blank_field_is_required(self):
         # passing a block list
-        field = StreamField([("paragraph", blocks.CharBlock())], blank=False)
+        field = StreamField(
+            [("paragraph", blocks.CharBlock())],
+            blank=False,
+            use_json_field=self.use_json_field,
+        )
         self.assertTrue(field.stream_block.required)
         with self.assertRaises(StreamBlockValidationError):
             field.stream_block.clean([])
@@ -318,25 +365,35 @@ class TestRequiredStreamField(TestCase):
                 required = False
 
         # passing a block instance
-        field = StreamField(MyStreamBlock(), blank=False)
+        field = StreamField(
+            MyStreamBlock(), blank=False, use_json_field=self.use_json_field
+        )
         self.assertTrue(field.stream_block.required)
         with self.assertRaises(StreamBlockValidationError):
             field.stream_block.clean([])
 
-        field = StreamField(MyStreamBlock(required=False), blank=False)
+        field = StreamField(
+            MyStreamBlock(required=False),
+            blank=False,
+            use_json_field=self.use_json_field,
+        )
         self.assertTrue(field.stream_block.required)
         with self.assertRaises(StreamBlockValidationError):
             field.stream_block.clean([])
 
         # passing a block class
-        field = StreamField(MyStreamBlock, blank=False)
+        field = StreamField(
+            MyStreamBlock, blank=False, use_json_field=self.use_json_field
+        )
         self.assertTrue(field.stream_block.required)
         with self.assertRaises(StreamBlockValidationError):
             field.stream_block.clean([])
 
     def test_blank_false_is_implied_by_default(self):
         # passing a block list
-        field = StreamField([("paragraph", blocks.CharBlock())])
+        field = StreamField(
+            [("paragraph", blocks.CharBlock())], use_json_field=self.use_json_field
+        )
         self.assertTrue(field.stream_block.required)
         with self.assertRaises(StreamBlockValidationError):
             field.stream_block.clean([])
@@ -348,25 +405,31 @@ class TestRequiredStreamField(TestCase):
                 required = False
 
         # passing a block instance
-        field = StreamField(MyStreamBlock())
+        field = StreamField(MyStreamBlock(), use_json_field=self.use_json_field)
         self.assertTrue(field.stream_block.required)
         with self.assertRaises(StreamBlockValidationError):
             field.stream_block.clean([])
 
-        field = StreamField(MyStreamBlock(required=False))
+        field = StreamField(
+            MyStreamBlock(required=False), use_json_field=self.use_json_field
+        )
         self.assertTrue(field.stream_block.required)
         with self.assertRaises(StreamBlockValidationError):
             field.stream_block.clean([])
 
         # passing a block class
-        field = StreamField(MyStreamBlock)
+        field = StreamField(MyStreamBlock, use_json_field=self.use_json_field)
         self.assertTrue(field.stream_block.required)
         with self.assertRaises(StreamBlockValidationError):
             field.stream_block.clean([])
 
     def test_blank_field_is_not_required(self):
         # passing a block list
-        field = StreamField([("paragraph", blocks.CharBlock())], blank=True)
+        field = StreamField(
+            [("paragraph", blocks.CharBlock())],
+            blank=True,
+            use_json_field=self.use_json_field,
+        )
         self.assertFalse(field.stream_block.required)
         field.stream_block.clean([])  # no validation error on empty stream
 
@@ -377,21 +440,35 @@ class TestRequiredStreamField(TestCase):
                 required = True
 
         # passing a block instance
-        field = StreamField(MyStreamBlock(), blank=True)
+        field = StreamField(
+            MyStreamBlock(), blank=True, use_json_field=self.use_json_field
+        )
         self.assertFalse(field.stream_block.required)
         field.stream_block.clean([])  # no validation error on empty stream
 
-        field = StreamField(MyStreamBlock(required=True), blank=True)
+        field = StreamField(
+            MyStreamBlock(required=True), blank=True, use_json_field=self.use_json_field
+        )
         self.assertFalse(field.stream_block.required)
         field.stream_block.clean([])  # no validation error on empty stream
 
         # passing a block class
-        field = StreamField(MyStreamBlock, blank=True)
+        field = StreamField(
+            MyStreamBlock, blank=True, use_json_field=self.use_json_field
+        )
         self.assertFalse(field.stream_block.required)
         field.stream_block.clean([])  # no validation error on empty stream
 
 
+class TestJSONRequiredStreamField(TestRequiredStreamField):
+    use_json_field = True
+
+
 class TestStreamFieldCountValidation(TestCase):
+    min_max_count_model = MinMaxCountStreamModel
+    block_counts_model = BlockCountsStreamModel
+    use_json_field = False
+
     def setUp(self):
         self.image = Image.objects.create(
             title="Test image", file=get_test_image_file()
@@ -402,14 +479,14 @@ class TestStreamFieldCountValidation(TestCase):
         self.text_body = {"type": "text", "value": "Hello, World!"}
 
     def test_minmax_pass_to_block(self):
-        instance = MinMaxCountStreamModel.objects.create(body=json.dumps([]))
+        instance = self.min_max_count_model.objects.create(body=json.dumps([]))
         internal_block = instance.body.stream_block
 
         self.assertEqual(internal_block.meta.min_num, 2)
         self.assertEqual(internal_block.meta.max_num, 5)
 
     def test_counts_pass_to_block(self):
-        instance = BlockCountsStreamModel.objects.create(body=json.dumps([]))
+        instance = self.block_counts_model.objects.create(body=json.dumps([]))
         block_counts = instance.body.stream_block.meta.block_counts
 
         self.assertEqual(block_counts.get("text"), {"min_num": 1})
@@ -419,7 +496,7 @@ class TestStreamFieldCountValidation(TestCase):
     def test_minimum_count(self):
         # Single block should fail validation
         body = [self.rich_text_body]
-        instance = MinMaxCountStreamModel.objects.create(body=json.dumps(body))
+        instance = self.min_max_count_model.objects.create(body=json.dumps(body))
         with self.assertRaises(StreamBlockValidationError) as catcher:
             instance.body.stream_block.clean(instance.body)
         self.assertEqual(
@@ -428,18 +505,18 @@ class TestStreamFieldCountValidation(TestCase):
 
         # 2 blocks okay
         body = [self.rich_text_body, self.text_body]
-        instance = MinMaxCountStreamModel.objects.create(body=json.dumps(body))
+        instance = self.min_max_count_model.objects.create(body=json.dumps(body))
         self.assertTrue(instance.body.stream_block.clean(instance.body))
 
     def test_maximum_count(self):
         # 5 blocks okay
         body = [self.rich_text_body] * 5
-        instance = MinMaxCountStreamModel.objects.create(body=json.dumps(body))
+        instance = self.min_max_count_model.objects.create(body=json.dumps(body))
         self.assertTrue(instance.body.stream_block.clean(instance.body))
 
         # 6 blocks should fail validation
         body = [self.rich_text_body, self.text_body] * 3
-        instance = MinMaxCountStreamModel.objects.create(body=json.dumps(body))
+        instance = self.min_max_count_model.objects.create(body=json.dumps(body))
         with self.assertRaises(StreamBlockValidationError) as catcher:
             instance.body.stream_block.clean(instance.body)
         self.assertEqual(
@@ -447,10 +524,10 @@ class TestStreamFieldCountValidation(TestCase):
         )
 
     def test_block_counts_minimums(self):
-        instance = BlockCountsStreamModel.objects.create(body=json.dumps([]))
+        instance = self.block_counts_model.objects.create(body=json.dumps([]))
 
         # Zero blocks should fail validation (requires one text, one image)
-        instance = BlockCountsStreamModel.objects.create(body=json.dumps([]))
+        instance = self.block_counts_model.objects.create(body=json.dumps([]))
         with self.assertRaises(StreamBlockValidationError) as catcher:
             instance.body.stream_block.clean(instance.body)
         errors = list(catcher.exception.params["__all__"])
@@ -461,7 +538,7 @@ class TestStreamFieldCountValidation(TestCase):
 
         # One plain text should fail validation
         body = [self.text_body]
-        instance = BlockCountsStreamModel.objects.create(body=json.dumps(body))
+        instance = self.block_counts_model.objects.create(body=json.dumps(body))
         with self.assertRaises(StreamBlockValidationError) as catcher:
             instance.body.stream_block.clean(instance.body)
         self.assertEqual(
@@ -471,15 +548,15 @@ class TestStreamFieldCountValidation(TestCase):
 
         # One text, one image should be okay
         body = [self.text_body, self.image_body]
-        instance = BlockCountsStreamModel.objects.create(body=json.dumps(body))
+        instance = self.block_counts_model.objects.create(body=json.dumps(body))
         self.assertTrue(instance.body.stream_block.clean(instance.body))
 
     def test_block_counts_maximums(self):
-        instance = BlockCountsStreamModel.objects.create(body=json.dumps([]))
+        instance = self.block_counts_model.objects.create(body=json.dumps([]))
 
         # Base is one text, one image
         body = [self.text_body, self.image_body]
-        instance = BlockCountsStreamModel.objects.create(body=json.dumps(body))
+        instance = self.block_counts_model.objects.create(body=json.dumps(body))
         self.assertTrue(instance.body.stream_block.clean(instance.body))
 
         # Two rich text should error
@@ -489,14 +566,14 @@ class TestStreamFieldCountValidation(TestCase):
             self.rich_text_body,
             self.rich_text_body,
         ]
-        instance = BlockCountsStreamModel.objects.create(body=json.dumps(body))
+        instance = self.block_counts_model.objects.create(body=json.dumps(body))
 
         with self.assertRaises(StreamBlockValidationError):
             instance.body.stream_block.clean(instance.body)
 
         # Two images should error
         body = [self.text_body, self.image_body, self.image_body]
-        instance = BlockCountsStreamModel.objects.create(body=json.dumps(body))
+        instance = self.block_counts_model.objects.create(body=json.dumps(body))
 
         with self.assertRaises(StreamBlockValidationError) as catcher:
             instance.body.stream_block.clean(instance.body)
@@ -507,7 +584,7 @@ class TestStreamFieldCountValidation(TestCase):
 
         # One text, one rich, one image should be okay
         body = [self.text_body, self.image_body, self.rich_text_body]
-        instance = BlockCountsStreamModel.objects.create(body=json.dumps(body))
+        instance = self.block_counts_model.objects.create(body=json.dumps(body))
         self.assertTrue(instance.body.stream_block.clean(instance.body))
 
     def test_streamfield_count_argument_precedence(self):
@@ -521,7 +598,7 @@ class TestStreamFieldCountValidation(TestCase):
                 block_counts = {"heading": {"max_num": 1}}
 
         # args being picked up from the class definition
-        field = StreamField(TestStreamBlock)
+        field = StreamField(TestStreamBlock, use_json_field=self.use_json_field)
         self.assertEqual(field.stream_block.meta.min_num, 2)
         self.assertEqual(field.stream_block.meta.max_num, 5)
         self.assertEqual(field.stream_block.meta.block_counts["heading"]["max_num"], 1)
@@ -532,6 +609,7 @@ class TestStreamFieldCountValidation(TestCase):
             min_num=3,
             max_num=6,
             block_counts={"heading": {"max_num": 2}},
+            use_json_field=self.use_json_field,
         )
         self.assertEqual(field.stream_block.meta.min_num, 3)
         self.assertEqual(field.stream_block.meta.max_num, 6)
@@ -539,8 +617,65 @@ class TestStreamFieldCountValidation(TestCase):
 
         # passing None from StreamField should cancel limits set at the block level
         field = StreamField(
-            TestStreamBlock, min_num=None, max_num=None, block_counts=None
+            TestStreamBlock,
+            min_num=None,
+            max_num=None,
+            block_counts=None,
+            use_json_field=self.use_json_field,
         )
         self.assertIsNone(field.stream_block.meta.min_num)
         self.assertIsNone(field.stream_block.meta.max_num)
         self.assertIsNone(field.stream_block.meta.block_counts)
+
+
+class TestJSONStreamFieldCountValidation(TestStreamFieldCountValidation):
+    min_max_count_model = JSONMinMaxCountStreamModel
+    block_counts_model = JSONBlockCountsStreamModel
+    use_json_field = True
+
+
+class TestJSONStreamField(TestCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.instance = JSONStreamModel.objects.create(
+            body=[{"type": "text", "value": "foo"}],
+        )
+
+    def test_use_json_field_warning(self):
+        message = "StreamField must explicitly set use_json_field argument to True/False instead of None."
+        with self.assertWarnsMessage(RemovedInWagtail219Warning, message):
+            StreamField([("paragraph", blocks.CharBlock())])
+
+    def test_internal_type(self):
+        text = StreamField([("paragraph", blocks.CharBlock())], use_json_field=False)
+        json = StreamField([("paragraph", blocks.CharBlock())], use_json_field=True)
+
+        self.assertEqual(text.get_internal_type(), "TextField")
+        self.assertEqual(json.get_internal_type(), "JSONField")
+
+    def test_json_body_equals_to_text_body(self):
+        instance_text = StreamModel.objects.create(
+            body=json.dumps([{"type": "text", "value": "foo"}]),
+        )
+        self.assertEqual(
+            instance_text.body.render_as_block(), self.instance.body.render_as_block()
+        )
+
+    def test_json_body_create_preserialised_value(self):
+        instance_preserialised = JSONStreamModel.objects.create(
+            body=json.dumps([{"type": "text", "value": "foo"}]),
+        )
+        self.assertEqual(
+            instance_preserialised.body.render_as_block(),
+            self.instance.body.render_as_block(),
+        )
+
+    @skipUnlessDBFeature("supports_json_field_contains")
+    def test_json_contains_lookup(self):
+        value = {"value": "foo"}
+        if connection.features.json_key_contains_list_matching_requires_list:
+            value = [value]
+        instance = JSONStreamModel.objects.filter(body__contains=value).first()
+        self.assertIsNotNone(instance)
+        self.assertEqual(instance.id, self.instance.id)