(beam) branch master updated: Add LRU cache eviction to CombinePerKeyPrecombineOperator (#37466)

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

damccorm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new e4cd040199d Add LRU cache eviction to CombinePerKeyPrecombineOperator (#37466)
e4cd040199d is described below

commit e4cd040199dfb7ac7e1f19c5237e3cd58a9815d9
Author: M Junaid Shaukat <15...@users.noreply.github.com>
AuthorDate: Tue Feb 3 04:09:49 2026 +0500

    Add LRU cache eviction to CombinePerKeyPrecombineOperator (#37466)
    
    * [TypeScript] Add LRU cache eviction to CombinePerKeyPrecombineOperator
    
    - Added touchKey() method to move accessed keys to end (most recently used)
    - Added flushLRU() method to evict entries from front (least recently used)
    - Updated process() to track key usage for proper LRU ordering
    
    JavaScript Maps preserve insertion order, so we implement LRU by:
    - On access: delete and re-insert the key (moves to end)
    - On eviction: iterate from front (oldest entries first)
    
    Fixes #37465
    
    * Address review comments: remove flush, inline touchKey
---
 .../typescript/src/apache_beam/worker/operators.ts | 44 +++++++++++++++-------
 1 file changed, 30 insertions(+), 14 deletions(-)

diff --git a/sdks/typescript/src/apache_beam/worker/operators.ts b/sdks/typescript/src/apache_beam/worker/operators.ts
index 941ad62ef2d..9de10afdeeb 100644
--- a/sdks/typescript/src/apache_beam/worker/operators.ts
+++ b/sdks/typescript/src/apache_beam/worker/operators.ts
@@ -427,6 +427,8 @@ export class CombinePerKeyPrecombineOperator<I, A, O>
   keyCoder: Coder<unknown>;
   windowCoder: Coder<Window>;
 
+  // Use a Map for LRU ordering: JavaScript Maps preserve insertion order,
+  // so we can implement LRU by deleting and re-inserting on access.
   groups: Map<string, A>;
   maxKeys: number = 10000;
 
@@ -474,18 +476,25 @@ export class CombinePerKeyPrecombineOperator<I, A, O>
         encodeToBase64(window, this.windowCoder) +
         " " +
         encodeToBase64(wvalue.value.key, this.keyCoder);
-      if (!this.groups.has(wkey)) {
-        this.groups.set(wkey, this.combineFn.createAccumulator());
+      const existingAccumulator = this.groups.get(wkey);
+      const newAccumulator =
+        existingAccumulator !== undefined
+          ? this.combineFn.addInput(existingAccumulator, wvalue.value.value)
+          : this.combineFn.addInput(
+              this.combineFn.createAccumulator(),
+              wvalue.value.value,
+            );
+      // Move to end (most recently used) by delete + set
+      // Only delete if the key already exists (avoids unnecessary operation for new keys)
+      if (existingAccumulator !== undefined) {
+        this.groups.delete(wkey);
       }
-      this.groups.set(
-        wkey,
-        this.combineFn.addInput(this.groups.get(wkey), wvalue.value.value),
-      );
+      this.groups.set(wkey, newAccumulator);
     }
     if (this.groups.size > this.maxKeys) {
-      // Flush a random 10% of the map to make more room.
-      // TODO: Tune this, or better use LRU or ARC for this cache.
-      return this.flush(this.maxKeys * 0.9);
+      // Flush the least recently used entries (at the front of the Map)
+      // until we're back under maxKeys.
+      return this.flushLRU(this.maxKeys);
     } else {
       return NonPromise;
     }
@@ -495,10 +504,19 @@ export class CombinePerKeyPrecombineOperator<I, A, O>
     this.groups = new Map();
   }
 
-  flush(target: number): ProcessResult {
+  /**
+   * Flushes entries from the cache using LRU eviction.
+   * Evicts the least recently used entries (from the front of the Map)
+   * until the cache size is at or below the target.
+   */
+  flushLRU(target: number): ProcessResult {
     const result = new ProcessResultBuilder();
     const toDelete: string[] = [];
+    // Iterate from the front (oldest/least recently used entries)
     for (const [wkey, values] of this.groups) {
+      if (this.groups.size - toDelete.length <= target) {
+        break;
+      }
       const parts = wkey.split(" ");
       const encodedWindow = parts[0];
       const encodedKey = parts[1];
@@ -515,9 +533,6 @@ export class CombinePerKeyPrecombineOperator<I, A, O>
         }),
       );
       toDelete.push(wkey);
-      if (this.groups.size - toDelete.length <= target) {
-        break;
-      }
     }
     for (const wkey of toDelete) {
       this.groups.delete(wkey);
@@ -526,7 +541,8 @@ export class CombinePerKeyPrecombineOperator<I, A, O>
   }
 
   async finishBundle() {
-    const maybePromise = this.flush(0);
+    // Flush all remaining entries using flushLRU(0)
+    const maybePromise = this.flushLRU(0);
     if (maybePromise !== NonPromise) {
       await maybePromise;
     }