🤖 Route chunked GlobalRtDiv storage entirely through metadata partition

Previously, chunk and manifest messages for GlobalRtDiv were buffered in
user-data storage (via prependHeaderAndWriteToStorageEngine) and assembled
using GenericChunkingAdapter reading from the user-data partition. This
violated the invariant that GlobalRtDiv state must not touch user-data
storage.

Add putGlobalRtDivChunk / getGlobalRtDivChunk / deleteGlobalRtDivChunk to
StorageEngine, AbstractStorageEngine, and DelegatingStorageEngine. These
store intermediate chunks in the metadata partition under a
GRTD_CK_{partitionId}_{chunkKey} key scheme.

Update putGlobalRtDivStateInMetadata in StoreIngestionTask:
- CHUNK: build value-with-header, write to metadata via putGlobalRtDivChunk
- MANIFEST: read manifest, assemble chunks from metadata, decompress, store
  via storageMetadataService.putGlobalRtDivState, then delete chunk entries
- Non-chunked: unchanged (decompress + store directly)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: KaiSernLim

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

@@ -34,7 +34,6 @@
 import com.linkedin.davinci.helix.LeaderFollowerPartitionStateModel;
 import com.linkedin.davinci.ingestion.LagType;
 import com.linkedin.davinci.listener.response.AdminResponse;
-import com.linkedin.davinci.listener.response.NoOpReadResponseStats;
 import com.linkedin.davinci.notifier.VeniceNotifier;
 import com.linkedin.davinci.stats.AggVersionedDIVStats;
 import com.linkedin.davinci.stats.AggVersionedDaVinciRecordTransformerStats;
@@ -43,9 +42,7 @@
 import com.linkedin.davinci.stats.ingestion.heartbeat.HeartbeatMonitoringService;
 import com.linkedin.davinci.storage.StorageMetadataService;
 import com.linkedin.davinci.storage.StorageService;
-import com.linkedin.davinci.storage.chunking.ChunkedValueManifestContainer;
 import com.linkedin.davinci.storage.chunking.ChunkingUtils;
-import com.linkedin.davinci.storage.chunking.GenericChunkingAdapter;
 import com.linkedin.davinci.store.DelegatingStorageEngine;
 import com.linkedin.davinci.store.StorageEngine;
 import com.linkedin.davinci.store.StoragePartitionAdjustmentTrigger;
@@ -119,7 +116,6 @@
 import com.linkedin.venice.pubsub.manager.TopicManagerRepository;
 import com.linkedin.venice.pushmonitor.ExecutionStatus;
 import com.linkedin.venice.schema.SchemaEntry;
-import com.linkedin.venice.serialization.RawBytesStoreDeserializerCache;
 import com.linkedin.venice.serialization.avro.AvroProtocolDefinition;
 import com.linkedin.venice.serialization.avro.ChunkedValueManifestSerializer;
 import com.linkedin.venice.serialization.avro.InternalAvroSpecificSerializer;
@@ -4111,50 +4107,59 @@ protected void putInStorageEngine(int partition, byte[] keyBytes, Put put) {
 
   protected void putGlobalRtDivStateInMetadata(int partition, byte[] keyBytes, Put put) {
     if (put.schemaId == CHUNK_SCHEMA_ID) {
-      // Intermediate chunk: store in user-data storage for later assembly when the manifest arrives.
-      prependHeaderAndWriteToStorageEngine(partition, keyBytes, put);
+      // Intermediate chunk: store in the metadata partition for later assembly when the manifest arrives.
+      byte[] chunkPayload = ByteUtils.extractByteArray(put.putValue);
+      byte[] valueWithHeader = new byte[ValueRecord.SCHEMA_HEADER_LENGTH + chunkPayload.length];
+      ByteUtils.writeInt(valueWithHeader, CHUNK_SCHEMA_ID, 0);
+      System.arraycopy(chunkPayload, 0, valueWithHeader, ValueRecord.SCHEMA_HEADER_LENGTH, chunkPayload.length);
+      executeStorageEngineRunnable(
+          partition,
+          () -> storageEngine.putGlobalRtDivChunk(partition, keyBytes, valueWithHeader));
       return;
     }
 
     if (put.schemaId == CHUNK_MANIFEST_SCHEMA_ID) {
-      // Manifest for a chunked GlobalRtDiv: write manifest to user-data storage, then assemble all chunks
-      // and persist the assembled (decompressed) value in the metadata partition.
-      prependHeaderAndWriteToStorageEngine(partition, keyBytes, put);
-      // Strip the KeyWithChunkingSuffixSerializer suffix to recover the original broker-URL key.
+      // Manifest for a chunked GlobalRtDiv: assemble all chunks from the metadata partition,
+      // decompress the result, then persist it in the metadata partition.
       byte[] originalKeyBytes = Arrays.copyOf(keyBytes, keyBytes.length - KEY_CHUNKING_SUFFIX_LENGTH);
       String key = new String(originalKeyBytes);
       if (!key.startsWith(GLOBAL_RT_DIV_KEY_PREFIX)) {
         throw new VeniceException("Invalid chunked Global RT DIV manifest key: " + Arrays.toString(keyBytes));
       }
       String brokerUrl = key.substring(GLOBAL_RT_DIV_KEY_PREFIX.length());
+      byte[] manifestBytes = ByteUtils.extractByteArray(put.putValue);
+      ChunkedValueManifest manifest = manifestSerializer.deserialize(manifestBytes, CHUNK_MANIFEST_SCHEMA_ID);
+      ByteBuffer assembled = ByteBuffer.allocate(manifest.size);
+      for (ByteBuffer chunkKeyBuf: manifest.keysWithChunkIdSuffix) {
+        byte[] chunkKey = ByteUtils.extractByteArray(chunkKeyBuf);
+        byte[] chunkBytes = storageEngine.getGlobalRtDivChunk(partition, chunkKey);
+        if (chunkBytes == null) {
+          throw new VeniceException(
+              "Missing GlobalRtDiv chunk in metadata partition for partition: " + partition + ", broker: " + brokerUrl);
+        }
+        if (ValueRecord.parseSchemaId(chunkBytes) != CHUNK_SCHEMA_ID) {
+          throw new VeniceException(
+              "Unexpected schema ID in GlobalRtDiv chunk: " + ValueRecord.parseSchemaId(chunkBytes));
+        }
+        assembled
+            .put(chunkBytes, ValueRecord.SCHEMA_HEADER_LENGTH, chunkBytes.length - ValueRecord.SCHEMA_HEADER_LENGTH);
+      }
+      assembled.flip();
       try {
-        // Use the serialized manifest key (keyBytes) directly with isChunked=false so the adapter does not
-        // double-serialize it. All chunks were already written to user-data storage above.
-        ByteBuffer assembledValue = (ByteBuffer) GenericChunkingAdapter.INSTANCE.get(
-            storageEngine,
+        byte[] valueBytes =
+            ByteUtils.extractByteArray(compressor.get().decompress(assembled.array(), 0, assembled.limit()));
+        executeStorageEngineRunnable(
             partition,
-            ByteBuffer.wrap(keyBytes),
-            false,
-            null,
-            null,
-            NoOpReadResponseStats.SINGLETON,
-            AvroProtocolDefinition.GLOBAL_RT_DIV_STATE.getCurrentProtocolVersion(),
-            RawBytesStoreDeserializerCache.getInstance(),
-            compressor.get(),
-            new ChunkedValueManifestContainer());
-        if (assembledValue != null) {
-          byte[] valueBytes = ByteUtils.extractByteArray(assembledValue);
-          executeStorageEngineRunnable(
-              partition,
-              () -> storageMetadataService.putGlobalRtDivState(kafkaVersionTopic, partition, brokerUrl, valueBytes));
-        } else {
-          LOGGER.warn("Assembled Global RT DIV value was null for partition: {}, broker: {}", partition, brokerUrl);
-        }
-      } catch (Exception e) {
+            () -> storageMetadataService.putGlobalRtDivState(kafkaVersionTopic, partition, brokerUrl, valueBytes));
+      } catch (IOException e) {
         throw new VeniceException(
-            "Failed to assemble chunked Global RT DIV state for partition: " + partition + ", broker: " + brokerUrl,
+            "Failed to decompress assembled GlobalRtDiv state for partition: " + partition + ", broker: " + brokerUrl,
             e);
       }
+      for (ByteBuffer chunkKeyBuf: manifest.keysWithChunkIdSuffix) {
+        byte[] chunkKey = ByteUtils.extractByteArray(chunkKeyBuf);
+        executeStorageEngineRunnable(partition, () -> storageEngine.deleteGlobalRtDivChunk(partition, chunkKey));
+      }
       return;
     }
 

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

@@ -57,6 +57,7 @@ public abstract class AbstractStorageEngine<Partition extends AbstractStoragePar
   private static final byte[] VERSION_METADATA_KEY = "VERSION_METADATA".getBytes();
   private static final String PARTITION_METADATA_PREFIX = "P_";
   private static final String GLOBAL_RT_DIV_METADATA_PREFIX = "GRTD_";
+  private static final String GLOBAL_RT_DIV_CHUNK_METADATA_PREFIX = "GRTD_CK_";
 
   // Using a large positive number for metadata partition id instead of -1 can avoid database naming issues.
   public static final int METADATA_PARTITION_ID = 1000_000_000;
@@ -721,6 +722,43 @@ public synchronized void clearGlobalRtDivState(int partitionId, String brokerUrl
     metadataPartition.delete(getGlobalRtDivMetadataKey(partitionId, brokerUrl));
   }
 
+  @Override
+  public synchronized void putGlobalRtDivChunk(int partitionId, byte[] chunkKey, byte[] chunkValue) {
+    if (!metadataPartitionCreated()) {
+      throw new StorageInitializationException("Metadata partition not created!");
+    }
+    if (partitionId < 0) {
+      throw new IllegalArgumentException("Invalid partition id argument in putGlobalRtDivChunk");
+    }
+    metadataPartition.put(getGlobalRtDivChunkMetadataKey(partitionId, chunkKey), chunkValue);
+  }
+
+  @Override
+  public synchronized byte[] getGlobalRtDivChunk(int partitionId, byte[] chunkKey) {
+    if (!metadataPartitionCreated()) {
+      throw new StorageInitializationException("Metadata partition not created!");
+    }
+    if (partitionId < 0) {
+      throw new IllegalArgumentException("Invalid partition id argument in getGlobalRtDivChunk");
+    }
+    return metadataPartition.get(getGlobalRtDivChunkMetadataKey(partitionId, chunkKey));
+  }
+
+  @Override
+  public synchronized void deleteGlobalRtDivChunk(int partitionId, byte[] chunkKey) {
+    if (!metadataPartitionCreated()) {
+      LOGGER.info(
+          "Metadata partition not created; there is nothing to clear for {} partition {} chunk",
+          storeVersionName,
+          partitionId);
+      return;
+    }
+    if (partitionId < 0) {
+      throw new IllegalArgumentException("Invalid partition id argument in deleteGlobalRtDivChunk");
+    }
+    metadataPartition.delete(getGlobalRtDivChunkMetadataKey(partitionId, chunkKey));
+  }
+
   /**
    * Return true or false based on whether a given partition exists within this storage engine
    *
@@ -783,6 +821,14 @@ private static byte[] getGlobalRtDivMetadataKey(int partitionId, String brokerUr
     return (GLOBAL_RT_DIV_METADATA_PREFIX + partitionId + "_" + brokerUrl).getBytes();
   }
 
+  private static byte[] getGlobalRtDivChunkMetadataKey(int partitionId, byte[] chunkKey) {
+    byte[] prefix = (GLOBAL_RT_DIV_CHUNK_METADATA_PREFIX + partitionId + "_").getBytes();
+    byte[] metaKey = new byte[prefix.length + chunkKey.length];
+    System.arraycopy(prefix, 0, metaKey, 0, prefix.length);
+    System.arraycopy(chunkKey, 0, metaKey, prefix.length, chunkKey.length);
+    return metaKey;
+  }
+
   private boolean metadataPartitionCreated() {
     return metadataPartition != null;
   }

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

@@ -294,6 +294,21 @@ public void clearGlobalRtDivState(int partitionId, String brokerUrl) {
     this.delegate.clearGlobalRtDivState(partitionId, brokerUrl);
   }
 
+  @Override
+  public void putGlobalRtDivChunk(int partitionId, byte[] chunkKey, byte[] chunkValue) {
+    this.delegate.putGlobalRtDivChunk(partitionId, chunkKey, chunkValue);
+  }
+
+  @Override
+  public byte[] getGlobalRtDivChunk(int partitionId, byte[] chunkKey) {
+    return this.delegate.getGlobalRtDivChunk(partitionId, chunkKey);
+  }
+
+  @Override
+  public void deleteGlobalRtDivChunk(int partitionId, byte[] chunkKey) {
+    this.delegate.deleteGlobalRtDivChunk(partitionId, chunkKey);
+  }
+
   @Override
   public boolean containsPartition(int partitionId) {
     return this.delegate.containsPartition(partitionId);

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

@@ -169,6 +169,21 @@ void putWithReplicationMetadata(int partitionId, byte[] key, ByteBuffer value, b
    */
   void clearGlobalRtDivState(int partitionId, String brokerUrl);
 
+  /**
+   * Put a GlobalRtDiv intermediate chunk (with schema header prepended) into the metadata partition.
+   */
+  void putGlobalRtDivChunk(int partitionId, byte[] chunkKey, byte[] chunkValue);
+
+  /**
+   * Retrieve a GlobalRtDiv intermediate chunk from the metadata partition.
+   */
+  byte[] getGlobalRtDivChunk(int partitionId, byte[] chunkKey);
+
+  /**
+   * Delete a GlobalRtDiv intermediate chunk from the metadata partition.
+   */
+  void deleteGlobalRtDivChunk(int partitionId, byte[] chunkKey);
+
   /**
    * Return true or false based on whether a given partition exists within this storage engine
    *