@snow SNOW-835618 Snowpipe Streaming: send uncompressed chunk length …

…from SDK to GS
snowflakedb · Jun 12, 2023 · 175c989 · 175c989
1 parent bdb0a4a
commit 175c989
Show file tree

Hide file tree

Showing 3 changed files with 22 additions and 1 deletion.
diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/BlobBuilder.java b/src/main/java/net/snowflake/ingest/streaming/internal/BlobBuilder.java
@@ -114,6 +114,7 @@ static <T> Blob constructBlobAndMetadata(
                 // The paddedChunkLength is used because it is the actual data size used for
                 // decompression and md5 calculation on server side.
                 .setChunkLength(paddedChunkLength)
+                .setUncompressedChunkLength((int) serializedChunk.chunkUncompressedSize)
                 .setChannelList(serializedChunk.channelsMetadataList)
                 .setChunkMD5(md5)
                 .setEncryptionKeyId(firstChannelFlushContext.getEncryptionKeyId())

diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/ChunkMetadata.java b/src/main/java/net/snowflake/ingest/streaming/internal/ChunkMetadata.java
@@ -15,6 +15,7 @@ class ChunkMetadata {
   private final String tableName;
   private Long chunkStartOffset;
   private final Integer chunkLength;
+  private final Integer uncompressedChunkLength;
   private final List<ChannelMetadata> channels;
   private final String chunkMD5;
   private final EpInfo epInfo;
@@ -33,6 +34,9 @@ static class Builder {
     private String tableName;
     private Long chunkStartOffset;
     private Integer chunkLength; // compressedChunkLength
+
+    private Integer uncompressedChunkLength;
+
     private List<ChannelMetadata> channels;
     private String chunkMD5;
     private EpInfo epInfo;
@@ -62,6 +66,11 @@ Builder setChunkLength(Integer chunkLength) {
       return this;
     }
 
+    public Builder setUncompressedChunkLength(Integer uncompressedChunkLength) {
+      this.uncompressedChunkLength = uncompressedChunkLength;
+      return this;
+    }
+
     Builder setChannelList(List<ChannelMetadata> channels) {
       this.channels = channels;
       return this;
@@ -110,6 +119,7 @@ private ChunkMetadata(Builder builder) {
     this.tableName = builder.tableName;
     this.chunkStartOffset = builder.chunkStartOffset;
     this.chunkLength = builder.chunkLength;
+    this.uncompressedChunkLength = builder.uncompressedChunkLength;
     this.channels = builder.channels;
     this.chunkMD5 = builder.chunkMD5;
     this.epInfo = builder.epInfo;
@@ -152,6 +162,11 @@ Integer getChunkLength() {
     return chunkLength;
   }
 
+  @JsonProperty("chunk_length_uncompressed")
+  public Integer getUncompressedChunkLength() {
+    return uncompressedChunkLength;
+  }
+
   @JsonProperty("channels")
   List<ChannelMetadata> getChannels() {
     return this.channels;

diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/FlushServiceTest.java b/src/test/java/net/snowflake/ingest/streaming/internal/FlushServiceTest.java
@@ -795,6 +795,7 @@ public void testBlobBuilder() throws Exception {
             .setOwningTableFromChannelContext(channel1.getChannelContext())
             .setChunkStartOffset(0L)
             .setChunkLength(dataSize)
+            .setUncompressedChunkLength(dataSize * 2)
             .setChannelList(Collections.singletonList(channelMetadata))
             .setChunkMD5("md5")
             .setEncryptionKeyId(1234L)
@@ -819,7 +820,8 @@ public void testBlobBuilder() throws Exception {
               Arrays.copyOfRange(blob, offset, offset += BLOB_TAG_SIZE_IN_BYTES),
               StandardCharsets.UTF_8));
       Assert.assertEquals(
-          bdecVersion, Arrays.copyOfRange(blob, offset, offset += BLOB_VERSION_SIZE_IN_BYTES)[0]);
+          bdecVersion.toByte(),
+          Arrays.copyOfRange(blob, offset, offset += BLOB_VERSION_SIZE_IN_BYTES)[0]);
       long totalSize =
           ByteBuffer.wrap(Arrays.copyOfRange(blob, offset, offset += BLOB_FILE_SIZE_SIZE_IN_BYTES))
               .getLong();
@@ -844,6 +846,9 @@ public void testBlobBuilder() throws Exception {
       Assert.assertEquals(chunkMetadata.getTableName(), map.get("table"));
       Assert.assertEquals(chunkMetadata.getSchemaName(), map.get("schema"));
       Assert.assertEquals(chunkMetadata.getDBName(), map.get("database"));
+      Assert.assertEquals(chunkMetadata.getChunkLength(), map.get("chunk_length"));
+      Assert.assertEquals(
+          chunkMetadata.getUncompressedChunkLength(), map.get("chunk_length_uncompressed"));
       Assert.assertEquals(
           Long.toString(chunkMetadata.getChunkStartOffset() - offset),
           map.get("chunk_start_offset").toString());