From a10a24b1b7e8dce25397a735047aef2b791490b4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Kevin=20Carpenter=E2=84=A2=EF=B8=8F?= <kevin@moderne.io>
Date: Fri, 3 Apr 2026 21:54:48 -0300
Subject: [PATCH] Add getAllRows() to DataTableStore for cross-bucket
 aggregation

InMemoryDataTableStore.getRows(name, group) only returns rows from the
first matching bucket. When multiple sub-recipes in a composite recipe
(like DevCenter) create separate DataTable instances with the same class
name but different instance names, each gets its own bucket. getRows()
finds only the first one, silently dropping rows from the others.

This causes non-deterministic data loss because ConcurrentHashMap
iteration order determines which bucket is found first.

Add DataTableStore.getAllRows(name) that aggregates rows across all
buckets with a given data table name, regardless of group or instance
name. InMemoryDataTableStore overrides it to scan all buckets.
---
 .../main/java/org/openrewrite/DataTableStore.java   | 11 +++++++++++
 .../org/openrewrite/InMemoryDataTableStore.java     | 13 +++++++++++++
 2 files changed, 24 insertions(+)

diff --git a/rewrite-core/src/main/java/org/openrewrite/DataTableStore.java b/rewrite-core/src/main/java/org/openrewrite/DataTableStore.java
index b337fe202b..ca7071f1c9 100644
--- a/rewrite-core/src/main/java/org/openrewrite/DataTableStore.java
+++ b/rewrite-core/src/main/java/org/openrewrite/DataTableStore.java
@@ -63,6 +63,17 @@ static DataTableStore noop() {
      */
     Stream<?> getRows(String dataTableName, @Nullable String group);
 
+    /**
+     * Stream all rows for a given data table name, merging across all
+     * groups and instance names.
+     *
+     * @param dataTableName the fully qualified class name of the data table
+     * @return a stream of all rows across all buckets with this name
+     */
+    default Stream<?> getAllRows(String dataTableName) {
+        return getRows(dataTableName, null);
+    }
+
     /**
      * Get the set of {@link DataTable} instances that have received rows.
      *
diff --git a/rewrite-core/src/main/java/org/openrewrite/InMemoryDataTableStore.java b/rewrite-core/src/main/java/org/openrewrite/InMemoryDataTableStore.java
index 5ce6612fa4..61bd45652a 100644
--- a/rewrite-core/src/main/java/org/openrewrite/InMemoryDataTableStore.java
+++ b/rewrite-core/src/main/java/org/openrewrite/InMemoryDataTableStore.java
@@ -78,6 +78,19 @@ private static Stream<Object> snapshotRows(Bucket bucket) {
         return snapshot.stream();
     }
 
+    @Override
+    public Stream<?> getAllRows(String dataTableName) {
+        List<Object> allRows = new ArrayList<>();
+        for (Bucket bucket : buckets.values()) {
+            if (bucket.dataTable.getName().equals(dataTableName)) {
+                synchronized (bucket.rows) {
+                    allRows.addAll(bucket.rows);
+                }
+            }
+        }
+        return allRows.stream();
+    }
+
     @Override
     public Collection<DataTable<?>> getDataTables() {
         List<DataTable<?>> result = new ArrayList<>(buckets.size());