Merge branch 'master' into sfc-gh-anavalos-test-unshaded

Author: sfc-gh-anavalos

README.md

@@ -8,14 +8,15 @@ Snowflake Ingest Service Java SDK
 The Snowflake Ingest Service SDK allows users to ingest files into their
 Snowflake data warehouse in a programmatic fashion via key-pair
 authentication. Currently, we support ingestion through the following APIs:
+
 1. [Snowpipe](https://docs.snowflake.com/en/user-guide/data-load-snowpipe-rest-gs.html#client-requirement-java-or-python-sdk)
 2. [Snowpipe Streaming](https://docs.snowflake.com/en/user-guide/data-load-snowpipe-streaming-overview)
 
 # Dependencies
 
 The Snowflake Ingest Service SDK depends on the following libraries:
 
-* snowflake-jdbc (3.16.1+) 
+* snowflake-jdbc (3.16.1+)
 * slf4j-api
 * com.github.luben:zstd-jni (1.5.0-1)
 
@@ -63,13 +64,16 @@ dependencies {
 
 ## Jar Versions
 
-The Snowflake Ingest SDK provides shaded and unshaded versions of its jar. The shaded version bundles the dependencies into its own jar,
-whereas the unshaded version declares its dependencies in `pom.xml`, which are fetched as standard transitive dependencies by the build system like Maven or Gradle.
-The shaded JAR can help avoid potential dependency conflicts, but the unshaded version provides finer graned control over transitive dependencies.
+The Snowflake Ingest SDK provides shaded and unshaded versions of its jar. The shaded version bundles the dependencies
+into its own jar, whereas the unshaded version declares its dependencies in `pom.xml`, which are fetched as standard transitive
+dependencies by the build system like Maven or Gradle.
+The shaded JAR can help avoid potential dependency conflicts, but the unshaded version provides finer graned control
+over transitive dependencies.
 
 ## Using with snowflake-jdbc-fips
 
-For use cases, which need to use `snowflake-jdbc-fips` instead of the default `snowflake-jdbc`, we recommend to take the following steps:
+For use cases, which need to use `snowflake-jdbc-fips` instead of the default `snowflake-jdbc`, we recommend to take the
+following steps:
 
 - Use the unshaded version of the Ingest SDK.
 - Exclude these transitive dependencies:
@@ -78,7 +82,8 @@ For use cases, which need to use `snowflake-jdbc-fips` instead of the default `s
     - `org.bouncycastle:bcprov-jdk18on`
 - Add a dependency on `snowflake-jdbc-fips`.
 
-See [this test](https://github.com/snowflakedb/snowflake-ingest-java/tree/master/e2e-jar-test/fips) for an example how to use Snowflake Ingest SDK together with Snowflake FIPS JDBC Driver.
+See [this test](https://github.com/snowflakedb/snowflake-ingest-java/tree/master/e2e-jar-test/fips) for an example how
+to use Snowflake Ingest SDK together with Snowflake FIPS JDBC Driver.
 
 # Example
 
@@ -89,16 +94,23 @@ Check out `SnowflakeIngestBasicExample.java`
 ## Snowpipe Streaming
 
 Check out `SnowflakeStreamingIngestExample.java`, which performs following operations:
-1. Reads a JSON file which contains details regarding Snowflake Account, User, Role and Private Key. Take a look at `profile_streaming.json.example` for more details.
-    1. [Here](https://docs.snowflake.com/en/user-guide/key-pair-auth.html#configuring-key-pair-authentication) are the steps required to generate a private key.
-2. Creates a `SnowflakeStreamingIngestClient` which can be used to open one or more Streaming Channels pointing to the same or different tables.
+
+1. Reads a JSON file which contains details regarding Snowflake Account, User, Role and Private Key. Take a look at
+   `profile_streaming.json.example` for more details.
+    1. [Here](https://docs.snowflake.com/en/user-guide/key-pair-auth.html#configuring-key-pair-authentication) are the
+       steps required to generate a private key.
+2. Creates a `SnowflakeStreamingIngestClient` which can be used to open one or more Streaming Channels pointing to the
+   same or different tables.
 3. Creates a `SnowflakeStreamingIngestChannel` against a Database, Schema and Table.
-    1. Please note: The database, schema and table is expected to be present before opening the Channel. Example SQL queries to create them:
+    1. Please note: The database, schema and table is expected to be present before opening the Channel. Example SQL
+       queries to create them:
+
 ```sql
 create or replace database MY_DATABASE;
 create or replace schema MY_SCHEMA;
 create or replace table MY_TABLE(c1 number);
 ```
+
 4. Inserts 1000 rows into the channel created in 3rd step using the `insertRows` API on the Channel object
     1. `insertRows` API also takes in an optional `offsetToken` String which can be associated to this batch of rows.
 5. Calls `getLatestCommittedOffsetToken` on the channel until the appropriate offset is found in Snowflake.
@@ -123,28 +135,34 @@ mvn package
 However, for general usage, pulling a pre-built jar from maven is
 recommended.
 
-If you would like to run SnowflakeIngestBasicExample.java or SnowflakeStreamingIngestExample.java in the example folder, 
+If you would like to run SnowflakeIngestBasicExample.java or SnowflakeStreamingIngestExample.java in the example folder,
 please edit `pom.xml` and change the scope of the dependency `slf4j-simple` from `test` to `runtime` in order to enable
-console log output.
-
+console log output. For changing the logging level or message format, updating the system property or a
+simplelogger.properties file on the classpath is required, please
+see https://www.slf4j.org/api/org/slf4j/simple/SimpleLogger.html for more details. If you run into formatting errors,
+please run format.sh to format the code.
 
 # Testing (SimpleIngestIT Test)
 
--   Modify `TestUtils.java` file and replace *PROFILE_PATH* with `profile.json.example` for testing.
+- Modify `TestUtils.java` file and replace *PROFILE_PATH* with `profile.json.example` for testing.
 
-    -   `profile.json` is used because an encrypted file will be
-            decrypted for Github Actions testing. Check `End2EndTest.yml`
+    - `profile.json` is used because an encrypted file will be
+      decrypted for Github Actions testing. Check `End2EndTest.yml`
 
--   Use an unencrypted version(Only for testing) of private key while generating keys(private and public pair) using OpenSSL.
+- Use an unencrypted version(Only for testing) of private key while generating keys(private and public pair) using
+  OpenSSL.
 
-    -   Here is the link for documentation [Key Pair
-            Generator](https://docs.snowflake.com/en/user-guide/key-pair-auth.html)
+    - Here is the link for documentation [Key Pair
+      Generator](https://docs.snowflake.com/en/user-guide/key-pair-auth.html)
 
 # Contributing to this repo
 
-Each PR must pass all required github action merge gates before approval and merge. In addition to those tests, you will need:
+Each PR must pass all required github action merge gates before approval and merge. In addition to those tests, you will
+need:
 
-- Formatter: run this script [`./format.sh`](https://github.com/snowflakedb/snowflake-ingest-java/blob/master/format.sh) from root
-- CLA: all contributers must sign the Snowflake CLA. This is a one time signature, please provide your email so we can work with you to get this signed after you open a PR.
+- Formatter: run this script [`./format.sh`](https://github.com/snowflakedb/snowflake-ingest-java/blob/master/format.sh)
+  from root
+- CLA: all contributers must sign the Snowflake CLA. This is a one time signature, please provide your email so we can
+  work with you to get this signed after you open a PR.
 
 Thank you for contributing! We will review and approve PRs as soon as we can.

e2e-jar-test/pom.xml

@@ -32,7 +32,7 @@
       <dependency>
         <groupId>net.snowflake</groupId>
         <artifactId>snowflake-ingest-sdk</artifactId>
-        <version>3.1.1</version>
+        <version>3.1.2</version>
       </dependency>
 
       <dependency>

pom.xml

@@ -9,7 +9,7 @@
   <!-- Arifact name and version information -->
   <groupId>net.snowflake</groupId>
   <artifactId>snowflake-ingest-sdk</artifactId>
-  <version>3.1.1</version>
+  <version>3.1.2</version>
   <packaging>jar</packaging>
   <name>Snowflake Ingest SDK</name>
   <description>Snowflake Ingest SDK</description>

src/main/java/net/snowflake/ingest/connection/RequestBuilder.java

@@ -110,7 +110,7 @@ public class RequestBuilder {
   // Don't change!
   public static final String CLIENT_NAME = "SnowpipeJavaSDK";
 
-  public static final String DEFAULT_VERSION = "3.1.1";
+  public static final String DEFAULT_VERSION = "3.1.2";
 
   public static final String JAVA_USER_AGENT = "JAVA";
 

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