diff --git a/datasette/views/base.py b/datasette/views/base.py
index 50b82aa7..be5023dc 100644
--- a/datasette/views/base.py
+++ b/datasette/views/base.py
@@ -199,7 +199,40 @@ class BaseView(RenderMixin):
 
         return await self.view_get(request, name, hash, **kwargs)
 
+    async def as_csv_stream(self, request, name, hash, **kwargs):
+        assert not request.args.get("_next") # TODO: real error
+        kwargs['_size'] = 'max'
+
+        async def stream_fn(r):
+            first = True
+            next = None
+            writer = csv.writer(r)
+            while first or next:
+                if next:
+                    kwargs['_next'] = next
+                data, extra_template_data, templates = await self.data(
+                    request, name, hash, **kwargs
+                )
+                if first:
+                    writer.writerow(data["columns"])
+                    first = False
+                next = data["next"]
+                for row in data["rows"]:
+                    writer.writerow(row)
+
+        return response.stream(
+            stream_fn,
+            # headers={
+            #     "Content-Disposition": 'attachment; filename="{}.csv"'.format(
+            #         name
+            #     )
+            # },
+            content_type="text/plain; charset=utf-8"
+        )
+
     async def as_csv(self, request, name, hash, **kwargs):
+        if request.args.get("_stream"):
+            return await self.as_csv_stream(request, name, hash, **kwargs)
         try:
             response_or_template_contexts = await self.data(
                 request, name, hash, **kwargs
@@ -226,12 +259,13 @@ class BaseView(RenderMixin):
 
         return response.stream(
             stream_fn,
-            headers={
-                "Content-Disposition": 'attachment; filename="{}.csv"'.format(
-                    name
-                )
-            },
-            content_type="text/csv; charset=utf-8"
+            # headers={
+            #     "Content-Disposition": 'attachment; filename="{}.csv"'.format(
+            #         name
+            #     )
+            # },
+            content_type="text/plain; charset=utf-8"
+            #content_type="text/csv; charset=utf-8"
         )
 
     async def view_get(self, request, name, hash, **kwargs):
diff --git a/datasette/views/table.py b/datasette/views/table.py
index ab66b880..21d1ca76 100644
--- a/datasette/views/table.py
+++ b/datasette/views/table.py
@@ -232,7 +232,7 @@ class RowTableShared(BaseView):
 
 class TableView(RowTableShared):
 
-    async def data(self, request, name, hash, table):
+    async def data(self, request, name, hash, table, _next=None, _size=None):
         table = urllib.parse.unquote_plus(table)
         canned_query = self.ds.get_canned_query(name, table)
         if canned_query is not None:
@@ -411,7 +411,7 @@ class TableView(RowTableShared):
         )
         count_sql = "select count(*) {}".format(from_sql)
 
-        _next = special_args.get("_next")
+        _next = _next or special_args.get("_next")
         offset = ""
         if _next:
             if is_view:
@@ -498,7 +498,7 @@ class TableView(RowTableShared):
 
         extra_args = {}
         # Handle ?_page_size=500
-        page_size = request.raw_args.get("_size")
+        page_size = _size or request.raw_args.get("_size")
         if page_size:
             if page_size == "max":
                 page_size = self.max_returned_rows
@@ -545,6 +545,8 @@ class TableView(RowTableShared):
             pass
         facet_results = {}
         for column in facets:
+            if _next:
+                continue
             facet_sql = """
                 select {col} as value, count(*) as count
                 {from_sql} {and_or_where} {col} is not null
@@ -649,6 +651,8 @@ class TableView(RowTableShared):
             for facet_column in columns:
                 if facet_column in facets:
                     continue
+                if _next:
+                    continue
                 suggested_facet_sql = '''
                     select distinct {column} {from_sql}
                     {and_or_where} {column} is not null