[dvc][doc] Refactor DaVinciRecordTransformerConfig to use Builder pattern and update DVRT docs (#1501)

1. Previously, the configs for DaVinciRecordTransformerConfig were set by passing in all of the
values into its constructor. This means that anytime we wanted to add another config, it would be
a breaking change. To resolve this, I refactored DaVinciRecordTransformerConfig to use a Builder pattern.
2. Fixed a bug where if a user set storeRecordsInDaVinci, it would always bootstrap from VT and
ignore the on-disk state. This isn't deseriable behavior. To fix this, I introduced a new config
called alwaysBootstrapFromVersionTopic. Although most users won't use this, this can serve use-cases
where the user is only storing their records in-memory. It is more preferable for users to at least
have storeRecordsInDaVinci set to true so they can bootstrap from disk and utilize blob transfer.
3. If a user doesn't intend on modifying the schema of their records, the user won't need to pass
inoutputValueSchema and outputValueClass to the DaVinciRecordTransformerConfig anymore. This can
serve use cases where the user just wants to redirect their records to an alternative storage such as DuckDB.
4. Updated the DVRT docs to include the latest changes.

Author: kvargha

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

@@ -69,6 +69,7 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -700,6 +701,11 @@ private VeniceConfigLoader buildVeniceConfig() {
       kafkaBootstrapServers = backendConfig.getString(KAFKA_BOOTSTRAP_SERVERS);
     }
 
+    String recordTransformerOutputValueSchema = "null";
+    if (daVinciConfig.isRecordTransformerEnabled()) {
+      recordTransformerOutputValueSchema = Objects.toString(recordTransformerConfig.getOutputValueSchema(), "null");
+    }
+
     VeniceProperties config = new PropertyBuilder().put(KAFKA_ADMIN_CLASS, ApacheKafkaAdminAdapter.class.getName())
         .put(ROCKSDB_LEVEL0_FILE_NUM_COMPACTION_TRIGGER, 4) // RocksDB default config
         .put(ROCKSDB_LEVEL0_SLOWDOWN_WRITES_TRIGGER, 20) // RocksDB default config
@@ -712,11 +718,7 @@ private VeniceConfigLoader buildVeniceConfig() {
         .put(KAFKA_BOOTSTRAP_SERVERS, kafkaBootstrapServers)
         .put(ROCKSDB_PLAIN_TABLE_FORMAT_ENABLED, daVinciConfig.getStorageClass() == StorageClass.MEMORY_BACKED_BY_DISK)
         .put(INGESTION_USE_DA_VINCI_CLIENT, true)
-        .put(
-            RECORD_TRANSFORMER_VALUE_SCHEMA,
-            daVinciConfig.isRecordTransformerEnabled()
-                ? recordTransformerConfig.getOutputValueSchema().toString()
-                : "null")
+        .put(RECORD_TRANSFORMER_VALUE_SCHEMA, recordTransformerOutputValueSchema)
         .put(INGESTION_ISOLATION_CONFIG_PREFIX + "." + INGESTION_MEMORY_LIMIT, -1) // Explicitly disable memory limiter
                                                                                    // in Isolated Process
         .put(backendConfig.toProperties())

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

@@ -29,8 +29,8 @@ public BlockingDaVinciRecordTransformer(
       Schema keySchema,
       Schema inputValueSchema,
       Schema outputValueSchema,
-      boolean storeRecordsInDaVinci) {
-    super(recordTransformer.getStoreVersion(), keySchema, inputValueSchema, outputValueSchema, storeRecordsInDaVinci);
+      DaVinciRecordTransformerConfig recordTransformerConfig) {
+    super(recordTransformer.getStoreVersion(), keySchema, inputValueSchema, outputValueSchema, recordTransformerConfig);
     this.recordTransformer = recordTransformer;
   }
 

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

@@ -43,6 +43,11 @@ public abstract class DaVinciRecordTransformer<K, V, O> implements Closeable {
    */
   private final boolean storeRecordsInDaVinci;
 
+  /**
+   * Boolean to determine if we should always bootstrap from the Version Topic.
+   */
+  private final boolean alwaysBootstrapFromVersionTopic;
+
   /**
    * The key schema, which is immutable inside DaVinciClient. Users can modify the key if they are storing records in an external storage engine, but this must be managed by the user.
    */
@@ -65,17 +70,17 @@ public abstract class DaVinciRecordTransformer<K, V, O> implements Closeable {
    * @param keySchema the key schema, which is immutable inside DaVinciClient. Users can modify the key if they are storing records in an external storage engine, but this must be managed by the user
    * @param inputValueSchema the value schema before transformation
    * @param outputValueSchema the value schema after transformation
-   * @param storeRecordsInDaVinci set this to false if you intend to store records in a custom storage,
-   *                              and not in the Da Vinci Client
+   * @param recordTransformerConfig the config for the record transformer
    */
   public DaVinciRecordTransformer(
       int storeVersion,
       Schema keySchema,
       Schema inputValueSchema,
       Schema outputValueSchema,
-      boolean storeRecordsInDaVinci) {
+      DaVinciRecordTransformerConfig recordTransformerConfig) {
     this.storeVersion = storeVersion;
-    this.storeRecordsInDaVinci = storeRecordsInDaVinci;
+    this.storeRecordsInDaVinci = recordTransformerConfig.getStoreRecordsInDaVinci();
+    this.alwaysBootstrapFromVersionTopic = recordTransformerConfig.getAlwaysBootstrapFromVersionTopic();
     this.keySchema = keySchema;
     // ToDo: Make use of inputValueSchema to support reader/writer schemas
     this.inputValueSchema = inputValueSchema;
@@ -227,6 +232,13 @@ public final boolean getStoreRecordsInDaVinci() {
     return storeRecordsInDaVinci;
   }
 
+  /**
+   * @return {@link #alwaysBootstrapFromVersionTopic}
+   */
+  public final boolean getAlwaysBootstrapFromVersionTopic() {
+    return alwaysBootstrapFromVersionTopic;
+  }
+
   /**
    * Returns the schema for the key used in {@link DaVinciClient}'s operations.
    *

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

@@ -1,5 +1,7 @@
 package com.linkedin.davinci.client;
 
+import com.linkedin.venice.exceptions.VeniceException;
+import java.util.Optional;
 import org.apache.avro.Schema;
 
 
@@ -10,19 +12,22 @@ public class DaVinciRecordTransformerConfig {
   private final DaVinciRecordTransformerFunctionalInterface recordTransformerFunction;
   private final Class outputValueClass;
   private final Schema outputValueSchema;
+  private final boolean storeRecordsInDaVinci;
+  private final boolean alwaysBootstrapFromVersionTopic;
 
-  /**
-   * @param recordTransformerFunction the functional interface for creating a {@link DaVinciRecordTransformer}
-   * @param outputValueClass the class of the output value
-   * @param outputValueSchema the schema of the output value
-   */
-  public DaVinciRecordTransformerConfig(
-      DaVinciRecordTransformerFunctionalInterface recordTransformerFunction,
-      Class outputValueClass,
-      Schema outputValueSchema) {
-    this.recordTransformerFunction = recordTransformerFunction;
-    this.outputValueClass = outputValueClass;
-    this.outputValueSchema = outputValueSchema;
+  public DaVinciRecordTransformerConfig(Builder builder) {
+    this.recordTransformerFunction = Optional.ofNullable(builder.recordTransformerFunction)
+        .orElseThrow(() -> new VeniceException("recordTransformerFunction cannot be null"));
+
+    this.outputValueClass = builder.outputValueClass;
+    this.outputValueSchema = builder.outputValueSchema;
+    if ((this.outputValueClass != null && this.outputValueSchema == null)
+        || (this.outputValueClass == null && this.outputValueSchema != null)) {
+      throw new VeniceException("outputValueClass and outputValueSchema must be defined together");
+    }
+
+    this.storeRecordsInDaVinci = Optional.ofNullable(builder.storeRecordsInDaVinci).orElse(true);
+    this.alwaysBootstrapFromVersionTopic = Optional.ofNullable(builder.alwaysBootstrapFromVersionTopic).orElse(false);
   }
 
   /**
@@ -45,4 +50,76 @@ public Class getOutputValueClass() {
   public Schema getOutputValueSchema() {
     return outputValueSchema;
   }
+
+  /**
+   * @return {@link #storeRecordsInDaVinci}
+   */
+  public boolean getStoreRecordsInDaVinci() {
+    return storeRecordsInDaVinci;
+  }
+
+  /**
+   * @return {@link #alwaysBootstrapFromVersionTopic}
+   */
+  public boolean getAlwaysBootstrapFromVersionTopic() {
+    return alwaysBootstrapFromVersionTopic;
+  }
+
+  public static class Builder {
+    private DaVinciRecordTransformerFunctionalInterface recordTransformerFunction;
+    private Class outputValueClass;
+    private Schema outputValueSchema;
+    private Boolean storeRecordsInDaVinci;
+    private Boolean alwaysBootstrapFromVersionTopic;
+
+    /**
+     * @param recordTransformerFunction the functional interface for creating a {@link DaVinciRecordTransformer}
+     */
+    public Builder setRecordTransformerFunction(DaVinciRecordTransformerFunctionalInterface recordTransformerFunction) {
+      this.recordTransformerFunction = recordTransformerFunction;
+      return this;
+    }
+
+    /**
+     * Set this if you modify the schema during transformation. Must be used in conjunction with {@link #setOutputValueSchema(Schema)}
+     * @param outputValueClass the class of the output value
+     */
+    public Builder setOutputValueClass(Class outputValueClass) {
+      this.outputValueClass = outputValueClass;
+      return this;
+    }
+
+    /**
+     * Set this if you modify the schema during transformation. Must be used in conjunction with {@link #setOutputValueClass(Class)}
+     * @param outputValueSchema the schema of the output value
+     */
+    public Builder setOutputValueSchema(Schema outputValueSchema) {
+      this.outputValueSchema = outputValueSchema;
+      return this;
+    }
+
+    /**
+     * @param storeRecordsInDaVinci set this to false if you intend to store records in a custom storage,
+     *                              and not in the Da Vinci Client.
+     *                              Default is true.
+     */
+    public Builder setStoreRecordsInDaVinci(boolean storeRecordsInDaVinci) {
+      this.storeRecordsInDaVinci = storeRecordsInDaVinci;
+      return this;
+    }
+
+    /**
+     * @param alwaysBootstrapFromVersionTopic set this to true if {@link #storeRecordsInDaVinci} is false, and you're
+     *                                        storing records in memory without being backed by disk.
+     *                                        Default is false.
+     */
+    public Builder setAlwaysBootstrapFromVersionTopic(boolean alwaysBootstrapFromVersionTopic) {
+      this.alwaysBootstrapFromVersionTopic = alwaysBootstrapFromVersionTopic;
+      return this;
+    }
+
+    public DaVinciRecordTransformerConfig build() {
+      return new DaVinciRecordTransformerConfig(this);
+    }
+  }
 }

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

@@ -13,5 +13,6 @@ DaVinciRecordTransformer apply(
       Integer storeVersion,
       Schema keySchema,
       Schema inputValueSchema,
-      Schema outputValueSchema);
+      Schema outputValueSchema,
+      DaVinciRecordTransformerConfig recordTransformerConfig);
 }

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

@@ -110,7 +110,7 @@ public final void onRecovery(
 
     boolean transformerLogicChanged = hasTransformerLogicChanged(classHash, offsetRecord);
 
-    if (!recordTransformer.getStoreRecordsInDaVinci() || transformerLogicChanged) {
+    if (recordTransformer.getAlwaysBootstrapFromVersionTopic() || transformerLogicChanged) {
       LOGGER.info("Bootstrapping directly from the VersionTopic for partition {}", partitionId);
 
       // Bootstrap from VT

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

@@ -481,15 +481,25 @@ public StoreIngestionTask(
       this.recordTransformerInputValueSchema = schemaRepository.getSupersetOrLatestValueSchema(storeName).getSchema();
       Schema outputValueSchema = recordTransformerConfig.getOutputValueSchema();
 
+      // User doesn't intend on transforming records. Use input value schema instead.
+      if (outputValueSchema == null) {
+        outputValueSchema = this.recordTransformerInputValueSchema;
+      }
+
       DaVinciRecordTransformer clientRecordTransformer = recordTransformerConfig.getRecordTransformerFunction()
-          .apply(versionNumber, keySchema, this.recordTransformerInputValueSchema, outputValueSchema);
+          .apply(
+              versionNumber,
+              keySchema,
+              this.recordTransformerInputValueSchema,
+              outputValueSchema,
+              recordTransformerConfig);
 
       this.recordTransformer = new BlockingDaVinciRecordTransformer(
           clientRecordTransformer,
           keySchema,
           this.recordTransformerInputValueSchema,
           outputValueSchema,
-          clientRecordTransformer.getStoreRecordsInDaVinci());
+          recordTransformerConfig);
       this.recordTransformerDeserializersByPutSchemaId = new SparseConcurrentList<>();
 
       versionedIngestionStats.registerTransformerLatencySensor(storeName, versionNumber);

clients/da-vinci-client/src/test/java/com/linkedin/davinci/client/AvroGenericDaVinciClientTest.java

@@ -60,15 +60,11 @@ public AvroGenericDaVinciClient setUpClientWithRecordTransformer(
       daVinciConfig = new DaVinciConfig();
     }
 
-    DaVinciRecordTransformerConfig recordTransformerConfig = new DaVinciRecordTransformerConfig(
-        (storeVersion, keySchema, inputValueSchema, outputValueSchema) -> new TestStringRecordTransformer(
-            storeVersion,
-            keySchema,
-            inputValueSchema,
-            outputValueSchema,
-            true),
-        String.class,
-        Schema.create(Schema.Type.STRING));
+    DaVinciRecordTransformerConfig recordTransformerConfig =
+        new DaVinciRecordTransformerConfig.Builder().setRecordTransformerFunction(TestStringRecordTransformer::new)
+            .setOutputValueClass(String.class)
+            .setOutputValueSchema(Schema.create(Schema.Type.STRING))
+            .build();
     daVinciConfig.setRecordTransformerConfig(recordTransformerConfig);
 
     VeniceProperties backendConfig = mock(VeniceProperties.class);

clients/da-vinci-client/src/test/java/com/linkedin/davinci/config/DaVinciConfigTest.java

@@ -8,6 +8,7 @@
 import com.linkedin.davinci.client.DaVinciConfig;
 import com.linkedin.davinci.client.DaVinciRecordTransformer;
 import com.linkedin.davinci.client.DaVinciRecordTransformerConfig;
+import com.linkedin.davinci.client.DaVinciRecordTransformerFunctionalInterface;
 import com.linkedin.davinci.client.DaVinciRecordTransformerResult;
 import com.linkedin.venice.utils.lazy.Lazy;
 import java.io.IOException;
@@ -22,8 +23,8 @@ public TestRecordTransformer(
         Schema keySchema,
         Schema inputValueSchema,
         Schema outputValueSchema,
-        boolean storeRecordsInDaVinci) {
-      super(storeVersion, keySchema, inputValueSchema, outputValueSchema, storeRecordsInDaVinci);
+        DaVinciRecordTransformerConfig recordTransformerConfig) {
+      super(storeVersion, keySchema, inputValueSchema, outputValueSchema, recordTransformerConfig);
     }
 
     @Override
@@ -47,16 +48,21 @@ public void testRecordTransformerEnabled() {
     DaVinciConfig config = new DaVinciConfig();
     assertFalse(config.isRecordTransformerEnabled());
 
-    DaVinciRecordTransformerConfig recordTransformerConfig = new DaVinciRecordTransformerConfig(
-        (storeVersion, keySchema, inputValueSchema, outputValueSchema) -> new TestRecordTransformer(
+    DaVinciRecordTransformerConfig dummyRecordTransformerConfig =
+        new DaVinciRecordTransformerConfig.Builder().setRecordTransformerFunction(TestRecordTransformer::new).build();
+
+    DaVinciRecordTransformerFunctionalInterface recordTransformerFunction =
+        (storeVersion, keySchema, inputValueSchema, outputValueSchema, config1) -> new TestRecordTransformer(
             storeVersion,
             keySchema,
             inputValueSchema,
             outputValueSchema,
-            true),
-        String.class,
-        Schema.create(Schema.Type.INT));
+            dummyRecordTransformerConfig);
+
+    DaVinciRecordTransformerConfig recordTransformerConfig =
+        new DaVinciRecordTransformerConfig.Builder().setRecordTransformerFunction(recordTransformerFunction).build();
     config.setRecordTransformerConfig(recordTransformerConfig);
+
     assertTrue(config.isRecordTransformerEnabled());
   }
 
@@ -65,16 +71,21 @@ public void testGetAndSetRecordTransformer() {
     DaVinciConfig config = new DaVinciConfig();
     assertNull(config.getRecordTransformerConfig());
 
-    DaVinciRecordTransformerConfig recordTransformerConfig = new DaVinciRecordTransformerConfig(
-        (storeVersion, keySchema, inputValueSchema, outputValueSchema) -> new TestRecordTransformer(
+    DaVinciRecordTransformerConfig dummyRecordTransformerConfig =
+        new DaVinciRecordTransformerConfig.Builder().setRecordTransformerFunction(TestRecordTransformer::new).build();
+
+    DaVinciRecordTransformerFunctionalInterface recordTransformerFunction =
+        (storeVersion, keySchema, inputValueSchema, outputValueSchema, config1) -> new TestRecordTransformer(
             storeVersion,
             keySchema,
             inputValueSchema,
             outputValueSchema,
-            true),
-        String.class,
-        Schema.create(Schema.Type.INT));
+            dummyRecordTransformerConfig);
+
+    DaVinciRecordTransformerConfig recordTransformerConfig =
+        new DaVinciRecordTransformerConfig.Builder().setRecordTransformerFunction(recordTransformerFunction).build();
     config.setRecordTransformerConfig(recordTransformerConfig);
+
     assertNotNull(config.getRecordTransformerConfig());
   }