diff --git a/datasette/app.py b/datasette/app.py
index d17f2096..e09e4fa5 100644
--- a/datasette/app.py
+++ b/datasette/app.py
@@ -619,9 +619,11 @@ class TableView(RowTableShared):
         if use_rowid:
             select = 'rowid, *'
             order_by = 'rowid'
+            order_by_pks = 'rowid'
         else:
             select = '*'
-            order_by = ', '.join(pks)
+            order_by_pks = ', '.join([escape_sqlite(pk) for pk in pks])
+            order_by = order_by_pks
 
         if is_view:
             order_by = ''
@@ -739,6 +741,9 @@ class TableView(RowTableShared):
                 # If a sort order is applied, the first of these is the sort value
                 if sort or sort_desc:
                     sort_value = components[0]
+                    # Special case for if non-urlencoded first token was $null
+                    if _next.split(',')[0] == '$null':
+                        sort_value = None
                     components = components[1:]
 
                 # Figure out the SQL for next-based-on-primary-key first
@@ -760,15 +765,38 @@ class TableView(RowTableShared):
 
                 # Now add the sort SQL, which may incorporate next_by_pk_clauses
                 if sort or sort_desc:
-                    where_clauses.append(
-                        '({column} {op} :p{p} or ({column} = :p{p} and {next_clauses}))'.format(
-                            column=escape_sqlite(sort or sort_desc),
-                            op='>' if sort else '<',
-                            p=len(params),
-                            next_clauses=' and '.join(next_by_pk_clauses),
+                    if sort_value is None:
+                        if sort_desc:
+                            # Just items where column is null ordered by pk
+                            where_clauses.append(
+                                '({column} is null and {next_clauses})'.format(
+                                    column=escape_sqlite(sort_desc),
+                                    next_clauses=' and '.join(next_by_pk_clauses),
+                                )
+                            )
+                        else:
+                            where_clauses.append(
+                                '({column} is not null or ({column} is null and {next_clauses}))'.format(
+                                    column=escape_sqlite(sort),
+                                    next_clauses=' and '.join(next_by_pk_clauses),
+                                )
+                            )
+                    else:
+                        where_clauses.append(
+                            '({column} {op} :p{p}{extra_desc_only} or ({column} = :p{p} and {next_clauses}))'.format(
+                                column=escape_sqlite(sort or sort_desc),
+                                op='>' if sort else '<',
+                                p=len(params),
+                                extra_desc_only='' if sort else ' or {column2} is null'.format(
+                                    column2=escape_sqlite(sort or sort_desc),
+                                ),
+                                next_clauses=' and '.join(next_by_pk_clauses),
+                            )
                         )
+                        params['p{}'.format(len(params))] = sort_value
+                    order_by = '{}, {}'.format(
+                        order_by, order_by_pks
                     )
-                    params['p{}'.format(len(params))] = sort_value
                 else:
                     where_clauses.extend(next_by_pk_clauses)
 
@@ -823,10 +851,12 @@ class TableView(RowTableShared):
                 next_value = path_from_row_pks(rows[-2], pks, use_rowid)
             # If there's a sort or sort_desc, add that value as a prefix
             if (sort or sort_desc) and not is_view:
-                prefix = str(rows[-2][sort or sort_desc])
-                next_value = '{},{}'.format(
-                    urllib.parse.quote_plus(prefix), next_value
-                )
+                prefix = rows[-2][sort or sort_desc]
+                if prefix is None:
+                    prefix = '$null'
+                else:
+                    prefix = urllib.parse.quote_plus(str(prefix))
+                next_value = '{},{}'.format(prefix, next_value)
                 added_args = {
                     '_next': next_value,
                 }
diff --git a/tests/fixtures.py b/tests/fixtures.py
index 38df8b00..29075d7f 100644
--- a/tests/fixtures.py
+++ b/tests/fixtures.py
@@ -55,6 +55,7 @@ def generate_sortable_rows(num):
             'sortable_with_nulls_2': rand.choice([
                 None, rand.random(), rand.random()
             ]),
+            'text': rand.choice(['$null', '$blah']),
         }
 
 
@@ -78,6 +79,7 @@ METADATA = {
                         'sortable',
                         'sortable_with_nulls',
                         'sortable_with_nulls_2',
+                        'text',
                     ]
                 },
                 'no_primary_key': {
@@ -153,6 +155,7 @@ CREATE TABLE sortable (
   sortable integer,
   sortable_with_nulls real,
   sortable_with_nulls_2 real,
+  text text,
   PRIMARY KEY (pk1, pk2)
 );
 
@@ -235,7 +238,7 @@ CREATE VIEW simple_view AS
 ]) + '\n'.join([
     '''INSERT INTO sortable VALUES (
         "{pk1}", "{pk2}", "{content}", {sortable},
-        {sortable_with_nulls}, {sortable_with_nulls_2});
+        {sortable_with_nulls}, {sortable_with_nulls_2}, "{text}");
     '''.format(
         **row
     ).replace('None', 'null') for row in generate_sortable_rows(201)
diff --git a/tests/test_api.py b/tests/test_api.py
index 21fc34bf..7d9548b0 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -156,7 +156,7 @@ def test_database_page(app_client):
     }, {
         'columns': [
             'pk1', 'pk2', 'content', 'sortable', 'sortable_with_nulls',
-            'sortable_with_nulls_2'
+            'sortable_with_nulls_2', 'text',
         ],
         'name': 'sortable',
         'count': 201,
@@ -426,6 +426,25 @@ def test_paginate_compound_keys_with_extra_filters(app_client):
 @pytest.mark.parametrize('query_string,sort_key,human_description_en', [
     ('_sort=sortable', lambda row: row['sortable'], 'sorted by sortable'),
     ('_sort_desc=sortable', lambda row: -row['sortable'], 'sorted by sortable descending'),
+    (
+        '_sort=sortable_with_nulls',
+        lambda row: (
+            1 if row['sortable_with_nulls'] is not None else 0,
+            row['sortable_with_nulls']
+        ),
+        'sorted by sortable_with_nulls'
+    ),
+    (
+        '_sort_desc=sortable_with_nulls',
+        lambda row: (
+            1 if row['sortable_with_nulls'] is None else 0,
+            -row['sortable_with_nulls'] if row['sortable_with_nulls'] is not None else 0,
+            row['content']
+        ),
+        'sorted by sortable_with_nulls descending'
+    ),
+    # text column contains '$null' - ensure it doesn't confuse pagination:
+    ('_sort=text', lambda row: row['text'], 'sorted by text'),
 ])
 def test_sortable(app_client, query_string, sort_key, human_description_en):
     path = '/test_tables/sortable.json?_shape=objects&{}'.format(query_string)