[server] Config-gate collection-merge element replacement for order:ignore fields (linkedin#2846)

Add config-gated element replacement for A/A collection merge                  

During Active/Active write-compute collection merge, SET_UNION on an array    
field only advanced the element RMD timestamp when the incoming element        
compared equal to an existing one. For records with order:ignore fields, this  
silently dropped incoming content in ignored fields while still reporting the  
write as partially updated.                                                    

Add element replacement in SortBasedCollectionFieldOpHandler behind the new    
server config:                                                                 
server.aa.collection.field.element.replacement.enabled, default false.         

When enabled, newer timestamps replace the stored element with the incoming    
one. Equal timestamps use a full element comparison, including order:ignore    
fields, to deterministically break ties and keep A/A colos converged.          

Thread the config through the A/A ingestion and write-compute path, and add    
unit and integration tests for array elements with order:ignore fields.

Author: sixpluszero

clients/da-vinci-client/src/main/java/com/linkedin/davinci/config/VeniceServerConfig.java

@@ -66,6 +66,7 @@
 import static com.linkedin.venice.ConfigKeys.PUBSUB_TOPIC_MANAGER_METADATA_FETCHER_CONSUMER_POOL_SIZE;
 import static com.linkedin.venice.ConfigKeys.PUBSUB_TOPIC_MANAGER_METADATA_FETCHER_THREAD_POOL_SIZE;
 import static com.linkedin.venice.ConfigKeys.ROUTER_PRINCIPAL_NAME;
+import static com.linkedin.venice.ConfigKeys.SERVER_AA_COLLECTION_FIELD_ELEMENT_REPLACEMENT_ENABLED;
 import static com.linkedin.venice.ConfigKeys.SERVER_AA_WC_INGESTION_STORAGE_LOOKUP_THREAD_POOL_SIZE;
 import static com.linkedin.venice.ConfigKeys.SERVER_AA_WC_WORKLOAD_PARALLEL_PROCESSING_ENABLED;
 import static com.linkedin.venice.ConfigKeys.SERVER_AA_WC_WORKLOAD_PARALLEL_PROCESSING_THREAD_POOL_SIZE;
@@ -481,6 +482,12 @@ public class VeniceServerConfig extends VeniceClusterConfig {
 
   private final boolean computeFastAvroEnabled;
 
+  /**
+   * Whether to replace an existing collection-merge array element with the incoming element on a conflict, instead of
+   * only advancing its replication-metadata timestamp. See SERVER_AA_COLLECTION_FIELD_ELEMENT_REPLACEMENT_ENABLED.
+   */
+  private final boolean activeActiveCollectionFieldElementReplacementEnabled;
+
   private final long participantMessageConsumptionDelayMs;
 
   /**
@@ -911,6 +918,8 @@ public VeniceServerConfig(VeniceProperties serverProperties, Map<String, Map<Str
     storeVersionMetadataWaitDuringStateTransitionTimeMs =
         serverProperties.getLong(SERVER_STORE_VERSION_METADATA_WAIT_DURING_STATE_TRANSITION_TIME_MS, 300_000);
     computeFastAvroEnabled = serverProperties.getBoolean(SERVER_COMPUTE_FAST_AVRO_ENABLED, true);
+    activeActiveCollectionFieldElementReplacementEnabled =
+        serverProperties.getBoolean(SERVER_AA_COLLECTION_FIELD_ELEMENT_REPLACEMENT_ENABLED, false);
     participantMessageConsumptionDelayMs = serverProperties.getLong(PARTICIPANT_MESSAGE_CONSUMPTION_DELAY_MS, 60000);
     serverPromotionToLeaderReplicaDelayMs =
         TimeUnit.SECONDS.toMillis(serverProperties.getLong(SERVER_PROMOTION_TO_LEADER_REPLICA_DELAY_SECONDS, 300));
@@ -1565,6 +1574,10 @@ public boolean isComputeFastAvroEnabled() {
     return computeFastAvroEnabled;
   }
 
+  public boolean isActiveActiveCollectionFieldElementReplacementEnabled() {
+    return activeActiveCollectionFieldElementReplacementEnabled;
+  }
+
   public long getParticipantMessageConsumptionDelayMs() {
     return participantMessageConsumptionDelayMs;
   }

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/ActiveActiveStoreIngestionTask.java

@@ -175,7 +175,8 @@ public ActiveActiveStoreIngestionTask(
             rmdSerDe,
             getStoreName(),
             isWriteComputationEnabled,
-            getServerConfig().isComputeFastAvroEnabled());
+            getServerConfig().isComputeFastAvroEnabled(),
+            getServerConfig().isActiveActiveCollectionFieldElementReplacementEnabled());
     this.remoteIngestionRepairService = builder.getRemoteIngestionRepairService();
     this.reusableObjectsSupplier = Objects.requireNonNull(builder.getReusableObjectsSupplier());
 

clients/da-vinci-client/src/main/java/com/linkedin/davinci/replication/merge/MergeConflictResolverFactory.java

@@ -23,12 +23,30 @@ public MergeConflictResolver createMergeConflictResolver(
       String storeName,
       boolean rmdUseFieldLevelTs,
       boolean fastAvroEnabled) {
+    return createMergeConflictResolver(
+        annotatedReadOnlySchemaRepository,
+        rmdSerDe,
+        storeName,
+        rmdUseFieldLevelTs,
+        fastAvroEnabled,
+        false);
+  }
+
+  public MergeConflictResolver createMergeConflictResolver(
+      StringAnnotatedStoreSchemaCache annotatedReadOnlySchemaRepository,
+      RmdSerDe rmdSerDe,
+      String storeName,
+      boolean rmdUseFieldLevelTs,
+      boolean fastAvroEnabled,
+      boolean collectionFieldElementReplacementEnabled) {
     MergeRecordHelper mergeRecordHelper = new CollectionTimestampMergeRecordHelper();
     return new MergeConflictResolver(
         annotatedReadOnlySchemaRepository,
         storeName,
         valueSchemaID -> new GenericData.Record(rmdSerDe.getRmdSchema(valueSchemaID)),
-        new MergeGenericRecord(new WriteComputeProcessor(mergeRecordHelper), mergeRecordHelper),
+        new MergeGenericRecord(
+            new WriteComputeProcessor(mergeRecordHelper, collectionFieldElementReplacementEnabled),
+            mergeRecordHelper),
         new MergeByteBuffer(),
         new MergeResultValueSchemaResolverImpl(annotatedReadOnlySchemaRepository, storeName),
         rmdSerDe,

clients/da-vinci-client/src/main/java/com/linkedin/davinci/schema/merge/AvroCollectionElementComparator.java

@@ -17,10 +17,22 @@
  */
 @ThreadSafe
 public class AvroCollectionElementComparator {
-  public final static AvroCollectionElementComparator INSTANCE = new AvroCollectionElementComparator();
+  public final static AvroCollectionElementComparator INSTANCE = new AvroCollectionElementComparator(false);
 
-  private AvroCollectionElementComparator() {
-    // Singleton class.
+  /**
+   * Unlike {@link #INSTANCE}, this comparator also compares record fields marked with {@code order: ignore}. It yields a
+   * total order over elements that {@link #INSTANCE} would consider equal because they only differ in ignored fields,
+   * which is required to deterministically break ties between two same-timestamp collection elements during
+   * Active/Active conflict resolution.
+   */
+  public final static AvroCollectionElementComparator FULL_COMPARISON_INSTANCE =
+      new AvroCollectionElementComparator(true);
+
+  private final boolean compareOrderIgnoredFields;
+
+  private AvroCollectionElementComparator(boolean compareOrderIgnoredFields) {
+    // Private constructor: instances are exposed only through the shared INSTANCE and FULL_COMPARISON_INSTANCE fields.
+    this.compareOrderIgnoredFields = compareOrderIgnoredFields;
   }
 
   /**
@@ -55,7 +67,7 @@ public int compare(Object o1, Object o2, Schema schema) {
 
   private int compareRecords(GenericRecord r1, GenericRecord r2, Schema schema) {
     for (Schema.Field field: schema.getFields()) {
-      if (field.order() == Schema.Field.Order.IGNORE) {
+      if (!compareOrderIgnoredFields && field.order() == Schema.Field.Order.IGNORE) {
         continue;
       }
       int pos = field.pos();

clients/da-vinci-client/src/main/java/com/linkedin/davinci/schema/merge/SortBasedCollectionFieldOpHandler.java

@@ -27,8 +27,22 @@
 
 @ThreadSafe
 public class SortBasedCollectionFieldOpHandler extends CollectionFieldOperationHandler {
+  /**
+   * When true, a collection-merge (SET_UNION) element that wins a conflict against an existing, comparison-equal element
+   * replaces the stored element (so content in {@code order: ignore} fields is propagated) rather than only advancing the
+   * existing element's replication-metadata timestamp. See ConfigKeys#SERVER_AA_COLLECTION_FIELD_ELEMENT_REPLACEMENT_ENABLED.
+   */
+  private final boolean elementReplacementEnabled;
+
   public SortBasedCollectionFieldOpHandler(AvroCollectionElementComparator elementComparator) {
+    this(elementComparator, false);
+  }
+
+  public SortBasedCollectionFieldOpHandler(
+      AvroCollectionElementComparator elementComparator,
+      boolean elementReplacementEnabled) {
     super(elementComparator);
+    this.elementReplacementEnabled = elementReplacementEnabled;
   }
 
   @Override
@@ -589,8 +603,29 @@ private UpdateResultStatus handleModifyCollectionMergeList(
         activeElementToTsMap.put(toAddElement, modifyTimestamp);
         updated = true;
       } else if (activeTimestamp < modifyTimestamp) {
+        if (elementReplacementEnabled) {
+          // The incoming element wins on a strictly newer timestamp. Drop the stored element first so the incoming one
+          // (which may carry different content in order:ignore fields) replaces it, rather than IndexedHashMap#put
+          // keeping the existing key and only updating its timestamp.
+          activeElementToTsMap.remove(toAddElement);
+        }
         activeElementToTsMap.put(toAddElement, modifyTimestamp);
         updated = true;
+      } else if (elementReplacementEnabled && activeTimestamp == modifyTimestamp) {
+        // Same timestamp: break the tie deterministically using full element content (including order:ignore fields),
+        // so all Active/Active colos converge on the same element. A negative index means the element was removed above
+        // as part of the put-only part, in which case the legacy behavior is preserved.
+        int existingElementIndex = activeElementToTsMap.indexOf(toAddElement);
+        if (existingElementIndex >= 0) {
+          Object existingElement = activeElementToTsMap.getByIndex(existingElementIndex).getKey();
+          Schema elementSchema = getArraySchema(currValueRecordField.schema()).getElementType();
+          if (AvroCollectionElementComparator.FULL_COMPARISON_INSTANCE
+              .compare(toAddElement, existingElement, elementSchema) > 0) {
+            activeElementToTsMap.remove(toAddElement);
+            activeElementToTsMap.put(toAddElement, modifyTimestamp);
+            updated = true;
+          }
+        }
       }
     }
 

clients/da-vinci-client/src/main/java/com/linkedin/davinci/schema/writecompute/WriteComputeHandlerV2.java

@@ -32,11 +32,15 @@ public class WriteComputeHandlerV2 extends WriteComputeHandlerV1 {
   private final CollectionFieldOperationHandler collectionFieldOperationHandler;
 
   WriteComputeHandlerV2(MergeRecordHelper mergeRecordHelper) {
+    this(mergeRecordHelper, false);
+  }
+
+  WriteComputeHandlerV2(MergeRecordHelper mergeRecordHelper, boolean collectionFieldElementReplacementEnabled) {
     Validate.notNull(mergeRecordHelper);
     this.mergeRecordHelper = mergeRecordHelper;
-    // TODO: get this variable as a argument passed to this constructor.
-    this.collectionFieldOperationHandler =
-        new SortBasedCollectionFieldOpHandler(AvroCollectionElementComparator.INSTANCE);
+    this.collectionFieldOperationHandler = new SortBasedCollectionFieldOpHandler(
+        AvroCollectionElementComparator.INSTANCE,
+        collectionFieldElementReplacementEnabled);
   }
 
   /**

clients/da-vinci-client/src/main/java/com/linkedin/davinci/schema/writecompute/WriteComputeProcessor.java

@@ -41,7 +41,11 @@ public class WriteComputeProcessor {
   private final WriteComputeHandlerV2 writeComputeHandlerV2;
 
   public WriteComputeProcessor(MergeRecordHelper mergeRecordHelper) {
-    this.writeComputeHandlerV2 = new WriteComputeHandlerV2(mergeRecordHelper);
+    this(mergeRecordHelper, false);
+  }
+
+  public WriteComputeProcessor(MergeRecordHelper mergeRecordHelper, boolean collectionFieldElementReplacementEnabled) {
+    this.writeComputeHandlerV2 = new WriteComputeHandlerV2(mergeRecordHelper, collectionFieldElementReplacementEnabled);
   }
 
   /**

clients/da-vinci-client/src/test/java/com/linkedin/davinci/schema/merge/SortBasedCollectionFieldElementReplacementTest.java

@@ -0,0 +1,195 @@
+package com.linkedin.davinci.schema.merge;
+
+import com.linkedin.davinci.schema.SchemaUtils;
+import com.linkedin.venice.schema.AvroSchemaParseUtils;
+import com.linkedin.venice.schema.rmd.RmdConstants;
+import com.linkedin.venice.schema.rmd.RmdSchemaGenerator;
+import com.linkedin.venice.schema.rmd.v1.CollectionRmdTimestamp;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+/**
+ * Verifies the collection-merge (SET_UNION) element-replacement behavior gated by
+ * {@code server.aa.collection.field.element.replacement.enabled}.
+ *
+ * The list elements are records with a field marked {@code order: ignore}. Two elements that differ only in that ignored
+ * field are equal under Avro comparison (and {@link Object#equals}/{@link Object#hashCode}), so a union-add of such an
+ * element collides with the stored one. The legacy behavior keeps the stored element and only advances its timestamp,
+ * dropping the incoming element's content in the ignored field; the new behavior (flag on) applies the winning element.
+ */
+public class SortBasedCollectionFieldElementReplacementTest {
+  private static final String LIST_FIELD = "RecordListField";
+  private static final String ID_FIELD = "id";
+  private static final String IGNORED_FIELD = "metadata";
+
+  private static final Schema VALUE_SCHEMA = AvroSchemaParseUtils.parseSchemaFromJSONStrictValidation(
+      "{\n" + "  \"type\": \"record\",\n" + "  \"name\": \"TestValueWithIgnoredField\",\n"
+          + "  \"namespace\": \"com.linkedin.davinci.schema.merge\",\n" + "  \"fields\": [\n" + "    {\n"
+          + "      \"name\": \"" + LIST_FIELD + "\",\n" + "      \"type\": {\n" + "        \"type\": \"array\",\n"
+          + "        \"items\": {\n" + "          \"type\": \"record\",\n"
+          + "          \"name\": \"ElementWithIgnoredField\",\n" + "          \"fields\": [\n"
+          + "            {\"name\": \"" + ID_FIELD + "\", \"type\": \"string\"},\n" + "            {\"name\": \""
+          + IGNORED_FIELD + "\", \"type\": \"string\", \"order\": \"ignore\", \"default\": \"\"}\n" + "          ]\n"
+          + "        }\n" + "      },\n" + "      \"default\": []\n" + "    }\n" + "  ]\n" + "}");
+
+  private static final Schema ELEMENT_SCHEMA = VALUE_SCHEMA.getField(LIST_FIELD).schema().getElementType();
+
+  private static final Schema RMD_SCHEMA =
+      SchemaUtils.annotateRmdSchema(RmdSchemaGenerator.generateMetadataSchema(VALUE_SCHEMA));
+  private static final Schema RMD_TIMESTAMP_SCHEMA =
+      RMD_SCHEMA.getField(RmdConstants.TIMESTAMP_FIELD_NAME).schema().getTypes().get(1);
+
+  @Test
+  public void testNewerTimestampKeepsStoredElementWhenDisabled() {
+    SortBasedCollectionFieldOpHandler handler =
+        new SortBasedCollectionFieldOpHandler(AvroCollectionElementComparator.INSTANCE);
+    GenericRecord value = valueWithElement(element("a", "old"));
+    CollectionRmdTimestamp<Object> rmd = collectionMetadata(1L, Collections.singletonList(5L));
+
+    UpdateResultStatus status = handler.handleModifyList(
+        10L,
+        rmd,
+        value,
+        value.getSchema().getField(LIST_FIELD),
+        Collections.singletonList(element("a", "new")),
+        Collections.emptyList());
+
+    Assert.assertEquals(status, UpdateResultStatus.PARTIALLY_UPDATED);
+    List<GenericRecord> result = currentList(value);
+    Assert.assertEquals(result.size(), 1);
+    // Legacy behavior: the ignored-field content of the incoming element is dropped, only the timestamp is advanced.
+    Assert.assertEquals(result.get(0).get(IGNORED_FIELD).toString(), "old");
+    Assert.assertEquals(rmd.getActiveElementTimestamps().get(0).longValue(), 10L);
+  }
+
+  @Test
+  public void testNewerTimestampReplacesStoredElementWhenEnabled() {
+    SortBasedCollectionFieldOpHandler handler =
+        new SortBasedCollectionFieldOpHandler(AvroCollectionElementComparator.INSTANCE, true);
+    GenericRecord value = valueWithElement(element("a", "old"));
+    CollectionRmdTimestamp<Object> rmd = collectionMetadata(1L, Collections.singletonList(5L));
+
+    UpdateResultStatus status = handler.handleModifyList(
+        10L,
+        rmd,
+        value,
+        value.getSchema().getField(LIST_FIELD),
+        Collections.singletonList(element("a", "new")),
+        Collections.emptyList());
+
+    Assert.assertEquals(status, UpdateResultStatus.PARTIALLY_UPDATED);
+    List<GenericRecord> result = currentList(value);
+    Assert.assertEquals(result.size(), 1);
+    // New behavior: the incoming element (newer timestamp) replaces the stored one, propagating the ignored field.
+    Assert.assertEquals(result.get(0).get(IGNORED_FIELD).toString(), "new");
+    Assert.assertEquals(rmd.getActiveElementTimestamps().get(0).longValue(), 10L);
+  }
+
+  @Test
+  public void testEqualTimestampReplacesWhenIncomingWinsTieBreak() {
+    SortBasedCollectionFieldOpHandler handler =
+        new SortBasedCollectionFieldOpHandler(AvroCollectionElementComparator.INSTANCE, true);
+    GenericRecord value = valueWithElement(element("a", "aaa"));
+    CollectionRmdTimestamp<Object> rmd = collectionMetadata(1L, Collections.singletonList(5L));
+
+    // Same timestamp as the stored element; "zzz" > "aaa" by full-content comparison, so the incoming element wins.
+    handler.handleModifyList(
+        5L,
+        rmd,
+        value,
+        value.getSchema().getField(LIST_FIELD),
+        Collections.singletonList(element("a", "zzz")),
+        Collections.emptyList());
+
+    List<GenericRecord> result = currentList(value);
+    Assert.assertEquals(result.size(), 1);
+    Assert.assertEquals(result.get(0).get(IGNORED_FIELD).toString(), "zzz");
+    Assert.assertEquals(rmd.getActiveElementTimestamps().get(0).longValue(), 5L);
+  }
+
+  @Test
+  public void testEqualTimestampKeepsExistingWhenIncomingLosesTieBreak() {
+    SortBasedCollectionFieldOpHandler handler =
+        new SortBasedCollectionFieldOpHandler(AvroCollectionElementComparator.INSTANCE, true);
+    GenericRecord value = valueWithElement(element("a", "zzz"));
+    CollectionRmdTimestamp<Object> rmd = collectionMetadata(1L, Collections.singletonList(5L));
+
+    // Same timestamp; "aaa" < "zzz" by full-content comparison, so the stored element is kept (deterministic
+    // tie-break).
+    handler.handleModifyList(
+        5L,
+        rmd,
+        value,
+        value.getSchema().getField(LIST_FIELD),
+        Collections.singletonList(element("a", "aaa")),
+        Collections.emptyList());
+
+    List<GenericRecord> result = currentList(value);
+    Assert.assertEquals(result.size(), 1);
+    Assert.assertEquals(result.get(0).get(IGNORED_FIELD).toString(), "zzz");
+    Assert.assertEquals(rmd.getActiveElementTimestamps().get(0).longValue(), 5L);
+  }
+
+  @Test
+  public void testEqualTimestampNoOpWhenContentIsIdentical() {
+    SortBasedCollectionFieldOpHandler handler =
+        new SortBasedCollectionFieldOpHandler(AvroCollectionElementComparator.INSTANCE, true);
+    GenericRecord value = valueWithElement(element("a", "same"));
+    CollectionRmdTimestamp<Object> rmd = collectionMetadata(1L, Collections.singletonList(5L));
+
+    // Same timestamp and identical content (full-content comparison == 0): the deterministic tie-break replaces
+    // nothing, so the element is untouched and no update is reported (no Active/Active churn on a redundant re-add).
+    UpdateResultStatus status = handler.handleModifyList(
+        5L,
+        rmd,
+        value,
+        value.getSchema().getField(LIST_FIELD),
+        Collections.singletonList(element("a", "same")),
+        Collections.emptyList());
+
+    Assert.assertEquals(status, UpdateResultStatus.NOT_UPDATED_AT_ALL);
+    List<GenericRecord> result = currentList(value);
+    Assert.assertEquals(result.size(), 1);
+    Assert.assertEquals(result.get(0).get(IGNORED_FIELD).toString(), "same");
+    Assert.assertEquals(rmd.getActiveElementTimestamps().get(0).longValue(), 5L);
+  }
+
+  private GenericRecord element(String id, String ignoredFieldValue) {
+    GenericRecord record = new GenericData.Record(ELEMENT_SCHEMA);
+    record.put(ID_FIELD, id);
+    record.put(IGNORED_FIELD, ignoredFieldValue);
+    return record;
+  }
+
+  private GenericRecord valueWithElement(GenericRecord element) {
+    GenericRecord value = new GenericData.Record(VALUE_SCHEMA);
+    value.put(LIST_FIELD, new ArrayList<>(Collections.singletonList(element)));
+    return value;
+  }
+
+  @SuppressWarnings("unchecked")
+  private List<GenericRecord> currentList(GenericRecord value) {
+    return (List<GenericRecord>) value.get(LIST_FIELD);
+  }
+
+  private CollectionRmdTimestamp<Object> collectionMetadata(
+      long topLevelTimestamp,
+      List<Long> activeElementTimestamps) {
+    CollectionTimestampBuilder builder = new CollectionTimestampBuilder(ELEMENT_SCHEMA);
+    builder.setTopLevelTimestamps(topLevelTimestamp);
+    builder.setTopLevelColoID(0);
+    builder.setPutOnlyPartLength(0);
+    builder.setActiveElementsTimestamps(activeElementTimestamps);
+    builder.setDeletedElementTimestamps(Collections.<Long>emptyList());
+    builder.setDeletedElements(ELEMENT_SCHEMA, Collections.emptyList());
+    builder.setCollectionTimestampSchema(RMD_TIMESTAMP_SCHEMA.getField(LIST_FIELD).schema());
+    return new CollectionRmdTimestamp<>(builder.build());
+  }
+}