fix: InsertStream CONFLICT_IGNORE + PER_STREAM rolled back stream on commit-time duplicate (#4214)

[robfrank]

Summary

Closes #4214.

• InsertStream with conflict_mode = CONFLICT_IGNORE + transaction_mode = PER_STREAM used to roll back the entire stream on a commit-time DuplicatedKeyException, reporting failed = received instead of bumping ignored and committing the remaining rows.
• Root cause: the synchronous catch in insertRows(...) already handled CONFLICT_IGNORE, but under PER_STREAM the unique-index enforcement is deferred to ctx.flushCommit(true) in onCompleted, where recordCommitException (added in Bug: InsertStream surfaces commit-time constraint violations as a stream-level INTERNAL error instead of populating InsertSummary.errors #4198) reclassifies every row as failed regardless of conflict_mode.
• Fix: resolve CONFLICT_IGNORE synchronously via a pre-save key-column lookup (keyExistsVertex / keyExistsDocument, mirroring the existing tryUpsertVertex / tryUpsertDocument pattern used by CONFLICT_UPDATE). The duplicate row never reaches save(), so the deferred commit never sees it; ignored is bumped exactly once per skipped row and the remaining rows commit cleanly.

Test plan

• New regression test Issue4214InsertStreamConflictIgnoreIT: inserts Bob, duplicate Alice, Carol against a pre-existing Alice under PER_STREAM + CONFLICT_IGNORE + key_columns = ["name"]. Asserts received=3, inserted=2, ignored=1, failed=0, errors=[], and confirms all three names persist via a follow-up SELECT count(*).
• Existing Issue4198InsertStreamCommitErrorIT still passes (CONFLICT_ERROR + commit-time violation still surfaces structured errors).
• Full GrpcServerIT suite passes (28 tests, including bulkInsertWithConflictIgnore under PER_BATCH).

[codacy-production]

Up to standards ✅

🟢 Issues 0 issues

Results:
0 new issues

View in Codacy

🟢 Metrics 9 complexity

Metric	Results
Complexity	9

View in Codacy

🟢 Coverage 90.00% diff coverage · -7.47% coverage variation

Metric	Results
Coverage variation	✅ -7.47% coverage variation
Diff coverage	✅ 90.00% diff coverage

View coverage diff in Codacy

Coverage variation details

	Coverable lines	Covered lines	Coverage
Common ancestor commit (6eb94b1)	127277	93796	73.69%
Head commit (8d2c7a6)	158965 (+31688)	105270 (+11474)	66.22% (-7.47%)

Coverage variation is the difference between the coverage for the head and common ancestor commits of the pull request branch: <coverage of head commit> - <coverage of common ancestor commit>

Diff coverage details

	Coverable lines	Covered lines	Diff coverage
Pull request (#4262)	10	9	90.00%

Diff coverage is the percentage of lines that are covered by tests out of the coverable lines that the pull request added or modified: <covered lines added or modified>/<coverable lines added or modified> * 100%

_{NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer
TIP This summary will be updated as you push new changes.}

[gemini-code-assist]

Code Review

This pull request implements support for the CONFLICT_IGNORE mode in gRPC insert streams, allowing the server to skip duplicate records based on specified key columns instead of failing the transaction. The changes include the addition of helper methods to check for existing records and updates to the insertion logic to track ignored rows. A new integration test verifies this behavior. Feedback from the review suggests consolidating the duplicated existence check methods into a single helper and removing redundant checks for empty key columns at the call sites.

[gemini-code-assist]

The methods keyExistsVertex and keyExistsDocument are identical in logic. They can be merged into a single helper method that accepts a Document parameter, as both MutableVertex and MutableDocument implement this interface. This reduces code duplication and improves maintainability.

  private boolean keyExists(final InsertContext ctx, final Document incoming) {
    if (ctx.keyCols.isEmpty())
      return false;
    final String where = String.join(" AND ", ctx.keyCols.stream().map(k -> k + " = ?").toList());
    final Object[] params = ctx.keyCols.stream().map(incoming::get).toArray();
    try (final ResultSet rs = ctx.db.query("sql", "SELECT FROM " + ctx.opts.getTargetClass() + " WHERE " + where, params)) {
      return rs.hasNext();
    }
  }

[robfrank]

Merged into a single keyExists(InsertContext, MutableDocument) in f78b9fa. Used MutableDocument instead of Document since MutableVertex extends MutableDocument - the call sites pass mutable instances directly.

[gemini-code-assist]

The check !ctx.keyCols.isEmpty() is redundant here because it is already performed inside the keyExists method (or the merged version suggested). Removing it simplifies the condition. Also, the call should be updated to use the merged keyExists method.

Suggested change

	} else if (ctx.opts.getConflictMode() == ConflictMode.CONFLICT_IGNORE && !ctx.keyCols.isEmpty() && keyExistsVertex(ctx, v)) {
	} else if (ctx.opts.getConflictMode() == ConflictMode.CONFLICT_IGNORE && keyExists(ctx, v)) {

[robfrank]

Dropped in f78b9fa.

[gemini-code-assist]

The check !ctx.keyCols.isEmpty() is redundant here because it is already performed inside the keyExists method (or the merged version suggested). Removing it simplifies the condition. Also, the call should be updated to use the merged keyExists method.

Suggested change

	} else if (ctx.opts.getConflictMode() == ConflictMode.CONFLICT_IGNORE && !ctx.keyCols.isEmpty() && keyExistsDocument(ctx, d)) {
	} else if (ctx.opts.getConflictMode() == ConflictMode.CONFLICT_IGNORE && keyExists(ctx, d)) {

[robfrank]

Dropped in f78b9fa.

[claude]

Code Review

Overview

This PR fixes a real regression (#4214) where InsertStream with CONFLICT_IGNORE + PER_STREAM would roll back the entire stream on a commit-time DuplicatedKeyException instead of silently skipping duplicates. The fix approach - detecting duplicates with a pre-save key lookup rather than catching a post-commit exception - is sound and mirrors the existing CONFLICT_UPDATE pattern.

---

Issues

1. Code duplication - keyExistsVertex / keyExistsDocument can be unified

These two methods are nearly identical. The only meaningful difference is the parameter type:

// keyExistsVertex
final var inDoc = incoming.asDocument();
ctx.keyCols.stream().map(inDoc::get)

// keyExistsDocument  
ctx.keyCols.stream().map(incoming::get)

Since MutableVertex extends MutableDocument, the incoming.asDocument() intermediate is unnecessary - incoming::get works directly on a vertex too. Both methods can be collapsed into one:

private boolean keyExists(final InsertContext ctx, final MutableDocument incoming) {
    if (ctx.keyCols.isEmpty())
        return false;
    final String where = String.join(" AND ", ctx.keyCols.stream().map(k -> k + " = ?").toList());
    final Object[] params = ctx.keyCols.stream().map(incoming::get).toArray();
    try (final ResultSet rs = ctx.db.query("sql", "SELECT FROM " + ctx.opts.getTargetClass() + " WHERE " + where, params)) {
        return rs.hasNext();
    }
}

Call sites become keyExists(ctx, v) and keyExists(ctx, d) (since MutableVertex IS-A MutableDocument).

2. Minor style inconsistency between the two methods

keyExistsVertex uses final var rs in the try-with-resources, while keyExistsDocument uses final ResultSet rs. The rest of the codebase (including tryUpsertVertex / tryUpsertDocument) uses explicit types. Pick one and be consistent - explicit ResultSet matches the surrounding code style better.

3. Redundant !ctx.keyCols.isEmpty() guard at call sites

} else if (ctx.opts.getConflictMode() == ConflictMode.CONFLICT_IGNORE && !ctx.keyCols.isEmpty() && keyExistsVertex(ctx, v)) {

The !ctx.keyCols.isEmpty() short-circuit is already the first thing checked inside keyExistsVertex/keyExistsDocument, so this guard is redundant. The method call overhead is negligible. Either drop the guard at the call site or add a comment explaining why the early check is intentional (e.g., "avoid object allocation from stream pipeline").

---

Potential Issue - TOCTOU under concurrent PER_STREAM inserts

The pre-save key-existence check and the subsequent save() are not atomic across transactions. If two concurrent streams both insert the same key under CONFLICT_IGNORE + PER_STREAM, both may pass keyExists(...) (neither sees the other's uncommitted data), both call save(), and the commit of one will still throw DuplicatedKeyException - landing back in recordCommitException and marking rows as failed.

This is an inherent MVCC limitation and hard to fix without retrying with row-level exclusion - the PR description doesn't claim to fix it, but it would be worth a short comment in the code (and/or issue) so future readers understand the boundary condition.

---

Performance Note

Each record processed under CONFLICT_IGNORE with non-empty key_columns now issues an extra SELECT before save(). For streams with many rows and few/no duplicates (the common case), this is an additional read per row. This mirrors the existing CONFLICT_UPDATE (tryUpsertVertex/tryUpsertDocument) overhead, so it's an accepted trade-off - but worth a brief note for anyone tuning bulk-load performance.

---

Test Coverage

The regression test is well-structured:

• Correct use of CountDownLatch + AtomicReference for async coordination
• Asserts all four counters (received, inserted, ignored, failed) and errorsList
• Verifies actual persistence with a follow-up SELECT count(*)
• Cleans up in finally
• Unique type name avoids cross-run collisions

One gap: the fix applies to both VERTEX and DOCUMENT types, but the test only covers the DOCUMENT path. A second @Test (or parameterized test) exercising VERTEX with CREATE VERTEX TYPE + CONFLICT_IGNORE + PER_STREAM would give confidence that keyExistsVertex is also exercised.

---

Summary

The fix is correct and targeted. The main actionable items before merge:

1. Unify keyExistsVertex / keyExistsDocument - straightforward refactor, reduces surface area
2. Add a vertex-type test case - low effort, closes the coverage gap
3. Document the TOCTOU boundary in a brief code comment or follow-up issue

Items 1 and 3 are style/maintenance concerns; item 2 is a minor coverage gap. None of them are blockers for correctness of the fix itself.

[robfrank]

Addressed in f78b9fa:

• Unify keyExistsVertex / keyExistsDocument - merged into a single keyExists(InsertContext, MutableDocument). MutableVertex extends MutableDocument so the vertex call site passes the mutable instance directly.
• Style consistency - merged method uses explicit ResultSet (matches tryUpsertDocument).
• Redundant !keyCols.isEmpty() guard at call sites - dropped; the method already short-circuits.
• Vertex test coverage gap - added conflictIgnorePerStreamShouldSkipDuplicateAndCommitRemainingRowsForVertices exercising the CREATE VERTEX TYPE path. 32 tests pass.

Not addressed:

• TOCTOU code comment - leaving out. The pre-save lookup + save() non-atomicity is a property of MVCC, not a hidden constraint of this fix; the same boundary holds for the existing tryUpsertVertex/tryUpsertDocument path and isn't documented there either. Per CLAUDE.md, comments should state non-obvious invariants - "two concurrent transactions can both pass a pre-save existence check" is database-101 for anyone touching this file. If we want to formalize the semantics of CONFLICT_IGNORE under concurrency, that belongs in a docs page or follow-up issue, not inline here.
• Performance note - no action requested; confirming it's the accepted trade-off mirroring CONFLICT_UPDATE.

[claude]

Code Review

Overview

This PR fixes a real bug: InsertStream with CONFLICT_IGNORE + PER_STREAM was rolling back the entire stream on a commit-time DuplicatedKeyException and reporting failed = received instead of silently skipping duplicates. The root cause is well-diagnosed and the fix is in the right place.

---

What works well

• Correct fix for the stated bug. The pre-check in keyExists prevents the duplicate row from reaching save(), so the deferred commit never sees it. Under PER_STREAM where all rows share a single transaction, the SELECT query inside that transaction correctly sees already-inserted-but-not-yet-committed rows too, so intra-stream duplicates are also caught.
• Follows the existing pattern. The approach mirrors tryUpsertVertex/tryUpsertDocument (read-before-write), keeping the code internally consistent.
• Good test coverage. Two integration tests cover both document and vertex types with the exact scenario from the bug report.
• Safety net preserved. The catch (DuplicatedKeyException dup) block (line 2322) still catches any synchronous duplicate that slips through (e.g., a concurrent stream that committed between the keyExists check and save()), so the fix does not remove that fallback.

---

Concerns

1. Residual blind spot in recordCommitException (medium)

recordCommitException does not inspect conflict_mode:

// line 1701
private static void recordCommitException(final Counts totals, final Exception e) {
    final long rolledBack = totals.inserted + totals.updated;
    totals.inserted = 0;
    totals.updated = 0;
    totals.failed += rolledBack;   // <-- ignores conflict_mode
    ...
}

After this fix, the common single-stream case is handled. But a truly concurrent duplicate - where two PER_STREAM streams both pass keyExists for the same key, one commits first, and the other hits a DuplicatedKeyException at flushCommit(true) - still triggers recordCommitException, reporting failed = received for CONFLICT_IGNORE. This is a narrower edge case than the original bug but it is a correctness gap. Consider adding a conflict_mode parameter to recordCommitException so it can bump ignored (and zero failed) when CONFLICT_IGNORE is in effect.

2. Extra SELECT per row in CONFLICT_IGNORE mode (minor - performance)

keyExists issues a SELECT FROM ... WHERE key = ? for every row, even non-duplicates. For large streams with few actual duplicates this is a per-row overhead added purely as a guard. The existing CONFLICT_UPDATE path has the same overhead via tryUpsert, so this is at least consistent. But it is worth noting in the code - a brief comment on keyExists explaining why the pre-check is needed (deferred-commit trap) would help future readers.

3. Code duplication (minor)

keyExists is tryUpsertVertex / tryUpsertDocument without the update step. The query-building and result-set iteration are identical:

String where = String.join(" AND ", keys.stream().map(k -> k + " = ?").toList());
Object[] params = keys.stream().map(incoming::get).toArray();
try (ResultSet rs = ctx.db.query("sql", "SELECT FROM " + ctx.opts.getTargetClass() + " WHERE " + where, params)) { ... }

One option is to extract a shared lookupExisting(InsertContext, MutableDocument) that returns an Optional<Result>, reused by both tryUpsert* and keyExists. Not a blocker, but it would reduce drift.

4. Test style nit

AuthClientInterceptor does not reference any outer-class state and could be private static class. The existing Issue4198InsertStreamCommitErrorIT uses the same pattern, so this is pre-existing - just noting it for consistency.

The @Override annotations on the same line as their method in the anonymous StreamObserver (@Override public void onNext(...)) differs from the style used in Issue4198InsertStreamCommitErrorIT. Minor, but may cause Prettier/formatter churn.

5. No test for PER_BATCH + CONFLICT_IGNORE

The PR summary mentions bulkInsertWithConflictIgnore (PER_BATCH) in GrpcServerIT passes, but the new IT class only covers PER_STREAM. Adding a PER_BATCH variant here would make this class a more complete regression guard for issue #4214.

---

Summary

The fix is correct and safe for the primary bug. The main actionable item is concern #1 - recordCommitException should be made aware of conflict_mode so that concurrent-stream CONFLICT_IGNORE still reports ignored rather than failed. The rest are minor or pre-existing.

[robfrank]

Addressed in 8d2c7a6:

• Bump jedis from 3.6.1 to 3.6.3 #5 PER_BATCH coverage - added conflictIgnorePerBatchShouldSkipDuplicateAndCommitRemainingRows. 3 tests pass.

Pushing back on the other items:

#1 recordCommitException should bump ignored for CONFLICT_IGNORE - the suggested simple fix has incorrect semantics. Suppose stream B has 10 rows, only one (rowC) is a true concurrent duplicate. The engine rolls back all 10 on flushCommit(true); recordCommitException is called with inserted=10. Zeroing failed and bumping ignored=10 would report 9 valid rows that the user wanted inserted as "ignored", which is wrong. A semantically correct fix needs per-row retry to identify which specific keys conflicted - meaningful new logic and out of scope for this PR. Worth filing as a follow-up if anyone hits it in practice.

#2 Comment on keyExists explaining the deferred-commit trap - the call site reads conflictMode == CONFLICT_IGNORE && keyExists(ctx, v). The "why pre-check" lives in that conditional and the surrounding tryUpsertVertex / tryUpsertDocument pattern. The method itself is a generic key lookup; adding a comment couples it to one specific caller. Per CLAUDE.md "default to no comments", skipping.

#3 Extract shared lookupExisting - out-of-scope refactor. The duplication between tryUpsert* and keyExists is real but the tryUpsert* pair predates this PR and CLAUDE.md says bug fixes shouldn't introduce surrounding cleanup. Reviewer also flagged it as non-blocking.

#4 Test style nits (static interceptor, @Override placement) - matches the existing Issue4198InsertStreamCommitErrorIT pattern by design; deliberately keeping consistency with the neighbour file rather than diverging.

[claude]

Code Review

Good fix for a real and reproducible bug. The approach is consistent with the existing CONFLICT_UPDATE pattern and the regression tests are well-structured. A few items worth discussing before merge:

---

Correctness

The core fix is sound. Pre-checking with keyExists before save() prevents the deferred DuplicatedKeyException from ever reaching flushCommit / recordCommitException, which was reclassifying the whole batch as failed.

In-stream duplicates (two rows with the same key, neither pre-existing): Covered by two mechanisms working together:

1. TransactionIndexContext.addIndexKeyLock throws DuplicatedKeyException immediately at save() time when two records with the same unique key are saved within the same transaction - so the existing catch (DuplicatedKeyException dup) block in insertRows handles this case.
2. The SQL query in keyExists routes through the index, which includes in-transaction indexChanges, so the second duplicate would also be detected by keyExists itself.

TOCTOU race (concurrent streams with the same key): Acknowledged in the PR description as acceptable, consistent with CONFLICT_UPDATE behavior.

---

Limitation worth documenting

When keyCols is empty and there are deferred unique-constraint violations, keyExists short-circuits to false and the original bug (#4198 path) still fires. This is arguably the correct semantic behavior (without key columns we cannot pre-check), but a brief comment on keyExists or in the PR description would prevent confusion if someone hits it.

---

Security - Column name injection (pre-existing, applies equally to new code)

// keyExists - and identically in tryUpsertVertex / tryUpsertDocument
final String where = String.join(" AND ", ctx.keyCols.stream().map(k -> k + " = ?").toList());

Column names and the target-class name are concatenated directly into the SQL string. Values are parameterized (safe), but column names from the gRPC request are not. A client supplying a key_column value of name) OR (1=1 -- could manipulate the query. The same issue exists in the two older upsert helpers and in the getTargetClass() concatenation.

This PR does not introduce the problem, but since it adds a third call site of the same pattern, it is a good moment to address it - either via schema validation (verify each keyCol is actually a declared property of targetClass before building the WHERE clause) or by quoting the identifiers. ArcadeDB's SQL parser likely has an identifier-quoting utility that could be used here.

---

Test coverage gap

All three test methods share the same scenario structure: one pre-existing Alice, then stream [Bob, Alice (dup), Carol]. Consider adding a fourth test for in-stream duplicates with no pre-existing row - e.g. stream [Alice, Alice] - to confirm that the immediate-throw path in TransactionIndexContext also results in ignored=1 (not failed=1) for CONFLICT_IGNORE. This would complete the coverage matrix.

---

Minor nits

Code duplication: The WHERE-clause construction pattern now appears three times (tryUpsertVertex, tryUpsertDocument, keyExists). Extracting a shared buildWhereClause(List<String> cols) helper would make future changes a single edit. Low priority.

AuthClientInterceptor: Declared as private class in both this test and the Issue4198 test. It does not reference any outer-class instance members, so private static class would be cleaner. Consistent with the existing precedent though, so not a blocker.

tryUpsertDocument parameter final keywords: The new keyExists method correctly applies final to its parameters (per project guidelines), but the neighbouring tryUpsertDocument does not. Not introduced by this PR; worth a follow-up pass.

---

Summary

The fix correctly resolves the reported bug for all tested modes (PER_STREAM documents, PER_BATCH documents, PER_STREAM vertices). The column-name injection concern and the missing in-stream-duplicate test are the two items I would prioritize before merge. Everything else is minor polish.

[codecov]

Codecov Report

❌ Patch coverage is 80.00000% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 64.64%. Comparing base (6eb94b1) to head (8d2c7a6).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
.../com/arcadedb/server/grpc/ArcadeDbGrpcService.java	80.00%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #4262   +/-   ##
=======================================
  Coverage   64.63%   64.64%           
=======================================
  Files        1644     1644           
  Lines      127277   127287   +10     
  Branches    27267    27269    +2     
=======================================
+ Hits        82266    82281   +15     
+ Misses      33509    33508    -1     
+ Partials    11502    11498    -4     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.