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

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);
       }
     }
   }