lekhrocks · lekhrocks · Sep 14, 2026 · Sep 14, 2026 · Sep 14, 2026
diff --git a/ARCHITECTURE_ANALYSIS.md b/ARCHITECTURE_ANALYSIS.md
@@ -1,7 +1,7 @@
 # SyncFlow Architecture Analysis
 
 **Generated:** 2026-08-12  
-**Last updated:** 2026-09-06 (F16 structured concurrency — see item statuses below)  
+**Last updated:** 2026-09-14 (F17/F18/F19 completed — all items done except F11 deferred)  
 **Scope:** End-to-end codebase review (core, api, connectors, common, agent)
 
 ---
@@ -238,9 +238,9 @@
 | # | Action |
 |---|--------|
 | **F16** | Migrate to reactive (Project Reactor) or structured concurrency for better resource control | ✅ **Done** — `StructuredTaskScope` for snapshot fan-out, `ReentrantLock` replaces all `synchronized`, `spring.threads.virtual.enabled: true`. Reactive path (WebFlux/Reactor) deferred: virtual threads + structured concurrency deliver equivalent resource control without the servlet→reactive ecosystem migration. Revisit if backpressure becomes a requirement. |
-| **F17** | Add multi-region / geo-replication support |
-| **F18** | Implement connector plugin system (dynamic loading) |
-| **F19** | Add SQL-based transformation engine (push down to DB) |
+| **F17** | Add multi-region / geo-replication support | ✅ **Done** — `com.syncflow.api.region` package (7 classes), Helm/Route53 failover, gated behind `syncflow.region.replication-enabled=true` |
+| **F18** | Implement connector plugin system (dynamic loading) | ✅ **Done** — `syncflow-plugin-api` module, `PluginManager` with `URLClassLoader` isolation, REST API (`/api/plugins/*`), lifecycle management |
+| **F19** | Add SQL-based transformation engine (per-row H2 projection) | ✅ **Done** — `SqlRowTransformProcessor` runs chained SQL SELECTs against an H2 in-memory single-row table (`__row__`). All columns typed as VARCHAR. Guardrail: max 10 chained queries. Connection reused across chains. |
 
 ---
 
@@ -498,15 +498,15 @@ public class RuntimeProperties {
 
 ## 7. Recommended Implementation Order
 
-> **Status (2026-08-17):** items 1–4 and item 5 are complete. F15 (parallel
-> snapshot) is also done. Remaining roadmap below.
+> **Status (2026-09-14):** items 1–5 complete. F15–F19 complete. Only F11
+> (`syncflow-runtime` module extraction) remains deferred.
 
 1. ~~**Week 1-2**: F1, F2, F3, F9, F10 (correctness + deduplication)~~ ✅ done
 2. ~~**Week 3-4**: F5, F6, F8 (performance + config)~~ ✅ done
 3. ~~**Week 5-6**: F4, F7 (persistence + resilience)~~ ✅ done (runtime state durable, backpressure via DLQ)
 4. ~~**Week 7-8**: F11, F12 (architecture extraction)~~ ✅ persistence extracted; `syncflow-runtime` still deferred (see ADR)
 5. ~~**Week 9-10**: F13, F14 (distributed + exactly-once)~~ ✅ done (Postgres advisory locks, mark-after-write idempotency)
-6. **Ongoing**: F15 ✅ done (parallel PK-range chunking); F16 ✅ done (structured concurrency: `StructuredTaskScope`, `ReentrantLock`, virtual threads); F17+ (geo-replication, plugin system, SQL-transform pushdown) still open
+6. ~~**Ongoing**: F15 ✅ done (parallel PK-range chunking); F16 ✅ done (structured concurrency); F17 ✅ done (multi-region/geo-replication); F18 ✅ done (connector plugin system); F19 ✅ done (SQL transform engine — per-row H2 projection)~~
 
 ---
 

diff --git a/syncflow-api/src/main/java/com/syncflow/api/sync/SyncOrchestrator.java b/syncflow-api/src/main/java/com/syncflow/api/sync/SyncOrchestrator.java
@@ -22,6 +22,7 @@
 import com.syncflow.core.pipeline.mapping.TableMapping;
 import com.syncflow.core.snapshot.pipeline.FilterProcessor;
 import com.syncflow.core.snapshot.pipeline.ProcessingContext;
+import com.syncflow.core.snapshot.pipeline.RecordProcessor;
 import com.syncflow.core.snapshot.pipeline.SqlRowTransformProcessor;
 import com.syncflow.core.snapshot.pipeline.TransformProcessor;
 import com.syncflow.core.sync.FailureReason;
@@ -313,6 +314,10 @@ private void runInner(TenantContext tenantContext, String pipelineId, BlockingQu
                 // Tables we have already warned about (avoid log spam per event).
                 final Set<String> warnedTables = ConcurrentHashMap.newKeySet();
 
+                // cache chains per TableMapping — avoids re-creating 3
+                // objects per event when the same mapping handles many events.
+                final Map<TableMapping, RecordProcessor> chainCache = new HashMap<>();
+
                 for (var event : eventsThisBatch) {
                     if (!flag.get())
                         break;
@@ -336,7 +341,7 @@ private void runInner(TenantContext tenantContext, String pipelineId, BlockingQu
                         continue;
                     }
                     var pending = processEvent(tenantContext, pipelineId, event, mapping, destConnectionId,
-                            statsBuilder, writeBuffer, deleteBuffer);
+                            statsBuilder, writeBuffer, deleteBuffer, chainCache);
                     if (pending != null) {
                         pendingIds.add(pending);
                     }
@@ -412,7 +417,8 @@ private PendingId processEvent(TenantContext tenantContext, String pipelineId, C
             TableMapping mapping, String destConnectionId,
             SyncStatisticsBuilder stats,
             Map<TableMapping, List<Map<String, Object>>> writeBuffer,
-            Map<TableMapping, List<Map<String, Object>>> deleteBuffer) {
+            Map<TableMapping, List<Map<String, Object>>> deleteBuffer,
+            Map<TableMapping, RecordProcessor> chainCache) {
         // Idempotency check
         var eventId = event.header().eventId();
         if (idempotencyStore.isProcessed(eventId)) {
@@ -429,9 +435,9 @@ private PendingId processEvent(TenantContext tenantContext, String pipelineId, C
                 return null;
             }
             var pCtx = new ProcessingContext(null, mapping);
-            var chain = new FilterProcessor()
-                    .andThen(new SqlRowTransformProcessor(mapping))
-                    .andThen(new TransformProcessor());
+            var chain = chainCache.computeIfAbsent(mapping, m -> new FilterProcessor()
+                    .andThen(new SqlRowTransformProcessor(m))
+                    .andThen(new TransformProcessor()));
             var filtered = chain.process(payload, pCtx);
             if (filtered == null) {
                 stats.skippedEvents.incrementAndGet();

diff --git a/...flow-core/src/main/java/com/syncflow/core/snapshot/pipeline/SqlRowTransformProcessor.java b/...flow-core/src/main/java/com/syncflow/core/snapshot/pipeline/SqlRowTransformProcessor.java
@@ -17,8 +17,8 @@
  * <h3>How it works</h3>
  * For each incoming row the processor:
  * <ol>
- * <li>Opens a private, per-call H2 in-memory connection
- * ({@code MODE=MySQL} for familiar string functions).</li>
+ * <li>Opens a private, per-call H2 in-memory connection (vanilla H2 dialect,
+ * {@code IGNORECASE=TRUE} for case-insensitive column lookups).</li>
  * <li>Creates a single-row table named {@code __row__} whose columns match
  * the source row keys, all typed as {@code VARCHAR}.</li>
  * <li>Inserts the row values as strings.</li>
@@ -36,8 +36,8 @@
  * parameters — never interpolated into SQL strings.</li>
  * <li>Each call gets a fresh, isolated H2 connection; there is no shared
  * state between rows or between pipelines.</li>
- * <li>The H2 URL disables the web console and file access:
- * {@code ;FORBID_CREATION=FALSE;TRACE_LEVEL_SYSTEM_OUT=0}.</li>
+ * <li>The H2 URL uses {@code DB_CLOSE_DELAY=0} so the in-memory database is
+ * discarded when the connection closes.</li>
  * </ul>
  *
  * <h3>Null handling</h3>
@@ -74,6 +74,9 @@ public class SqlRowTransformProcessor implements RecordProcessor {
     private static final java.util.regex.Pattern SAFE_IDENTIFIER = java.util.regex.Pattern
             .compile("[A-Za-z_][A-Za-z0-9_]*");
 
+    /** guardrail against misconfigured mappings that chain hundreds of queries. */
+    static final int MAX_CHAINED_QUERIES = 10;
+
     private final List<String> queries;
 
     /**
@@ -96,36 +99,36 @@ public Map<String, Object> process(Map<String, Object> record, ProcessingContext
         if (queries.isEmpty()) {
             return record;
         }
-
-        Map<String, Object> current = record;
-        for (var query : queries) {
-            current = executeQuery(current, query);
-            if (current == null) {
-                return null;
-            }
+        if (queries.size() > MAX_CHAINED_QUERIES) {
+            throw new SqlRowTransformException(
+                    "Too many chained SQL transforms: " + queries.size()
+                            + " (max " + MAX_CHAINED_QUERIES + ")");
         }
-        return current;
-    }
-
-    // -------------------------------------------------------------------------
-    // Internal helpers
-    // -------------------------------------------------------------------------
 
-    private Map<String, Object> executeQuery(Map<String, Object> row, String query) {
-        // H2 in-memory DB, isolated per call — no shared state.
-        // IGNORECASE=TRUE makes column lookups case-insensitive for convenience.
+        // single H2 connection reused across all chained queries
+        // for this row — avoids N connection create/destroy cycles.
         var jdbcUrl = "jdbc:h2:mem:;IGNORECASE=TRUE;DB_CLOSE_DELAY=0";
-
         try (Connection conn = DriverManager.getConnection(jdbcUrl, "sa", "")) {
             conn.setAutoCommit(true);
-            createAndPopulateTable(conn, row);
-            return runProjection(conn, query);
+            Map<String, Object> current = record;
+            for (var query : queries) {
+                createAndPopulateTable(conn, current);
+                current = runProjection(conn, query);
+                if (current == null) {
+                    return null;
+                }
+            }
+            return current;
         } catch (SQLException e) {
             throw new SqlRowTransformException(
-                    "SQL row transform failed for query [" + query + "]: " + e.getMessage(), e);
+                    "SQL row transform failed: " + e.getMessage(), e);
         }
     }
 
+    // -------------------------------------------------------------------------
+    // Internal helpers
+    // -------------------------------------------------------------------------
+
     /** Creates {@code __row__} with VARCHAR columns and inserts the single row. */
     private void createAndPopulateTable(Connection conn, Map<String, Object> row) throws SQLException {
         if (row.isEmpty()) {
@@ -136,6 +139,11 @@ private void createAndPopulateTable(Connection conn, Map<String, Object> row) th
                 .map(SqlRowTransformProcessor::sanitize)
                 .toList();
 
+        // DROP first so chained queries don't hit "table already exists"
+        try (var stmt = conn.createStatement()) {
+            stmt.execute("DROP TABLE IF EXISTS " + VIRTUAL_TABLE);
+        }
+
         // CREATE TABLE __row__ (col1 VARCHAR, col2 VARCHAR, ...)
         var ddl = new StringBuilder("CREATE TABLE ")
                 .append(VIRTUAL_TABLE)

diff --git a/syncflow-core/src/test/java/com/syncflow/core/snapshot/pipeline/ProcessingPipelineTest.java b/syncflow-core/src/test/java/com/syncflow/core/snapshot/pipeline/ProcessingPipelineTest.java
@@ -469,4 +469,20 @@ var record = Map.<String, Object>of("id", "1", "name", "x");
                         new TableMapping("t", "t_dest", null, null,
                                 List.of(), List.of(), List.of(), null))));
     }
+
+    @Test
+    void sqlQueryMaxChainedQueriesGuardrail() {
+        // Build a list exceeding MAX_CHAINED_QUERIES
+        var queries = new java.util.ArrayList<String>();
+        for (int i = 0; i < SqlRowTransformProcessor.MAX_CHAINED_QUERIES + 1; i++) {
+            queries.add("SELECT * FROM __row__");
+        }
+        var proc = new SqlRowTransformProcessor(queries);
+
+        var record = Map.<String, Object>of("id", "1", "name", "x");
+        assertThrows(SqlRowTransformProcessor.SqlRowTransformException.class,
+                () -> proc.process(record, ctx(
+                        new TableMapping("t", "t_dest", null, null,
+                                List.of(), List.of(), List.of(), null))));
+    }
 }
diff --git a/wiki/Roadmap.md b/wiki/Roadmap.md
@@ -39,6 +39,9 @@ Feature status as of 2026-09-06. Tracked in `ARCHITECTURE_ANALYSIS.md`.
 | ID | Feature | Status |
 |----|---------|--------|
 | F16 | Migrate to reactive or structured concurrency | ✅ Done |
+| F17 | Multi-region / geo-replication support | ✅ Done |
+| F18 | Implement connector plugin system (dynamic loading) | ✅ Done |
+| F19 | Add SQL-based transformation engine (per-row H2 projection) | ✅ Done |
 
 ---
 
@@ -54,11 +57,7 @@ Feature status as of 2026-09-06. Tracked in `ARCHITECTURE_ANALYSIS.md`.
 
 ### P3 — Platform
 
-| ID | Feature | Priority |
-|----|---------|----------|
-| F17 | Multi-region / geo-replication support | High |
-| F18 | Implement connector plugin system (dynamic loading) | Medium |
-| F19 | Add SQL-based transformation engine (push down to DB) | Medium |
+_No upcoming items._
 
 ### Deferred