SNOW-1950183 Resolve Iceberg ingestion filename mismatch between upload and registration (#958)

Author: sfc-gh-alhuang

src/main/java/net/snowflake/ingest/streaming/internal/FlushService.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2021-2024 Snowflake Computing Inc. All rights reserved.
+ * Copyright (c) 2021-2025 Snowflake Computing Inc. All rights reserved.
  */
 
 package net.snowflake.ingest.streaming.internal;
@@ -660,13 +660,8 @@ BlobMetadata upload(
 
     Timer.Context uploadContext = Utils.createTimerContext(this.owningClient.uploadLatency);
 
-    // The returned etag is non-empty ONLY in case of iceberg uploads. With iceberg files, the XP
-    // scanner expects the
-    // MD5 value to exactly match the etag value in S3. This assumption doesn't hold when multipart
-    // upload kicks in,
-    // causing scan time failures and table corruption. By plugging in the etag value instead of the
-    // md5 value,
-    Optional<String> etag = storage.put(blobPath, blob);
+    // The returned icebergPostUploadMetadata is non-empty if and only if it's iceberg uploads.
+    Optional<IcebergPostUploadMetadata> icebergPostUploadMetadata = storage.put(blobPath, blob);
 
     if (uploadContext != null) {
       blobStats.setUploadDurationMs(uploadContext);
@@ -677,17 +672,22 @@ BlobMetadata upload(
     }
 
     logger.logInfo(
-        "Finish uploading blob={}, size={}, timeInMillis={}, etag={}",
+        "Finish uploading blob={}, size={}, timeInMillis={}, icebergPostUploadMetadata={}",
         blobPath.fileRegistrationPath,
         blob.length,
         System.currentTimeMillis() - startTime,
-        etag);
+        icebergPostUploadMetadata);
 
     // at this point we know for sure if the BDEC file has data for more than one chunk, i.e.
     // spans mixed tables or not
     return BlobMetadata.createBlobMetadata(
-        blobPath.fileRegistrationPath,
-        etag.isPresent() ? etag.get() : BlobBuilder.computeMD5(blob),
+        icebergPostUploadMetadata
+            .map(IcebergPostUploadMetadata::getBlobPath)
+            .orElse(blobPath)
+            .fileRegistrationPath,
+        icebergPostUploadMetadata
+            .flatMap(IcebergPostUploadMetadata::getEtag)
+            .orElse(BlobBuilder.computeMD5(blob)),
         bdecVersion,
         metadata,
         blobStats,

src/main/java/net/snowflake/ingest/streaming/internal/IStorage.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2024 Snowflake Computing Inc. All rights reserved.
+ * Copyright (c) 2024-2025 Snowflake Computing Inc. All rights reserved.
  */
 
 package net.snowflake.ingest.streaming.internal;
@@ -17,8 +17,8 @@ interface IStorage {
    *
    * @param blobPath
    * @param blob
-   * @return The String ETag returned by the upload. Can be null in situations where the underlying
-   *     layer does not have an ETag to return.
+   * @return The IcebergPostUploadMetadata returned by the upload. Present iff and only if the
+   *     implementation is creating Iceberg blobs.
    */
-  Optional<String> put(BlobPath blobPath, byte[] blob);
+  Optional<IcebergPostUploadMetadata> put(BlobPath blobPath, byte[] blob);
 }

src/main/java/net/snowflake/ingest/streaming/internal/IStorageManager.java

@@ -1,16 +1,18 @@
 /*
- * Copyright (c) 2024 Snowflake Computing Inc. All rights reserved.
+ * Copyright (c) 2024-2025 Snowflake Computing Inc. All rights reserved.
  */
 
 package net.snowflake.ingest.streaming.internal;
 
+import com.google.common.annotations.VisibleForTesting;
 import java.util.Optional;
 
 /**
  * Interface to manage {@link InternalStage} and {@link PresignedUrlExternalVolume} for {@link
  * FlushService}
  */
-interface IStorageManager {
+@VisibleForTesting
+public interface IStorageManager {
 
   /** Default max upload retries for streaming ingest storage */
   int DEFAULT_MAX_UPLOAD_RETRIES = 5;

src/main/java/net/snowflake/ingest/streaming/internal/IcebergPostUploadMetadata.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright (c) 2025 Snowflake Computing Inc. All rights reserved.
+ */
+
+package net.snowflake.ingest.streaming.internal;
+
+import java.util.Optional;
+import javax.annotation.Nullable;
+
+/** Stores Iceberg post-upload metadata. */
+class IcebergPostUploadMetadata {
+  /**
+   * With iceberg files, the XP scanner expects the MD5 value to exactly match the etag value in S3.
+   * This assumption doesn't hold when multipart upload kicks in, causing scan time failures and
+   * table corruption. By plugging in the etag value instead of the md5 value, we can ensure that
+   * the XP scanner can successfully scan the file.
+   */
+  private final @Nullable String etag;
+
+  /** The final uploaded blob path of the file within the table's external volume. */
+  private final BlobPath blobPath;
+
+  /**
+   * Constructor for IcebergPostUploadMetadata.
+   *
+   * @param etag The etag of the uploaded file.
+   * @param blobPath The updated blob path of the file within the table's external volume.
+   */
+  IcebergPostUploadMetadata(@Nullable String etag, BlobPath blobPath) {
+    this.etag = etag;
+    this.blobPath = blobPath;
+  }
+
+  Optional<String> getEtag() {
+    return Optional.ofNullable(etag);
+  }
+
+  BlobPath getBlobPath() {
+    return blobPath;
+  }
+
+  @Override
+  public String toString() {
+    return String.format("IcebergPostUploadMetadata(etag=%s, blobPath=%s)", etag, blobPath);
+  }
+}