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.setOldEntryKey(entryKeyOf(oldRecord).pack());
+                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.setNewEntryKey(entryKeyOf(newRecord).pack());
+                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;
     }
 
     /**

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

@@ -63,8 +63,8 @@
         return maintainer.isIdempotent() && !isSyntheticIndex(state);
     }
 
     @Override
-    @Nonnull
+    @Nullable
     public <M extends Message> Any serializePendingWriteQueue(@Nullable final FDBIndexableRecord<M> oldRecord,
                                                               @Nullable final FDBIndexableRecord<M> newRecord) {
         return serializePendingWrites(state, oldRecord, newRecord);
@@ -88,9 +88,8 @@
      * @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,
+    private 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) {

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

@@ -68,7 +68,8 @@
 import com.google.protobuf.Message;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
@@ -359,7 +360,7 @@
     }
 
     @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 +377,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 +393,28 @@
         } 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 (oldIndexEntries != null && newIndexEntries != null && 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));
+        if (oldIndexEntries != null) {
+            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));
+        if (newIndexEntries != null) {
+            for (final IndexEntry entry : newIndexEntries) {
+                future = future.thenCompose(ignore -> updateIndexEntry(entry, false));
+            }
         }
         return future;
     }
@@ -415,6 +436,18 @@
                 Tuple.fromBytes(entry.getPrimaryKey().toByteArray()));
     }
 
+    @Nullable
+    private List<IndexEntry> fromProto(@Nonnull final List<IndexBuildProto.IndexEntry> protoEntries) {
+        if (protoEntries.isEmpty()) {
+            return null;
+        }
+        final List<IndexEntry> indexEntries = new ArrayList<>(protoEntries.size());
+        for (final IndexBuildProto.IndexEntry entry : protoEntries) {
+            indexEntries.add(fromProto(entry));
+        }
+        return indexEntries;
+    }
+
     @Override
     public boolean canDeleteWhere(@Nonnull final QueryToKeyMatcher matcher, @Nonnull final Key.Evaluated evaluated) {
         if (!super.canDeleteWhere(matcher, evaluated)) {