From 055153f43f1f1cb30cf5fd880a69fe5afa90cb03 Mon Sep 17 00:00:00 2001 From: lekhrocks Date: Mon, 14 Sep 2026 18:46:47 +0530 Subject: [PATCH 1/2] =?UTF-8?q?fix(F19):=20harden=20SQL=20transform=20engi?= =?UTF-8?q?ne=20=E2=80=94=20reuse=20H2=20connection,=20add=20guardrail,=20?= =?UTF-8?q?fix=20docs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SqlRowTransformProcessor: - Reuse single H2 connection across chained queries (was N per row) - Add DROP TABLE IF EXISTS before CREATE TABLE for chained query support - Cap chained queries at MAX_CHAINED_QUERIES=10 - Fix Javadoc: remove false MODE=MySQL claim, correct H2 URL description - Add sqlQueryMaxChainedQueriesGuardrail test SyncOrchestrator: - Cache processing chains per TableMapping via chainCache.computeIfAbsent to avoid re-creating 3 objects per CDC event Docs: - Move F17/F18/F19 from Upcoming to Completed in wiki/Roadmap.md - Update ARCHITECTURE_ANALYSIS.md with accurate F17/F18/F19 status --- ARCHITECTURE_ANALYSIS.md | 14 ++--- .../syncflow/api/sync/SyncOrchestrator.java | 17 ++++-- .../pipeline/SqlRowTransformProcessor.java | 56 +++++++++++-------- .../pipeline/ProcessingPipelineTest.java | 16 ++++++ wiki/Roadmap.md | 9 ++- 5 files changed, 71 insertions(+), 41 deletions(-) diff --git a/ARCHITECTURE_ANALYSIS.md b/ARCHITECTURE_ANALYSIS.md index cec7740..d3ea7cc 100644 --- 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 index d391c52..c0f155a 100644 --- 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 warnedTables = ConcurrentHashMap.newKeySet(); + // cache chains per TableMapping — avoids re-creating 3 + // objects per event when the same mapping handles many events. + final Map 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>> writeBuffer, - Map>> deleteBuffer) { + Map>> deleteBuffer, + Map chainCache) { // Idempotency check var eventId = event.header().eventId(); if (idempotencyStore.isProcessed(eventId)) { @@ -429,9 +435,10 @@ 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/syncflow-core/src/main/java/com/syncflow/core/snapshot/pipeline/SqlRowTransformProcessor.java b/syncflow-core/src/main/java/com/syncflow/core/snapshot/pipeline/SqlRowTransformProcessor.java index 43b2e2c..a789cce 100644 --- a/syncflow-core/src/main/java/com/syncflow/core/snapshot/pipeline/SqlRowTransformProcessor.java +++ b/syncflow-core/src/main/java/com/syncflow/core/snapshot/pipeline/SqlRowTransformProcessor.java @@ -17,8 +17,8 @@ *

How it works

* For each incoming row the processor: *
    - *
  1. Opens a private, per-call H2 in-memory connection - * ({@code MODE=MySQL} for familiar string functions).
    2. + *
  2. Opens a private, per-call H2 in-memory connection (vanilla H2 dialect, + * {@code IGNORECASE=TRUE} for case-insensitive column lookups).
    3. *
  3. Creates a single-row table named {@code __row__} whose columns match * the source row keys, all typed as {@code VARCHAR}.
    4. *
  4. Inserts the row values as strings.
    5. @@ -36,8 +36,8 @@ * parameters — never interpolated into SQL strings. *
  5. Each call gets a fresh, isolated H2 connection; there is no shared * state between rows or between pipelines.
    6. - *
  6. The H2 URL disables the web console and file access: - * {@code ;FORBID_CREATION=FALSE;TRACE_LEVEL_SYSTEM_OUT=0}.
    7. + *
  7. The H2 URL uses {@code DB_CLOSE_DELAY=0} so the in-memory database is + * discarded when the connection closes.
    8. * * *

    Null handling

    @@ -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 queries; /** @@ -96,36 +99,36 @@ public Map process(Map record, ProcessingContext if (queries.isEmpty()) { return record; } - - Map 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 executeQuery(Map 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 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 row) throws SQLException { if (row.isEmpty()) { @@ -136,6 +139,11 @@ private void createAndPopulateTable(Connection conn, Map 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 index f5fe8ba..a6aaaec 100644 --- 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.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(); + for (int i = 0; i < SqlRowTransformProcessor.MAX_CHAINED_QUERIES + 1; i++) { + queries.add("SELECT * FROM __row__"); + } + var proc = new SqlRowTransformProcessor(queries); + + var record = Map.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 index 314ab54..c3c4a46 100644 --- 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 From 67bbebef972670bfb376d5035125ef508b0a840b Mon Sep 17 00:00:00 2001 From: lekhrocks Date: Mon, 14 Sep 2026 18:49:36 +0530 Subject: [PATCH 2/2] fix: apply spotless --- .../main/java/com/syncflow/api/sync/SyncOrchestrator.java | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) 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 index c0f155a..e050713 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/sync/SyncOrchestrator.java +++ b/syncflow-api/src/main/java/com/syncflow/api/sync/SyncOrchestrator.java @@ -435,10 +435,9 @@ private PendingId processEvent(TenantContext tenantContext, String pipelineId, C return null; } var pCtx = new ProcessingContext(null, mapping); - var chain = chainCache.computeIfAbsent(mapping, m -> - new FilterProcessor() - .andThen(new SqlRowTransformProcessor(m)) - .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();