diff --git a/datasette/renderer.py b/datasette/renderer.py
index 66ac169b..45089498 100644
--- a/datasette/renderer.py
+++ b/datasette/renderer.py
@@ -29,6 +29,7 @@ def convert_specific_columns_to_json(rows, columns, json_cols):
 def json_renderer(args, data, view_name):
     """Render a response as JSON"""
     status_code = 200
+
     # Handle the _json= parameter which may modify data["rows"]
     json_cols = []
     if "_json" in args:
@@ -44,6 +45,9 @@ def json_renderer(args, data, view_name):
 
     # Deal with the _shape option
     shape = args.get("_shape", "arrays")
+    # if there's an error, ignore the shape entirely
+    if data.get("error"):
+        shape = "arrays"
 
     next_url = data.get("next_url")
 
diff --git a/datasette/templates/query.html b/datasette/templates/query.html
index 9b3fff25..633e53b4 100644
--- a/datasette/templates/query.html
+++ b/datasette/templates/query.html
@@ -33,7 +33,10 @@
 {% block description_source_license %}{% include "_description_source_license.html" %}{% endblock %}
 
 <form class="sql" action="{{ urls.database(database) }}{% if canned_query %}/{{ canned_query }}{% endif %}" method="{% if canned_write %}post{% else %}get{% endif %}">
-    <h3>Custom SQL query{% if display_rows %} returning {% if truncated %}more than {% endif %}{{ "{:,}".format(display_rows|length) }} row{% if display_rows|length == 1 %}{% else %}s{% endif %}{% endif %} <span class="show-hide-sql">{% if hide_sql %}(<a href="{{ path_with_removed_args(request, {'_hide_sql': '1'}) }}">show</a>){% else %}(<a href="{{ path_with_added_args(request, {'_hide_sql': '1'}) }}">hide</a>){% endif %}</span></h3>
+    <h3>Custom SQL query{% if display_rows %} returning {% if truncated %}more than {% endif %}{{ "{:,}".format(display_rows|length) }} row{% if display_rows|length == 1 %}{% else %}s{% endif %}{% endif %}{% if not query_error %} <span class="show-hide-sql">{% if hide_sql %}(<a href="{{ path_with_removed_args(request, {'_hide_sql': '1'}) }}">show</a>){% else %}(<a href="{{ path_with_added_args(request, {'_hide_sql': '1'}) }}">hide</a>){% endif %}</span>{% endif %}</h3>
+    {% if query_error %}
+        <p class="message-error">{{ query_error }}</p>
+    {% endif %}
     {% if not hide_sql %}
         {% if editable and allow_execute_sql %}
             <p><textarea id="sql-editor" name="sql">{% if query and query.sql %}{{ query.sql }}{% else %}select * from {{ tables[0].name|escape_sqlite }}{% endif %}</textarea></p>
diff --git a/datasette/views/base.py b/datasette/views/base.py
index e2583034..94f54787 100644
--- a/datasette/views/base.py
+++ b/datasette/views/base.py
@@ -294,6 +294,8 @@ class DataView(BaseView):
             )
             if isinstance(response_or_template_contexts, Response):
                 return response_or_template_contexts
+            elif len(response_or_template_contexts) == 4:
+                data, _, _, _ = response_or_template_contexts
             else:
                 data, _, _ = response_or_template_contexts
         except (sqlite3.OperationalError, InvalidSql) as e:
@@ -467,7 +469,7 @@ class DataView(BaseView):
 
         extra_template_data = {}
         start = time.perf_counter()
-        status_code = 200
+        status_code = None
         templates = []
         try:
             response_or_template_contexts = await self.data(
@@ -475,7 +477,14 @@ class DataView(BaseView):
             )
             if isinstance(response_or_template_contexts, Response):
                 return response_or_template_contexts
-
+            # If it has four items, it includes an HTTP status code
+            if len(response_or_template_contexts) == 4:
+                (
+                    data,
+                    extra_template_data,
+                    templates,
+                    status_code,
+                ) = response_or_template_contexts
             else:
                 data, extra_template_data, templates = response_or_template_contexts
         except QueryInterrupted:
@@ -542,12 +551,15 @@ class DataView(BaseView):
             if isinstance(result, dict):
                 r = Response(
                     body=result.get("body"),
-                    status=result.get("status_code", 200),
+                    status=result.get("status_code", status_code or 200),
                     content_type=result.get("content_type", "text/plain"),
                     headers=result.get("headers"),
                 )
             elif isinstance(result, Response):
                 r = result
+                if status_code is not None:
+                    # Over-ride the status code
+                    r.status = status_code
             else:
                 assert False, f"{result} should be dict or Response"
         else:
@@ -607,7 +619,8 @@ class DataView(BaseView):
             if "metadata" not in context:
                 context["metadata"] = self.ds.metadata
             r = await self.render(templates, request=request, context=context)
-            r.status = status_code
+            if status_code is not None:
+                r.status = status_code
 
         ttl = request.args.get("_ttl", None)
         if ttl is None or not ttl.isdigit():
diff --git a/datasette/views/database.py b/datasette/views/database.py
index 96b2ca91..58168ed7 100644
--- a/datasette/views/database.py
+++ b/datasette/views/database.py
@@ -14,6 +14,7 @@ from datasette.utils import (
     path_with_added_args,
     path_with_format,
     path_with_removed_args,
+    sqlite3,
     InvalidSql,
 )
 from datasette.utils.asgi import AsgiFileDownload, Response, Forbidden
@@ -239,6 +240,8 @@ class QueryView(DataView):
 
         templates = [f"query-{to_css_class(database)}.html", "query.html"]
 
+        query_error = None
+
         # Execute query - as write or as read
         if write:
             if request.method == "POST":
@@ -320,10 +323,15 @@ class QueryView(DataView):
                 params_for_query = MagicParameters(params, request, self.ds)
             else:
                 params_for_query = params
-            results = await self.ds.execute(
-                database, sql, params_for_query, truncate=True, **extra_args
-            )
-            columns = [r[0] for r in results.description]
+            try:
+                results = await self.ds.execute(
+                    database, sql, params_for_query, truncate=True, **extra_args
+                )
+                columns = [r[0] for r in results.description]
+            except sqlite3.DatabaseError as e:
+                query_error = e
+                results = None
+                columns = []
 
         if canned_query:
             templates.insert(
@@ -337,7 +345,7 @@ class QueryView(DataView):
 
         async def extra_template():
             display_rows = []
-            for row in results.rows:
+            for row in results.rows if results else []:
                 display_row = []
                 for column, value in zip(results.columns, row):
                     display_value = value
@@ -423,17 +431,20 @@ class QueryView(DataView):
 
         return (
             {
+                "ok": not query_error,
                 "database": database,
                 "query_name": canned_query,
-                "rows": results.rows,
-                "truncated": results.truncated,
+                "rows": results.rows if results else [],
+                "truncated": results.truncated if results else False,
                 "columns": columns,
                 "query": {"sql": sql, "params": params},
+                "error": str(query_error) if query_error else None,
                 "private": private,
                 "allow_execute_sql": allow_execute_sql,
             },
             extra_template,
             templates,
+            400 if query_error else 200,
         )
 
 
diff --git a/tests/test_canned_queries.py b/tests/test_canned_queries.py
index 65f23cc7..4186a97c 100644
--- a/tests/test_canned_queries.py
+++ b/tests/test_canned_queries.py
@@ -352,5 +352,5 @@ def test_magic_parameters_cannot_be_used_in_arbitrary_queries(magic_parameters_c
     response = magic_parameters_client.get(
         "/data.json?sql=select+:_header_host&_shape=array"
     )
-    assert 500 == response.status
+    assert 400 == response.status
     assert "You did not supply a value for binding 1." == response.json["error"]