Page datastore records past HDX's row cap, warn on truncated SQL results, and document TDE

documentation/index.md

@@ -976,9 +976,13 @@ using the getter:
 
 ### Datastore
 
-A resource's data can be loaded into the HDX datastore and queried. HDX
-requires an API key for all datastore actions and dumps, including reads of
-public resources. To create, update or delete the datastore for a resource:
+A resource's data can be loaded into the HDX datastore and queried. HDX calls this
+[Tabular Data Endpoints (TDE)](https://docs.humdata.org/build/hdx-apis/tabular-data-endpoints),
+and it is available only for resources with TDE enabled (their `datastore_active`
+field is true). Contact hdx@un.org to request TDE for your datasets. HDX requires an
+API key for all datastore actions and dumps, including reads of public resources,
+and limits each IP address to 60 requests per minute. To create, update or delete
+the datastore for a resource:
 
     resource.create_datastore(
         [{"id": "code", "type": "text"}, {"id": "value", "type": "numeric"}],
@@ -1020,7 +1024,9 @@ internal `_id` field:
 
 **search_datastore_sql** wraps CKAN's `datastore_search_sql`, running a
 SELECT statement where tables are named by double quoted resource id and
-returning the list of records:
+returning the list of records. HDX returns at most 32,000 rows per query and
+cuts off larger results, in which case a warning is logged (see
+[paging in the TDE docs](https://docs.humdata.org/build/hdx-apis/tabular-data-endpoints/troubleshooting-and-error-handling#pagination)):
 
     records = Resource.search_datastore_sql(
         'SELECT code, SUM(value) AS total FROM "RESOURCE_ID" GROUP BY code'

src/hdx/data/resource.py

@@ -950,9 +950,11 @@ def get_datastore_records(
             )
             records = result["records"]
             yield from records
-            if len(records) < page_size:
+            # HDX caps limit at its maximum rows per call and returns the limit applied
+            limit = result["limit"]
+            if len(records) < limit:
                 return
-            offset += page_size
+            offset += limit
 
     def get_datastore_fields(self) -> list[dict]:
         """Get the field definitions of the resource datastore excluding the
@@ -971,7 +973,8 @@ def search_datastore_sql(
     ) -> list[dict]:
         """Run a read-only SQL query against the HDX datastore using CKAN
         datastore_search_sql. Tables are named by resource id, which must be
-        double quoted eg. SELECT * FROM "RESOURCE_ID".
+        double quoted eg. SELECT * FROM "RESOURCE_ID". Logs a warning if HDX
+        truncates the results.
 
         Args:
             sql: SQL SELECT statement
@@ -982,7 +985,12 @@ def search_datastore_sql(
         """
         resource = Resource(configuration=configuration)
         result = resource._write_to_hdx("datastore_search_sql", {"sql": sql})
-        return result["records"]
+        records = result["records"]
+        if result.get("records_truncated"):
+            logger.warning(
+                f"SQL query results truncated by HDX to {len(records)} rows!"
+            )
+        return records
 
     def delete_datastore(self) -> None:
         """Delete a resource from the HDX datastore

tests/hdx/data/test_resource.py

@@ -2,6 +2,7 @@
 
 import copy
 import json
+import logging
 import os
 from datetime import datetime, timezone
 from os import remove
@@ -340,6 +341,7 @@ def mockresourceview(url, decodedata):
 class TestResource:
     datastore = None
     datastore_request = None
+    datastore_rows_max = None
 
     @pytest.fixture(scope="class")
     def static_yaml(self, configfolder):
@@ -604,6 +606,7 @@ def post(url, data, **kwargs):
 
     @pytest.fixture(scope="function")
     def post_datastore_query(self):
+        TestResource.datastore_rows_max = None
         all_records = [
             {"_id": 1, "code": "A", "value": 1.0},
             {"_id": 2, "code": "B", "value": 2.0},
@@ -626,6 +629,8 @@ def post(url, data, **kwargs):
                             '{"success": false, "error": {"message": "Bad SQL", "__type": "Validation Error"}, "help": "http://test-data.humdata.org/api/3/action/help_show?name=datastore_search_sql"}',
                         )
                     result = {"records": all_records[:2], "fields": fields}
+                    if "truncated" in datadict["sql"]:
+                        result["records_truncated"] = True
                     return MockResponse(
                         200,
                         json.dumps(
@@ -645,10 +650,13 @@ def post(url, data, **kwargs):
                     TestResource.datastore_request = datadict
                     offset = datadict["offset"]
                     limit = datadict["limit"]
+                    if TestResource.datastore_rows_max is not None:
+                        limit = min(limit, TestResource.datastore_rows_max)
                     result = {
                         "resource_id": datadict["resource_id"],
                         "fields": fields,
                         "records": all_records[offset : offset + limit],
+                        "limit": limit,
                     }
                     if datadict["include_total"]:
                         result["total"] = len(all_records)
@@ -1175,7 +1183,7 @@ def test_datastore(self, configuration, post_datastore, topline_yaml, topline_js
         resource.update_datastore(records, method="insert")
         assert TestResource.datastore == "upsert"
 
-    def test_datastore_queries(self, configuration, post_datastore_query):
+    def test_datastore_queries(self, configuration, post_datastore_query, caplog):
         resource = Resource({"id": "datastore_query_resource"})
         result = resource.search_datastore()
         assert result["total"] == 3
@@ -1220,6 +1228,13 @@ def test_datastore_queries(self, configuration, post_datastore_query):
         assert len(records) == 3
         assert TestResource.datastore_request["offset"] == 3
 
+        TestResource.datastore_rows_max = 2
+        records = list(resource.get_datastore_records(page_size=5))
+        assert [record["code"] for record in records] == ["A", "B", "C"]
+        assert TestResource.datastore_request["offset"] == 2
+        assert TestResource.datastore_request["limit"] == 5
+        TestResource.datastore_rows_max = None
+
         assert resource.get_datastore_fields() == [
             {"id": "code", "type": "text"},
             {"id": "value", "type": "numeric"},
@@ -1230,6 +1245,13 @@ def test_datastore_queries(self, configuration, post_datastore_query):
             'SELECT * FROM "datastore_query_resource" LIMIT 2'
         )
         assert [record["code"] for record in records] == ["A", "B"]
+        assert "truncated" not in caplog.text
+        with caplog.at_level(logging.WARNING):
+            records = Resource.search_datastore_sql(
+                'SELECT * FROM "truncated_resource"'
+            )
+        assert len(records) == 2
+        assert "truncated by HDX to 2 rows" in caplog.text
         with pytest.raises(HDXError):
             Resource.search_datastore_sql("error")