Optimize serialization for WriteOnly with PendingWritesQueue

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/ContextSessionKey.java

@@ -47,8 +47,8 @@
      * Session key for the set of write-only-with-queue index names updated in this transaction.
      * Value type: {@code Set<String>}. Returns {@code null} if no write-only-with-queue index was updated.
      * Useful for diagnosing conflicts that may happen when an index is updated by the indexer.
      */
-    public static final ContextSessionKey<Set<String>> WRITE_ONLY_WITH_QUEUE_INDEXES_UPDATED = new ContextSessionKey<>("writeOnlyIndexesUpdated");
+    public static final ContextSessionKey<Set<String>> WRITE_ONLY_WITH_QUEUE_INDEXES_UPDATED = new ContextSessionKey<>("writeOnlyWithQueueIndexesUpdated");
     /**
      * Session key for the set of readable index names updated in this transaction.
      * Note that this captures both {@link com.apple.foundationdb.record.IndexState#READABLE} and

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStore.java

@@ -780,13 +780,19 @@
                 case WRITE_ONLY_WITH_QUEUE:
                     // Push the old/new record to a write pending queue instead of updating the index directly. The
                     // ongoing online indexer will drain the queue and perform the actual index update. A maintainer that
                     // does not support the queue will throw from serializePendingWriteQueue below.
-                    future = IndexingPendingWriteQueue.enqueuePendingIndexUpdate(this, index,
-                            IndexBuildProto.PendingWritesQueueEntry.newBuilder()
-                                    .setOperation(IndexBuildProto.PendingWritesQueueEntry.Operation.UPDATE)
-                                    .setData(maintainer.serializePendingWriteQueue(oldRecord, newRecord))
-                                    .build());
-                    context.addToSessionSet(ContextSessionKey.WRITE_ONLY_WITH_QUEUE_INDEXES_UPDATED, index.getName());
+                    final Any pendingWriteData = maintainer.serializePendingWriteQueue(oldRecord, newRecord);
+                    if (pendingWriteData == null) {
+                        // Nothing to defer onto the queue.
+                        future = AsyncUtil.DONE;
+                    } else {
+                        future = IndexingPendingWriteQueue.enqueuePendingIndexUpdate(this, index,
+                                IndexBuildProto.PendingWritesQueueEntry.newBuilder()
+                                        .setOperation(IndexBuildProto.PendingWritesQueueEntry.Operation.UPDATE)
+                                        .setData(pendingWriteData)
+                                        .build());
+                        context.addToSessionSet(ContextSessionKey.WRITE_ONLY_WITH_QUEUE_INDEXES_UPDATED, index.getName());
+                    }
                     break;
 
                 case WRITE_ONLY:

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java

@@ -163,10 +163,10 @@
      *
      * @param oldRecord the previous stored record or <code>null</code> if a new record is being created
      * @param newRecord the new record or <code>null</code> if an old record is being deleted
      * @param <M> type of message
-     * @return a packed message to save in the pending write queue
+     * @return a packed message to save in the pending write queue. Null is returned if no update is needed.
      */
-    @Nonnull
+    @Nullable
     @API(API.Status.EXPERIMENTAL)
     public <M extends Message> Any serializePendingWriteQueue(@Nullable FDBIndexableRecord<M> oldRecord,
                                                               @Nullable FDBIndexableRecord<M> newRecord) {

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java

@@ -440,23 +440,33 @@
             return future;
         });
     }
 
-    @Nonnull
+    @Nullable
     @Override
     public <M extends Message> Any serializePendingWriteQueue(@Nullable final FDBIndexableRecord<M> oldRecord, @Nullable final FDBIndexableRecord<M> newRecord) {
         // The maintenance filter is applied here, at enqueue time, so a record filtered out of this index is never
         // deferred onto the queue and updateFromQueue does not need the record to re-check it.
         final IndexBuildProto.SlidingWindowQueueEntry.Builder builder =
                 IndexBuildProto.SlidingWindowQueueEntry.newBuilder();
+        boolean anyChange = false;
         if (shouldMaintain(oldRecord)) {
             builder.setOldEntryKey(entryKeyOf(oldRecord).pack());
-            builder.setDelegatedDelete(delegate.serializePendingWriteQueue(oldRecord, null));
+            final Any delegatedDelete = delegate.serializePendingWriteQueue(oldRecord, null);
+            if (delegatedDelete != null) {
+                builder.setDelegatedDelete(delegatedDelete);
+            }
+            anyChange = true;
         }
         if (shouldMaintain(newRecord)) {
             builder.setNewEntryKey(entryKeyOf(newRecord).pack());
-            builder.setDelegatedInsert(delegate.serializePendingWriteQueue(null, newRecord));
+            final Any delegatedInsert = delegate.serializePendingWriteQueue(null, newRecord);
+            if (delegatedInsert != null) {
+                builder.setDelegatedInsert(delegatedInsert);
+            }
+            anyChange = true;
         }
-        return Any.pack(builder.build());
+        // If nothing was maintained for either records, return null to indicate that no change is needed
+        return anyChange ? Any.pack(builder.build()) : null;
     }
 
     /**
@@ -486,11 +496,11 @@
         final Any delegateDelete = entry.hasDelegatedDelete() ? entry.getDelegatedDelete() : null;
         final Any delegateInsert = entry.hasDelegatedInsert() ? entry.getDelegatedInsert() : null;
         // The maintenance filter was already applied when the update was first deferred, so it is not re-evaluated here.
-        validateOrThrowEx(oldKey == null || delegateDelete != null, "old record key without delegate delete");
-        validateOrThrowEx(newKey == null || delegateInsert != null, "new record key without delegate insert");
+        validateOrThrowEx(delegateDelete == null || oldKey != null, "delegate delete without old record key");
+        validateOrThrowEx(delegateInsert == null || newKey != null, "delegate insert without new record key");
         return updateWindowWhileWriteOnly(oldKey, newKey,
-                () -> delegate.updateFromQueue(delegateDelete),
-                () -> delegate.updateFromQueue(delegateInsert));
+                () -> delegateDelete == null ? AsyncUtil.DONE : delegate.updateFromQueue(delegateDelete),
+                () -> delegateInsert == null ? AsyncUtil.DONE : delegate.updateFromQueue(delegateInsert));
     }
 
     private void validateOrThrowEx(boolean isValid, @Nonnull String msg) {

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainerWithQueue.java

@@ -63,11 +63,18 @@
         return maintainer.isIdempotent() && !isSyntheticIndex(state);
     }
 
     @Override
-    @Nonnull
-    public <M extends Message> Any serializePendingWriteQueue(@Nullable final FDBIndexableRecord<M> oldRecord,
-                                                              @Nullable final FDBIndexableRecord<M> newRecord) {
-        return serializePendingWrites(state, oldRecord, newRecord);
+    @Nullable
+    public <M extends Message> Any serializePendingWriteQueue(@Nullable FDBIndexableRecord<M> oldRecord,
+                                                              @Nullable FDBIndexableRecord<M> newRecord) {
+        final IndexBuildProto.OldAndNewRecords.Builder builder = IndexBuildProto.OldAndNewRecords.newBuilder();
+        if (oldRecord != null) {
+            builder.setOldRecords(serializePendingRecord(state, oldRecord));
+        }
+        if (newRecord != null) {
+            builder.setNewRecord(serializePendingRecord(state, newRecord));
+        }
+        return Any.pack(builder.build());
     }
 
     @Override
@@ -80,31 +87,9 @@
                 records.hasNewRecord() ? deserializePendingRecord(state, records.getNewRecord()) : null);
     }
 
-    /**
-     * Serialize an old/new record pair into the {@link Any}-packed payload deferred onto the pending write queue.
-     * @param state the maintainer state whose store serializer is used
-     * @param oldRecord the previous stored record, or {@code null} for an insert
-     * @param newRecord the new record, or {@code null} for a delete
-     * @param <M> type of message
-     * @return the packed payload to enqueue
-     */
-    @Nonnull
-    static <M extends Message> Any serializePendingWrites(@Nonnull final IndexMaintainerState state,
-                                                          @Nullable final FDBIndexableRecord<M> oldRecord,
-                                                          @Nullable final FDBIndexableRecord<M> newRecord) {
-        final IndexBuildProto.OldAndNewRecords.Builder builder = IndexBuildProto.OldAndNewRecords.newBuilder();
-        if (oldRecord != null) {
-            builder.setOldRecords(serializePendingRecord(state, oldRecord));
-        }
-        if (newRecord != null) {
-            builder.setNewRecord(serializePendingRecord(state, newRecord));
-        }
-        return Any.pack(builder.build());
-    }
-
     /**
      * Unpack the {@link IndexBuildProto.OldAndNewRecords} payload from a queue entry's data.
-     * @param data the {@link Any}-packed payload produced by {@link #serializePendingWrites}
+     * @param data the {@link Any}-packed payload produced by {@link #serializePendingWriteQueue(FDBIndexableRecord, FDBIndexableRecord)}
      * @return the unpacked old/new records
      */
     @Nonnull

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/VectorIndexMaintainer.java

@@ -358,8 +358,8 @@
         return true;
     }
 
     @Override
-    @Nonnull
+    @Nullable
     public <M extends Message> Any serializePendingWriteQueue(@Nullable final FDBIndexableRecord<M> oldRecord,
                                                               @Nullable final FDBIndexableRecord<M> newRecord) {
         // Serialize the computed index entries rather than the whole record.
@@ -376,6 +376,10 @@
             Verify.verify(newEntries.size() == 1);
             builder.addNewEntries(toProto(newEntries.get(0), newRecord.getPrimaryKey()));
         }
+        if (oldEntries == null && newEntries == null) {
+            // Both records were filtered out of this index; there is nothing to defer onto the queue.
+            return null;
+        }
         return Any.pack(builder.build());
     }
 
@@ -388,12 +392,24 @@
         } catch (InvalidProtocolBufferException ex) {
             throw new RecordCoreException("failed to parse vector index pending write queue entry data", ex);
         }
+        List<IndexEntry> oldIndexEntries = fromProto(entries.getOldEntriesList());
+        List<IndexEntry> newIndexEntries = fromProto(entries.getNewEntriesList());
+        if (skipUpdateForUnchangedKeys()) {
+            // Remove unchanged keys from the lists of keys to update, mirroring StandardIndexMaintainer.update.
+            final List<IndexEntry> commonKeys = commonKeys(oldIndexEntries, newIndexEntries);
+            if (!commonKeys.isEmpty()) {
+                oldIndexEntries = makeMutable(oldIndexEntries);
+                oldIndexEntries.removeAll(commonKeys);
+                newIndexEntries = makeMutable(newIndexEntries);
+                newIndexEntries.removeAll(commonKeys);
+            }
+        }
         CompletableFuture<Void> future = AsyncUtil.DONE;
-        for (final IndexBuildProto.IndexEntry entry : entries.getOldEntriesList()) {
-            future = future.thenCompose(ignore -> updateIndexEntry(fromProto(entry), true));
+        for (final IndexEntry entry : oldIndexEntries) {
+            future = future.thenCompose(ignore -> updateIndexEntry(entry, true));
         }
-        for (final IndexBuildProto.IndexEntry entry : entries.getNewEntriesList()) {
-            future = future.thenCompose(ignore -> updateIndexEntry(fromProto(entry), false));
+        for (final IndexEntry entry : newIndexEntries) {
+            future = future.thenCompose(ignore -> updateIndexEntry(entry, false));
         }
         return future;
     }
@@ -415,6 +431,11 @@
                 Tuple.fromBytes(entry.getPrimaryKey().toByteArray()));
     }
 
+    @Nonnull
+    private List<IndexEntry> fromProto(@Nonnull final List<IndexBuildProto.IndexEntry> protoEntries) {
+        return protoEntries.stream().map(this::fromProto).toList();
+    }
+
     @Override
     public boolean canDeleteWhere(@Nonnull final QueryToKeyMatcher matcher, @Nonnull final Key.Evaluated evaluated) {
         if (!super.canDeleteWhere(matcher, evaluated)) {

fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java

@@ -2152,6 +2152,50 @@ void writeOnlyWithQueueUpdateMovesRecordOutOfWindow() throws Exception {
         }
     }
 
+    @Test
+    void writeOnlyWithQueueUpdateOfWindowValueOnlyRefilesEntry() throws Exception {
+        // Test the case of record change that affects the window, but not the delegate
+        try (FDBRecordContext context = openContext()) {
+            openStore(context, 3, Direction.DESC);
+            rec(1, 100);   // the boundary: the worst of three entries in a window of 3
+            rec(2, 200);
+            rec(3, 300);
+            assertThat(slidingWindow()).hasSizeOf(3).underlyingHnsw().containsInAnyOrder(1, 2, 3);
+            commit(context);
+        }
+
+        try (FDBRecordContext context = openContext()) {
+            openStore(context, 3, Direction.DESC);
+            recordStore.markIndexWriteOnlyWithQueue(INDEX_NAME).join();
+            rec(1, 250);  // update: 100 -> 250, same vector, deferred to the queue
+            commit(context);
+        }
+
+        drainQueue(3, Direction.DESC);
+
+        try (FDBRecordContext context = openContext()) {
+            openStore(context, 3, Direction.DESC);
+            assertThat(slidingWindow())
+                    .hasSizeOf(3)
+                    .underlyingHnsw().containsInAnyOrder(1, 2, 3);
+            commit(context);
+        }
+
+        // The count and the delegate look identical whether the entry was re-filed, so probe the boundary
+        // instead: rec 2 at 200 is now the worst entry in the window, so a new record at 220 has to evict it. Had
+        // rec 1's entry been left behind at 100, rec 1 would still be the boundary and would have been evicted.
+        try (FDBRecordContext context = openContext()) {
+            openStore(context, 3, Direction.DESC);
+            assertTrue(recordStore.isIndexReadable(index()));
+            rec(4, 220);
+            assertThat(slidingWindow())
+                    .as("the boundary must have moved to rec 2 when rec 1 was re-filed at its higher window value")
+                    .hasSizeOf(3)
+                    .underlyingHnsw().containsInAnyOrder(1, 3, 4);
+            commit(context);
+        }
+    }
+
     @Test
     void writeOnlyWithQueueGroupedRoutesPerGroup() throws Exception {
         // Each partition maintains its own window when writes are deferred and later drained.