Fixing OOM error in new-updates source component

Author: jcortes

components/google_sheets/sources/common/new-updates.mjs

@@ -20,8 +20,21 @@ export default {
       label: "Worksheet ID(s)",
       description: "Select one or more worksheet(s), or provide an array of worksheet IDs.",
     },
+    maxRows: {
+      type: "integer",
+      label: "Max Rows to Monitor",
+      description: "Maximum number of rows to monitor for changes. Defaults to 10000. Increase with caution as larger values may cause memory issues.",
+      optional: true,
+      default: 10000,
+    },
   },
   methods: {
+    getBatchSize() {
+      return 1000; // Process 1000 rows at a time
+    },
+    getMaxRows() {
+      return this.maxRows || 10000;
+    },
     getMeta(spreadsheet, worksheet) {
       const {
         sheetId: worksheetId,
@@ -56,6 +69,9 @@ export default {
     getWorksheetIds() {
       return this.worksheetIDs.map((i) => i.toString());
     },
+    _getBatchKey(baseId, batchIndex) {
+      return `${baseId}_batch_${batchIndex}`;
+    },
     _getSheetValues(id) {
       const stringBuffer = this.db.get(id);
 
@@ -72,6 +88,51 @@ export default {
       const stringBuffer = compressed.toString("base64");
       this.db.set(id, stringBuffer);
     },
+    _getBatchedSheetValues(baseId) {
+      const allValues = [];
+      let batchIndex = 0;
+      let hasMore = true;
+
+      while (hasMore) {
+        const batchKey = this._getBatchKey(baseId, batchIndex);
+        const batchValues = this._getSheetValues(batchKey);
+
+        if (!batchValues) {
+          hasMore = false;
+          break;
+        }
+
+        allValues.push(...batchValues);
+        batchIndex++;
+      }
+
+      return allValues.length > 0
+        ? allValues
+        : null;
+    },
+    _setBatchedSheetValues(baseId, sheetValues) {
+      const batchSize = this.getBatchSize();
+      const maxRows = this.getMaxRows();
+
+      // Limit to maxRows
+      const limitedValues = sheetValues.slice(0, maxRows);
+
+      // Clear old batches first
+      let batchIndex = 0;
+      while (this.db.get(this._getBatchKey(baseId, batchIndex))) {
+        this.db.set(this._getBatchKey(baseId, batchIndex), undefined);
+        batchIndex++;
+      }
+
+      // Store in batches
+      batchIndex = 0;
+      for (let i = 0; i < limitedValues.length; i += batchSize) {
+        const batch = limitedValues.slice(i, i + batchSize);
+        const batchKey = this._getBatchKey(baseId, batchIndex);
+        this._setSheetValues(batchKey, batch);
+        batchIndex++;
+      }
+    },
     indexToColumnLabel(index) {
       let columnLabel = "";
       while (index >= 0) {
@@ -133,10 +194,8 @@ export default {
     },
     async getContentDiff(spreadsheet, worksheet) {
       const sheetId = this.getSheetId();
-      const oldValues =
-        this._getSheetValues(
-          `${spreadsheet.spreadsheetId}${worksheet.properties.sheetId}`,
-        ) || null;
+      const baseId = `${spreadsheet.spreadsheetId}${worksheet.properties.sheetId}`;
+      const oldValues = this._getBatchedSheetValues(baseId) || null;
       const currentValues = await this.googleSheets.getSpreadsheetValues(
         sheetId,
         worksheet.properties.title,
@@ -169,9 +228,11 @@ export default {
           continue;
         }
 
-        const offsetLength = Math.max(values.length - offset, 0);
+        const maxRows = this.getMaxRows();
+        const offsetLength = Math.max(Math.min(values.length, maxRows) - offset, 0);
         const offsetValues = values.slice(0, offsetLength);
-        this._setSheetValues(`${sheetId}${worksheetId}`, offsetValues);
+        const baseId = `${sheetId}${worksheetId}`;
+        this._setBatchedSheetValues(baseId, offsetValues);
       }
     },
     async processSpreadsheet(spreadsheet) {
@@ -191,8 +252,11 @@ export default {
           spreadsheet,
           worksheet,
         );
-        const newValues = currentValues.values || [];
+        const maxRows = this.getMaxRows();
+        const rawNewValues = currentValues.values || [];
+        const newValues = rawNewValues.slice(0, maxRows);
         let changes = [];
+
         // check if there are differences in the spreadsheet values
         if (JSON.stringify(oldValues) !== JSON.stringify(newValues)) {
           let rowCount = this.getRowCount(newValues, oldValues);
@@ -214,10 +278,8 @@ export default {
             this.getMeta(spreadsheet, worksheet),
           );
         }
-        this._setSheetValues(
-          `${spreadsheet.spreadsheetId}${worksheet.properties.sheetId}`,
-          newValues || [],
-        );
+        const baseId = `${spreadsheet.spreadsheetId}${worksheet.properties.sheetId}`;
+        this._setBatchedSheetValues(baseId, newValues || []);
       }
     },
   },

components/google_sheets/sources/new-updates-polling/new-updates-polling.mjs

@@ -20,6 +20,12 @@ export default {
         intervalSeconds: DEFAULT_POLLING_SOURCE_TIMER_INTERVAL,
       },
     },
+    // eslint-disable-next-line pipedream/props-label, pipedream/props-description
+    alert: {
+      type: "alert",
+      alertType: "info",
+      content: "**Important**: If your sheet has more than 1000 rows, please set the **Monitoring Range** below to avoid performance issues and potential disruptions. Example: `A1:Z1000` to monitor the first 1000 rows with columns A through Z.",
+    },
     watchedDrive: {
       propDefinition: [
         googleSheets,
@@ -36,11 +42,50 @@ export default {
         }),
       ],
     },
-    ...common.props,
+    worksheetIDs: {
+      propDefinition: [
+        googleSheets,
+        "worksheetIDs",
+        (c) => ({
+          sheetId: c.sheetID,
+        }),
+      ],
+      type: "integer[]",
+      label: "Worksheet ID(s)",
+      description: "Select one or more worksheet(s), or provide an array of worksheet IDs.",
+    },
+    monitoringRange: {
+      type: "string",
+      label: "Monitoring Range",
+      description: "The A1 notation range to monitor for changes (e.g., `A1:B100` or `Sheet1!A1:Z1000`). If not specified, the entire sheet will be monitored up to 10000 rows. **Recommended for sheets with more than 1000 rows**.",
+      optional: true,
+    },
   },
   methods: {
     ...base.methods,
     ...common.methods,
+    getMonitoringRange() {
+      return this.monitoringRange;
+    },
+    async getContentDiff(spreadsheet, worksheet) {
+      const sheetId = this.getSheetId();
+      const baseId = `${spreadsheet.spreadsheetId}${worksheet.properties.sheetId}`;
+      const oldValues = this._getBatchedSheetValues(baseId) || null;
+
+      // Use monitoring range if specified, otherwise use worksheet title
+      const range = this.monitoringRange
+        ? this.monitoringRange
+        : worksheet.properties.title;
+
+      const currentValues = await this.googleSheets.getSpreadsheetValues(
+        sheetId,
+        range,
+      );
+      return {
+        oldValues,
+        currentValues,
+      };
+    },
   },
   hooks: {
     async deploy() {

components/google_sheets/sources/new-updates/new-updates.mjs

@@ -9,7 +9,7 @@ export default {
   type: "source",
   name: "New Updates (Instant)",
   description: "Emit new event each time a row or cell is updated in a spreadsheet.",
-  version: "0.3.3",
+  version: "0.3.4",
   dedupe: "unique",
   props: {
     ...httpBase.props,