Merge pull request #21996 from overleaf/jpa-stream-pg-result

[history-v1] postgres: getProjectBlobsBatch: stream records

GitOrigin-RevId: 94ed6dfc4a263fd9369cd380e6cc25c7bbf6decc

package-lock.json

@@ -29399,6 +29399,14 @@
       "resolved": "https://registry.npmjs.org/pg-copy-streams/-/pg-copy-streams-2.2.2.tgz",
       "integrity": "sha512-mjSqs6hrsRhBojCuY2hxyg48B+3th5ARBjMxBCEisIqBvdRD0g5ETdbts20TzrOfha8ueJQOmQCJCprtczJtGQ=="
     },
+    "node_modules/pg-cursor": {
+      "version": "2.12.1",
+      "resolved": "https://registry.npmjs.org/pg-cursor/-/pg-cursor-2.12.1.tgz",
+      "integrity": "sha512-V13tEaA9Oq1w+V6Q3UBIB/blxJrwbbr35/dY54r/86soBJ7xkP236bXaORUTVXUPt9B6Ql2BQu+uwQiuMfRVgg==",
+      "peerDependencies": {
+        "pg": "^8"
+      }
+    },
     "node_modules/pg-int8": {
       "version": "1.0.1",
       "resolved": "https://registry.npmjs.org/pg-int8/-/pg-int8-1.0.1.tgz",
@@ -29420,6 +29428,17 @@
       "resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.5.0.tgz",
       "integrity": "sha512-muRttij7H8TqRNu/DxrAJQITO4Ac7RmX3Klyr/9mJEOBeIpgnF8f9jAfRz5d3XwQZl5qBjF9gLsUtMPJE0vezQ=="
     },
+    "node_modules/pg-query-stream": {
+      "version": "4.7.1",
+      "resolved": "https://registry.npmjs.org/pg-query-stream/-/pg-query-stream-4.7.1.tgz",
+      "integrity": "sha512-UMgsgn/pOIYsIifRySp59vwlpTpLADMK9HWJtq5ff0Z3MxBnPMGnCQeaQl5VuL+7ov4F96mSzIRIcz+Duo6OiQ==",
+      "dependencies": {
+        "pg-cursor": "^2.12.1"
+      },
+      "peerDependencies": {
+        "pg": "^8"
+      }
+    },
     "node_modules/pg-types": {
       "version": "2.2.0",
       "resolved": "https://registry.npmjs.org/pg-types/-/pg-types-2.2.0.tgz",
@@ -39731,6 +39750,7 @@
         "mongodb": "6.7.0",
         "overleaf-editor-core": "*",
         "pg": "^8.7.1",
+        "pg-query-stream": "^4.2.4",
         "swagger-tools": "^0.10.4",
         "temp": "^0.8.3",
         "throng": "^4.0.0",
@@ -69378,6 +69398,7 @@
         "node-fetch": "^2.7.0",
         "overleaf-editor-core": "*",
         "pg": "^8.7.1",
+        "pg-query-stream": "^4.2.4",
         "sinon": "^9.0.2",
         "swagger-client": "^3.10.0",
         "swagger-tools": "^0.10.4",
@@ -69989,6 +70010,12 @@
       "resolved": "https://registry.npmjs.org/pg-copy-streams/-/pg-copy-streams-2.2.2.tgz",
       "integrity": "sha512-mjSqs6hrsRhBojCuY2hxyg48B+3th5ARBjMxBCEisIqBvdRD0g5ETdbts20TzrOfha8ueJQOmQCJCprtczJtGQ=="
     },
+    "pg-cursor": {
+      "version": "2.12.1",
+      "resolved": "https://registry.npmjs.org/pg-cursor/-/pg-cursor-2.12.1.tgz",
+      "integrity": "sha512-V13tEaA9Oq1w+V6Q3UBIB/blxJrwbbr35/dY54r/86soBJ7xkP236bXaORUTVXUPt9B6Ql2BQu+uwQiuMfRVgg==",
+      "requires": {}
+    },
     "pg-int8": {
       "version": "1.0.1",
       "resolved": "https://registry.npmjs.org/pg-int8/-/pg-int8-1.0.1.tgz",
@@ -70005,6 +70032,14 @@
       "resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.5.0.tgz",
       "integrity": "sha512-muRttij7H8TqRNu/DxrAJQITO4Ac7RmX3Klyr/9mJEOBeIpgnF8f9jAfRz5d3XwQZl5qBjF9gLsUtMPJE0vezQ=="
     },
+    "pg-query-stream": {
+      "version": "4.7.1",
+      "resolved": "https://registry.npmjs.org/pg-query-stream/-/pg-query-stream-4.7.1.tgz",
+      "integrity": "sha512-UMgsgn/pOIYsIifRySp59vwlpTpLADMK9HWJtq5ff0Z3MxBnPMGnCQeaQl5VuL+7ov4F96mSzIRIcz+Duo6OiQ==",
+      "requires": {
+        "pg-cursor": "^2.12.1"
+      }
+    },
     "pg-types": {
       "version": "2.2.0",
       "resolved": "https://registry.npmjs.org/pg-types/-/pg-types-2.2.0.tgz",

services/history-v1/package.json

@@ -35,6 +35,7 @@
     "mongodb": "6.7.0",
     "overleaf-editor-core": "*",
     "pg": "^8.7.1",
+    "pg-query-stream": "^4.2.4",
     "swagger-tools": "^0.10.4",
     "temp": "^0.8.3",
     "throng": "^4.0.0",

services/history-v1/storage/lib/blob_store/postgres.js

@@ -83,11 +83,11 @@ async function getProjectBlobsBatch(projectIds) {
   const blobs = new Map()
   if (projectIds.length === 0) return { nBlobs, blobs }
 
-  const records = await knex('project_blobs')
+  const cursor = knex('project_blobs')
     .select('project_id', 'hash_bytes', 'byte_length', 'string_length')
     .whereIn('project_id', projectIds)
-
-  for (const record of records) {
+    .stream()
+  for await (const record of cursor) {
     const found = blobs.get(record.project_id)
     if (found) {
       found.push(recordToBlob(record))