[server][common][controller][vpj] Materialized view projection and filter support

Add projection and filtering support for materialized view (MV) to be more efficient about unwanted data for view
consumers. Projection can be enabled by setting projection fields in the materialized view parameters. Similarly
filtering can be enabled by setting filter by fields. These two features can be enabled separately or together.
If enabled together the filter by fields will be included in the projecting fields automatically. Here is an example
MV configuration to illustrate the ideas:

Record containing fields: {a, b, c, d, e}
Projecting fields: {b, c}
Filtering fields: {a}

The only filtering option for now is to skip if none of the filter by fields changed. The filtering is also only applied
during hybrid ingestion since it doesn't make sense to have a change filter on batch push. With the above setup we will
project and write all batch data to the MV ({a, b, c}). RT updates (full PUT or UPDATE) will project and write the
resulting record to the MV ({a, b, c}) only if the value of field (a) is different from the old value. All DELETE events
will be written to the MV (no filtering).

In order to achieve the above behavior there are several changes:

1. Previously we've used pub sub message headers to perform forwarding to handle chunks during NR pass-through in remote
regions. This strategy will not work with projection because in order for us to perform projection on batch data in
remote regions, we will need the remote partition leaders to assemble the chunks during NR pass-through. We are
replacing the forwarding strategy with InMemoryChunkAssembler. To ensure leaders don't resume in-between chunks we will
also buffer and delay writing the chunks to drainer until we have a fully assembled record and produced it to view
topic(s).

2. Added enforcement in controller to ensure view configs are immutable. Projection schema is generated when adding a
new materialized view and stored with the view config. Since there can only be one schema version per view, the znode
size should be manageable with compression. If this becomes a concern we can also store it separately or generate it on
the fly. We also verify the filtering by fields and projection fields to ensure they exist in latest superset or value
schema and have default values.

3. Projection is performed in ComplexVeniceWriter as part of complexPut so both VPJ and leaders can use the same code
for projection. Filtering is performed in MaterializedViewWriter since current offering of change filter is applicable
only to hyrbid writes.

Author: xunyin8

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

@@ -104,7 +104,7 @@
 public class VeniceChangelogConsumerImpl<K, V> implements VeniceChangelogConsumer<K, V> {
   private static final Logger LOGGER = LogManager.getLogger(VeniceChangelogConsumerImpl.class);
   private static final int MAX_SUBSCRIBE_RETRIES = 5;
-  private static final String ROCKSDB_BUFFER_FOLDER = "rocksdb-chunk-buffer";
+  private static final String ROCKSDB_BUFFER_FOLDER_PREFIX = "rocksdb-chunk-buffer-";
   protected long subscribeTime = Long.MAX_VALUE;
 
   protected final ReadWriteLock subscriptionLock = new ReentrantReadWriteLock();
@@ -191,8 +191,11 @@ public VeniceChangelogConsumerImpl(
         throw new VeniceException("bootstrapFileSystemPath must be configured for consuming view store: " + storeName);
       }
       // Create a new folder in user provided path so if the path contains other important files we don't drop it.
-      rocksDBBufferProperties
-          .put(DATA_BASE_PATH, RocksDBUtils.composeStoreDbDir(rocksDBBufferPath, ROCKSDB_BUFFER_FOLDER));
+      rocksDBBufferProperties.put(
+          DATA_BASE_PATH,
+          RocksDBUtils.composeStoreDbDir(
+              rocksDBBufferPath,
+              ROCKSDB_BUFFER_FOLDER_PREFIX + changelogClientConfig.getConsumerName()));
       // These properties are required to build a VeniceServerConfig but is never used by RocksDBStorageEngineFactory.
       // Instead of setting these configs, we could refactor RocksDBStorageEngineFactory to take a more generic config.
       rocksDBBufferProperties.put(CLUSTER_NAME, "");

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

@@ -71,6 +71,7 @@
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BiConsumer;
 import java.util.function.BooleanSupplier;
+import java.util.function.Consumer;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.io.BinaryDecoder;
 import org.apache.helix.manager.zk.ZKHelixAdmin;
@@ -456,7 +457,8 @@ private PubSubMessageProcessedResult processActiveActiveMessage(
             consumerRecord.getTopicPartition(),
             valueManifestContainer,
             beforeProcessingBatchRecordsTimestampMs));
-    if (hasChangeCaptureView || (hasComplexVenicePartitionerMaterializedView && msgType == MessageType.DELETE)) {
+    if (hasChangeCaptureView || hasFilterByFieldsMaterializedView
+        || (hasComplexVenicePartitionerMaterializedView && msgType == MessageType.DELETE)) {
       /**
        * Since this function will update the transient cache before writing the view, and if there is
        * a change capture view writer, we need to lookup first, otherwise the transient cache will be populated
@@ -655,8 +657,9 @@ protected void processMessageAndMaybeProduceToKafka(
       // following function
       // call in this context much less obtrusive, however, it implies that all views can only work for AA stores
 
-      // Write to views
-      Runnable produceToVersionTopic = () -> producePutOrDeleteToKafka(
+      // Write to views. In A/A ingestion we never need to delay VT writes. Using local variables is sufficient to
+      // define the produceToVersionTopic consumer.
+      Consumer<PubSubMessageProcessedResultWrapper> produceToVersionTopic = (ignored) -> producePutOrDeleteToKafka(
           mergeConflictResultWrapper,
           partitionConsumptionState,
           keyBytes,
@@ -676,25 +679,25 @@ protected void processMessageAndMaybeProduceToKafka(
         ByteBuffer oldValueBB = mergeConflictResultWrapper.getOldValueByteBufferProvider().get();
         int oldValueSchemaId =
             oldValueBB == null ? -1 : mergeConflictResultWrapper.getOldValueProvider().get().writerSchemaId();
-        Lazy<GenericRecord> valueProvider = mergeConflictResultWrapper.getValueProvider();
         // The helper function takes in a BiFunction but the parameter for view partition set will never be used and
         // always null for A/A ingestion of the RT topic.
         queueUpVersionTopicWritesWithViewWriters(
             partitionConsumptionState,
-            (viewWriter, ignored) -> viewWriter.processRecord(
+            (viewWriter) -> viewWriter.processRecord(
                 mergeConflictResultWrapper.getUpdatedValueBytes(),
                 oldValueBB,
                 keyBytes,
                 mergeConflictResult.getValueSchemaId(),
                 oldValueSchemaId,
                 mergeConflictResult.getRmdRecord(),
-                valueProvider),
-            null,
-            produceToVersionTopic);
+                mergeConflictResultWrapper.getValueProvider(),
+                mergeConflictResultWrapper.getDeserializedOldValueProvider()),
+            produceToVersionTopic,
+            Collections.singletonList(consumerRecordWrapper));
       } else {
         // This function may modify the original record in KME and it is unsafe to use the payload from KME directly
         // after this call.
-        produceToVersionTopic.run();
+        produceToVersionTopic.accept(consumerRecordWrapper);
       }
     }
   }