Merge a delta into an Index without visiting entries only the receiver holds (#786)

Index.mergeDifferences calls the merge function for every key present on
only one side, so merging a one-entry index into a large one walked the
whole receiver: 100,001 calls at 100,000 entries, resolving every child.

Index.merge(Index, MergeFunction) is the function-taking form of
AMap.merge: for each key of the other index it applies the function to the
receiver's value (or null) and the other's, and keeps keys only the
receiver holds as they are without calling it. It reuses the structural
merge through a private marker on the function, so MergeFunction is
unchanged; sub-tries off the paths to the merged keys are shared without
being traversed. The stack-safe fallback walks only the other index.

RowLWW.mergeRows uses it when the incoming side carries only rows, as a row
delivery does, where the creation check cannot affect rows the table holds,
and the live count is then adjusted from the delivered keys instead of
recounting the table. Tests check the contract against a naive oracle, one
call for one merged key into 10,000 entries, unreached sub-tries never
read, the deep fallback, and a row delivery into a table whose other rows
are not loaded.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: mikera

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Let applications choose how each Convex DB database merges, with per-row last-write-wins provided as one policy. ([Merge Policy](convex-db/docs/LATTICE-DB-ARCHITECTURE.md#merge-policy), [#779](https://github.com/Convex-Dev/convex/issues/779), [#784](https://github.com/Convex-Dev/convex/issues/784))
 - Let applications supply the clock that stamps Convex DB writes. ([Clock](convex-db/docs/LATTICE-DB-ARCHITECTURE.md#clock), [#781](https://github.com/Convex-Dev/convex/issues/781))
+- Merge a small delta into a large `Index` without visiting the rest of it, with `Index.merge`; Convex DB row deliveries now cost the rows delivered, not the table. ([#786](https://github.com/Convex-Dev/convex/issues/786))
 
 ### Changed
 

convex-core/src/main/java/convex/core/data/Index.java

@@ -1489,6 +1489,59 @@ public Index<K, V> mergeDifferences(Index<K, V> b, MergeFunction<V> func) {
 		}
 	}
 
+	/**
+	 * Merges another index into this one with a merge function, as {@link AMap#merge(AMap)}
+	 * merges entries. For every key in {@code b}, the result holds {@code func} applied to
+	 * this index's value (or {@code null}) and {@code b}'s value: a {@code null} result
+	 * removes the key, and a key present in both with equal values is kept without calling
+	 * {@code func}. Keys present only in this index are kept as they are and {@code func}
+	 * is not called for them.
+	 *
+	 * <p>Sub-tries of this index off the paths to {@code b}'s keys are shared unchanged
+	 * without being traversed: only nodes on those paths are rebuilt, reading their direct
+	 * children. The cost is therefore proportional to {@code b} and the key depth, not to
+	 * the size of this index. {@link #mergeDifferences(Index, MergeFunction)} instead
+	 * applies {@code func} to every key present on only one side.</p>
+	 *
+	 * <p>Identity contract as for {@link #mergeDifferences(Index, MergeFunction)}: returns
+	 * {@code this} by reference identity whenever the merged result equals {@code this}.</p>
+	 *
+	 * @param b Index whose entries are merged into this one
+	 * @param func Merge function for Index values
+	 * @return Merged Index, or {@code this} by reference identity if the result is unchanged
+	 */
+	public Index<K, V> merge(Index<K, V> b, MergeFunction<V> func) {
+		try {
+			return mergeNode(this, b, new KeepOwn<>(func));
+		} catch (StackOverflowError e) {
+			// Fully unwound, as for mergeDifferences
+			return mergeIntoDeep(this, b, func);
+		}
+	}
+
+	/**
+	 * Marks a merge that keeps entries present only in its left (own) operand as they are,
+	 * without calling the wrapped function. Only {@link #merge(Index, MergeFunction)}
+	 * creates one; the structural merge recognises it in place of a flag.
+	 */
+	private static final class KeepOwn<V> implements MergeFunction<V> {
+		private final MergeFunction<V> func;
+
+		KeepOwn(MergeFunction<V> func) {
+			this.func = func;
+		}
+
+		@Override
+		public V merge(V a, V b) {
+			return func.merge(a, b);
+		}
+
+		@Override
+		public V merge(Object key, V a, V b) {
+			return func.merge(key, a, b);
+		}
+	}
+
 	/**
 	 * Gets the child Ref for a specific digit, or null if not found.
 	 */
@@ -1543,6 +1596,7 @@ private static <K extends ABlobLike<?>, V extends ACell> Index<K, V> mergeAligne
 			if (aref == null) {
 				nc = applySide(bref.getValue(), func, false);      // b-only child
 			} else if (bref == null) {
+				if (func instanceof KeepOwn) continue;              // a-only child kept, without traversing it
 				ac = aref.getValue();
 				nc = applySide(ac, func, true);                    // a-only child
 			} else if (aref.equals(bref)) {
@@ -1576,7 +1630,8 @@ private static <K extends ABlobLike<?>, V extends ACell> Index<K, V> mergeNested
 		MapEntry<K, V> ne = singleEntry(shallow.entry, func, shallowIsLeft); // shallow entry is single-side
 		Index<K, V>[] kids = null; // built lazily on first change
 		boolean changed = (ne != shallow.entry);
-		int sm = shallow.mask & 0xFFFF;
+		// The shallow node's other children are single-side; an own side that is kept is not traversed
+		int sm = (shallowIsLeft && func instanceof KeepOwn) ? 0 : shallow.mask & 0xFFFF;
 		for (int d = 0; d < 16; d++) {
 			if (d == dig) continue;
 			if (((sm >> d) & 1) == 0) continue;
@@ -1620,9 +1675,13 @@ private static <K extends ABlobLike<?>, V extends ACell> Index<K, V> branchDisjo
 		return new Index<K, V>(m, null, refs, (short) mask, A.count + B.count);
 	}
 
-	/** Apply {@code func} to every entry of a node as a single side (other side null). Returns the same node if unchanged. */
+	/**
+	 * Apply {@code func} to every entry of a node as a single side (other side null). Returns the
+	 * same node if unchanged, and at once for an own side that a {@link KeepOwn} merge keeps.
+	 */
 	private static <K extends ABlobLike<?>, V extends ACell> Index<K, V> applySide(Index<K, V> node, MergeFunction<V> func, boolean left) {
 		if (node.count == 0) return node;
+		if (left && func instanceof KeepOwn) return node;
 		MapEntry<K, V> ne = singleEntry(node.entry, func, left);
 		Index<K, V>[] kids = null;
 		int m = node.mask & 0xFFFF;
@@ -1689,6 +1748,34 @@ static <K extends ABlobLike<?>, V extends ACell> Index<K,V> mergeDeep(
 		return result;
 	}
 
+	/**
+	 * Merge of {@code b} into {@code a} with no recursion, as {@link #merge(Index, MergeFunction)},
+	 * used once the structural merge has overflowed the thread stack. Walks {@code b} only,
+	 * looking up each key in {@code a}, so entries only {@code a} holds are neither visited nor
+	 * passed to {@code func}. Preserves the identity contract. Package-private so tests can
+	 * reach it directly.
+	 */
+	static <K extends ABlobLike<?>, V extends ACell> Index<K,V> mergeIntoDeep(
+			Index<K,V> a, Index<K,V> b, MergeFunction<V> func) {
+		Index<K,V> result=a;
+		long n=b.count;
+		for (long i=0; i<n; i++) {
+			MapEntry<K,V> be=b.entryAt(i);
+			MapEntry<K,V> ae=a.getEntry(be.getKey());
+			K key=(ae==null) ? be.getKey() : ae.getKey();
+			V av=(ae==null) ? null : ae.getValue();
+			V bv=be.getValue();
+			if (Utils.equals(av,bv)) continue;
+			V nv=func.merge(key,av,bv);
+			if (nv==null) {
+				if (ae!=null) result=result.dissocDeep(key);
+			} else if (!Utils.equals(av,nv)) {
+				result=result.assocEntryDeep(MapEntry.create(key,nv),0);
+			}
+		}
+		return result;
+	}
+
 	/** Snapshot a node's children into a digit-indexed array of size 16 (absent digits are null). */
 	@SuppressWarnings("unchecked")
 	private static <K extends ABlobLike<?>, V extends ACell> Index<K, V>[] collectKids(Index<K, V> node) {
@@ -1746,6 +1833,7 @@ private static <K extends ABlobLike<?>, V extends ACell> MapEntry<K, V> mergeEnt
 	/** Apply {@code func} to a single-side entry (other side null). Returns the entry, or null if removed. */
 	private static <K extends ABlobLike<?>, V extends ACell> MapEntry<K, V> singleEntry(MapEntry<K, V> e, MergeFunction<V> func, boolean left) {
 		if (e == null) return null;
+		if (left && func instanceof KeepOwn) return e;
 		V v = e.getValue();
 		V nv = left ? func.merge(e.getKey(), v, null) : func.merge(e.getKey(), null, v);
 		if (nv == null) return null;

convex-core/src/test/java/convex/core/data/IndexMergeTest.java

@@ -275,6 +275,117 @@ public void testNoOpNestedCase() {
 		}
 	}
 
+	// ---- merge: entries only the receiver holds are kept without calling the function ----
+
+	/** Reference for {@link Index#merge(Index, MergeFunction)}: each key of b in turn; keys only in a untouched. */
+	static Index<ABlob, CVMLong> naiveInto(Index<ABlob, CVMLong> a, Index<ABlob, CVMLong> b, MergeFunction<CVMLong> func) {
+		Index<ABlob, CVMLong> result = a;
+		long nb = b.count();
+		for (long i = 0; i < nb; i++) {
+			MapEntry<ABlob, CVMLong> me = b.entryAt(i);
+			MapEntry<ABlob, CVMLong> ae = a.getEntry(me.getKey());
+			CVMLong av = (ae == null) ? null : ae.getValue();
+			if (Utils.equals(av, me.getValue())) continue;
+			CVMLong nv = func.merge(me.getKey(), av, me.getValue());
+			if (nv == null) result = result.dissoc(me.getKey());
+			else if (!Utils.equals(av, nv)) result = result.assoc(me.getKey(), nv);
+		}
+		return result;
+	}
+
+	@Test
+	public void testMergeMatchesNaive() {
+		for (int seed = 3000; seed < 3300; seed++) {
+			Random r = new Random(seed);
+			Index<ABlob, CVMLong> a = Index.none();
+			Index<ABlob, CVMLong> b = Index.none();
+			int np = 1 + r.nextInt(40);
+			for (int i = 0; i < np; i++) {
+				ABlob k = randomKey(r);
+				CVMLong va = CVMLong.create(r.nextInt(100));
+				CVMLong vb = CVMLong.create(r.nextInt(100));
+				switch (r.nextInt(4)) { // 0:a-only 1:b-only 2:both-equal 3:both-different
+					case 0 -> a = a.assoc(k, va);
+					case 1 -> b = b.assoc(k, vb);
+					case 2 -> { a = a.assoc(k, va); b = b.assoc(k, va); }
+					default -> { a = a.assoc(k, va); b = b.assoc(k, vb); }
+				}
+			}
+			for (MergeFunction<CVMLong> f : FUNCS) {
+				String ctx = "seed=" + seed + " func=" + FUNCS.indexOf(f);
+				Index<ABlob, CVMLong> exp = naiveInto(a, b, f);
+				Index<ABlob, CVMLong> got = a.merge(b, f);
+				assertEquals(exp, got, ctx);
+				assertEquals(exp.getHash(), got.getHash(), ctx);
+				assertEquals(exp, Index.mergeIntoDeep(a, b, f), "deep " + ctx);
+				try {
+					got.validate();
+				} catch (InvalidDataException e) {
+					fail(ctx + " produced non-canonical Index: " + e.getMessage());
+				}
+
+				// Merging a subset, or nothing, is a no-op for every function: a's own entries are not touched
+				assertSame(a, a.merge(subsetOf(a, r), f), "subset " + ctx);
+				assertSame(a, a.merge(Index.none(), f), "empty " + ctx);
+				assertSame(a, Index.mergeIntoDeep(a, subsetOf(a, r), f), "deep subset " + ctx);
+			}
+		}
+	}
+
+	private static ABlob longKey(long i) {
+		return Blob.fromHex(String.format("%016x", i));
+	}
+
+	/** The function is called once per key of the merged index, however large the receiver. */
+	@Test
+	public void testMergeCallsFunctionOnlyForMergedKeys() {
+		Index<ABlob, CVMLong> a = Index.none();
+		for (long i = 0; i < 10_000; i++) a = a.assoc(longKey(i), CVMLong.create(i));
+		long[] calls = {0};
+		MergeFunction<CVMLong> counting = (x, y) -> { calls[0]++; return y; };
+
+		Index<ABlob, CVMLong> added = a.merge(Index.of(longKey(20_000), CVMLong.ONE), counting);
+		assertEquals(1, calls[0]);
+		assertEquals(10_001, added.count());
+
+		calls[0] = 0;
+		Index<ABlob, CVMLong> updated = a.merge(Index.of(longKey(5_000), CVMLong.create(-1)), counting);
+		assertEquals(1, calls[0]);
+		assertEquals(CVMLong.create(-1), updated.get(longKey(5_000)));
+		assertEquals(10_000, updated.count());
+
+		calls[0] = 0;
+		assertSame(a, a.merge(Index.of(longKey(7), CVMLong.create(7)), counting), "an equal value needs no call");
+		assertEquals(0, calls[0]);
+	}
+
+	/**
+	 * Sub-tries of the receiver off the paths to the merged keys are not traversed. Rebuilding a
+	 * node on a path reads its direct children, so the sibling here is resident, but its own
+	 * children are hash-only and would fail if resolved.
+	 */
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	public void testMergeDoesNotTraverseUnreachedSubtries() {
+		Index<ABlob, CVMLong> g1 = Index.of(Blob.fromHex("ab1010"), CVMLong.create(11));
+		Index<ABlob, CVMLong> g2 = Index.of(Blob.fromHex("ab1020"), CVMLong.create(12));
+		Ref<Index<ABlob, CVMLong>>[] siblingChildren = new Ref[] { missingChildRef(g1), missingChildRef(g2) };
+		Index<ABlob, CVMLong> sibling = Index.unsafeCreate(4, MapEntry.create(Blob.fromHex("ab10"), CVMLong.ONE),
+			(Ref[]) siblingChildren, (1 << 1) | (1 << 2), 3);
+		Index<ABlob, CVMLong> resident = Index.of(Blob.fromHex("ab20"), CVMLong.create(2));
+		Ref<Index<ABlob, CVMLong>>[] rootChildren = new Ref[] { sibling.getRef(), resident.getRef() };
+		Index<ABlob, CVMLong> a = Index.unsafeCreate(2, MapEntry.create(Blob.fromHex("ab"), CVMLong.ZERO),
+			(Ref[]) rootChildren, (1 << 1) | (1 << 2), 5);
+
+		for (MergeFunction<CVMLong> f : FUNCS) {
+			Index<ABlob, CVMLong> merged = a.merge(Index.of(Blob.fromHex("ab25"), CVMLong.create(5)), f);
+			assertEquals(6, merged.count());
+			assertEquals(CVMLong.ZERO, merged.get(Blob.fromHex("ab")), "own entries are kept, even by DROP_EVEN");
+		}
+		// A key under no existing child
+		assertEquals(6, a.merge(Index.of(Blob.fromHex("ab35"), CVMLong.create(3)), PICK_B).count());
+	}
+
 	@Test
 	public void testDisjointAndNested() {
 		// Disjoint prefixes (different first digit).

convex-core/src/test/java/convex/core/data/IndexTest.java

@@ -1325,6 +1325,11 @@ public void testDeepVariantsMatchRecursive() {
 			assertEquals(m.mergeDifferences(changed,(a,b) -> b),Index.mergeDeep(m,changed,(a,b) -> b));
 			assertEquals(m.mergeDifferences(changed,(a,b) -> (a==null)?b:a),Index.mergeDeep(m,changed,(a,b) -> (a==null)?b:a));
 			assertEquals(m.mergeDifferences(changed,(a,b) -> null),Index.mergeDeep(m,changed,(a,b) -> null));
+
+			assertSame(m,Index.mergeIntoDeep(m,m,(a,b) -> b));
+			assertEquals(m.merge(changed,(a,b) -> b),Index.mergeIntoDeep(m,changed,(a,b) -> b));
+			assertEquals(m.merge(changed,(a,b) -> (a==null)?b:a),Index.mergeIntoDeep(m,changed,(a,b) -> (a==null)?b:a));
+			assertEquals(m.merge(changed,(a,b) -> null),Index.mergeIntoDeep(m,changed,(a,b) -> null));
 		}
 	}
 }

convex-db/src/main/java/convex/db/lattice/RowLWW.java

@@ -149,6 +149,9 @@ public static AVector<ACell> mergeTable(AVector<ACell> own, AVector<ACell> other
 			liveCount = SQLTable.getLiveCount(own);
 		} else if (mergedRows == rowsB) {
 			liveCount = SQLTable.getLiveCount(other);
+		} else if (timeB == null && hasLiveCount(own)) {
+			// Delivered rows only: adjust the count held without reading the rest of the table
+			liveCount = SQLTable.getLiveCount(own) + liveChange(rowsA, mergedRows, rowsB);
 		} else {
 			// Rows actually merged — recompute
 			liveCount = SQLTable.computeLiveCount(mergedRows);
@@ -178,6 +181,10 @@ public static AVector<ACell> mergeTable(AVector<ACell> own, AVector<ACell> other
 	 * at or after the other side's utime: written before that table was created, it belongs
 	 * to an earlier table of the same name, or before it was dropped, to the dropped table.
 	 *
+	 * <p>When {@code other} carries only rows, as a delivery below an entry does, every row
+	 * held only by {@code own} survives, so those rows are not visited: the cost is
+	 * proportional to the delivered rows, not to the table.</p>
+	 *
 	 * @param own Established rows; null if absent
 	 * @param other Rows to merge in; null if absent
 	 * @param ownTime Utime of the entry holding {@code own}; null if it carries only rows
@@ -189,14 +196,33 @@ public static Index<ABlob, AVector<ACell>> mergeRows(Index<ABlob, AVector<ACell>
 		if (own == null && other == null) return null;
 		Index<ABlob, AVector<ACell>> base = (own == null) ? Index.none() : own;
 		Index<ABlob, AVector<ACell>> incoming = (other == null) ? Index.none() : other;
-		return base.mergeDifferences(incoming, new MergeFunction<AVector<ACell>>() {
+		MergeFunction<AVector<ACell>> rule = new MergeFunction<AVector<ACell>>() {
 			@Override
 			public AVector<ACell> merge(AVector<ACell> a, AVector<ACell> b) {
 				if (a == null) return writtenSince(b, ownTime) ? b : null;
 				if (b == null) return writtenSince(a, otherTime) ? a : null;
 				return mergeRow(a, b);
 			}
+		};
+		return (otherTime == null) ? base.merge(incoming, rule) : base.mergeDifferences(incoming, rule);
+	}
+
+	/** Whether an entry holds a live row count, which entries written before the field existed lack. */
+	private static boolean hasLiveCount(AVector<ACell> entry) {
+		return entry.count() > SQLTable.POS_LIVE_COUNT && entry.get(SQLTable.POS_LIVE_COUNT) instanceof CVMLong;
+	}
+
+	/** The change in live rows from merging {@code delivered} into {@code before}, giving {@code after}. */
+	private static long liveChange(Index<ABlob, AVector<ACell>> before, Index<ABlob, AVector<ACell>> after,
+			Index<ABlob, AVector<ACell>> delivered) {
+		if (delivered == null) return 0;
+		long[] change = {0};
+		delivered.forEach((key, row) -> {
+			boolean was = (before != null) && SQLRow.isLive(before.get(key));
+			boolean is = (after != null) && SQLRow.isLive(after.get(key));
+			change[0] += (is ? 1 : 0) - (was ? 1 : 0);
 		});
+		return change[0];
 	}
 
 	/** Whether a row was written at or after a table's utime; always so when there is none. */

convex-db/src/test/java/convex/db/lattice/RowLWWTest.java

@@ -13,10 +13,15 @@
 import convex.core.data.ABlob;
 import convex.core.data.ACell;
 import convex.core.data.AVector;
+import convex.core.data.Blob;
 import convex.core.data.Index;
+import convex.core.data.MapEntry;
+import convex.core.data.Ref;
+import convex.core.data.RefSoft;
 import convex.core.data.Strings;
 import convex.core.data.Vectors;
 import convex.core.data.prim.CVMLong;
+import convex.core.store.NullStore;
 
 /**
  * The per-row last-write-wins policy. Rows merge by key whichever schema wins, and a row
@@ -142,6 +147,41 @@ public void testRowDeliveryMergesAsWholeEntry() {
 		assertNull(RowLWW.mergeTable(null, rowsOnly(fresh)), "an absent table takes nothing");
 	}
 
+	/**
+	 * A row delivery reads only the delivered rows' paths. Rows elsewhere in the table are
+	 * hash-only here and would fail if read, as a persisted table would load them.
+	 */
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	public void testRowDeliveryDoesNotReadTheRestOfTheTable() {
+		ABlob nearKey = Blob.fromHex("ab10");
+		Index<ABlob, AVector<ACell>> g1 = Index.of(Blob.fromHex("ab1010"), row("g1", 2));
+		Index<ABlob, AVector<ACell>> g2 = Index.of(Blob.fromHex("ab1020"), row("g2", 2));
+		Ref[] unread = { notLoaded(g1), notLoaded(g2) };
+		Index<ABlob, AVector<ACell>> sibling = Index.unsafeCreate(4, MapEntry.create(nearKey, row("near", 2)),
+			unread, (1 << 1) | (1 << 2), 3);
+		Index<ABlob, AVector<ACell>> resident = Index.of(Blob.fromHex("ab20"), row("resident", 2));
+		Index<ABlob, AVector<ACell>> rows = Index.unsafeCreate(2, MapEntry.create(Blob.fromHex("ab"), row("root", 2)),
+			new Ref[] { sibling.getRef(), resident.getRef() }, (1 << 1) | (1 << 2), 5);
+		AVector<ACell> table = Vectors.of(SCHEMA, rows, CVMLong.create(1), CVMLong.create(5));
+
+		// A new row, a newer version of a held row, and a deletion
+		AVector<ACell> added = RowLWW.mergeTable(table, rowsOnly(Index.of(Blob.fromHex("ab25"), row("new", 3))));
+		assertEquals(6, SQLTable.getLiveCount(added));
+		AVector<ACell> updated = RowLWW.mergeTable(table, rowsOnly(Index.of(Blob.fromHex("ab20"), row("newer", 3))));
+		assertEquals(5, SQLTable.getLiveCount(updated));
+		assertEquals(row("newer", 3), rows(updated).get(Blob.fromHex("ab20")));
+		AVector<ACell> deleted = RowLWW.mergeTable(table, rowsOnly(Index.of(Blob.fromHex("ab20"), SQLRow.createTombstone(CVMLong.create(3)))));
+		assertEquals(4, SQLTable.getLiveCount(deleted));
+		// A row from before the table's creation is refused, still without reading the rest
+		assertEquals(table, RowLWW.mergeTable(table, rowsOnly(Index.of(Blob.fromHex("ab25"), row("old", 0)))));
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	private static Ref notLoaded(Index<ABlob, AVector<ACell>> node) {
+		return RefSoft.createForHash(node.getHash(), NullStore.INSTANCE).markEmbedded(false);
+	}
+
 	// ========== Lattice laws ==========
 
 	/**