[SNOW-3249917] JDBC removal Step 8d: Replicate storage client implementations

.plans/JDBC_REMOVAL_PLAN.md

@@ -253,6 +253,35 @@ is replaced.
 
 ---
 
+### Progress Summary (updated 2026-03-26)
+
+| Step | Status | PR |
+|---|---|---|
+| Step 1 — Simple utilities | ✅ MERGED | #1100 |
+| Step 2 — Enums, constants, proxy utils | ✅ MERGED | #1109 |
+| Step 2b — SFLogger infrastructure | ✅ MERGED | #1112 |
+| Step 3 — Storage data types + utils | ✅ MERGED | #1110 |
+| Step 4 — HTTP settings + SSL factory | ✅ MERGED | #1111 |
+| Step 5a — ErrorCode, ResourceBundleManager | ✅ MERGED | #1113 |
+| Step 5b — SnowflakeSQLException, SnowflakeSQLLoggedException | ✅ MERGED | #1114 |
+| Step 6 — Telemetry | ✅ MERGED | #1115 |
+| Step 7a — Inline error helpers, swap exceptions | ✅ Open | #1116 |
+| Step 7b — Replicate getFileTransferMetadatas | ✅ Open | #1119 |
+| Step 8a — Replicate StageInfo, RemoteStoreFileEncryptionMaterial | ✅ Open | #1120 |
+| Step 8b — Replicate MetadataV1, ObjectMapperFactory, SqlState | ✅ Open | #1121 |
+| Step 8c — Replicate storage helpers + interface | ✅ Open | #1123 |
+| Step 8d — Replicate storage client implementations | ⬜ TODO | — |
+| Step 8e — Swap ALL imports at once | ⬜ TODO | — |
+| Step 9a — Clean up replicated classes' JDBC imports | ⬜ TODO | — |
+| Step 9b — Clean up SnowflakeSQLLoggedException telemetry | ⬜ TODO | — |
+| Step 9c — Clean up TelemetryClient JDBC imports | ⬜ TODO | — |
+| Step 10 — Remove JDBC dependency | ⬜ TODO | — |
+
+**Closed PRs:** #1117 (reverted 7b approach), #1122 (reverted 8c approach)
+**Other PRs:** #1118 (error/exception tests on master)
+
+---
+
 ### Step 1 — Simple utilities ✅ MERGED
 
 **Done:** `Power10`, `SFPair`, `Stopwatch`. Swap imports.
@@ -457,14 +486,60 @@ Swap imports everywhere. Now `InternalStage` can use ingest's types throughout.
 
 ---
 
-### Step 8c — Route uploadWithoutConnection through IcebergFileTransferAgent ⬜ TODO
+### Step 8c — Replicate storage helper classes and interface ✅ OPEN (PR #1123)
+
+**Done:** Verbatim replication of JDBC storage infrastructure:
+- `SnowflakeFileTransferConfig` (237 lines) — config builder
+- `SnowflakeStorageClient` (452 lines) — storage client interface
+- `MatDesc` (101 lines) — encryption material descriptor
+- `EncryptionProvider` (214 lines) — AES CBC encryption
+- `GcmEncryptionProvider` (254 lines) — AES GCM encryption
+- `StorageProviderException` (51 lines) — storage error wrapper
+- `StorageObjectSummary` (161 lines) — storage object properties
+- `StorageObjectSummaryCollection` (60 lines) — iterator (stubbed)
+
+---
+
+### Step 8d — Replicate storage client implementations ⬜ TODO
+
+Replicate the three cloud storage client classes and their factory verbatim
+from JDBC. These implement `SnowflakeStorageClient` (from 8c) and provide
+the upload/download operations for each cloud provider.
+
+**Classes to replicate:**
+- `StorageClientFactory` (~234 lines) — creates cloud-specific clients
+- `SnowflakeS3Client` (~1038 lines) — S3 upload/download with encryption
+- `SnowflakeAzureClient` (~1055 lines) — Azure Blob upload/download
+- `SnowflakeGCSClient` (~1282 lines) — GCS upload/download with presigned URLs
+- `S3ObjectMetadata` / `CommonObjectMetadata` — `StorageObjectMetadata` impls
+  used by `StorageClientFactory.createStorageMetadataObj()`
+- `HttpHeadersCustomizer` interface + `HeaderCustomizerHttpRequestInterceptor`
+  — HTTP header customization for S3
+
+**Also add to replicated `SnowflakeFileTransferAgent`:**
+- `uploadWithoutConnection(SnowflakeFileTransferConfig)` — main upload method
+- `pushFileToRemoteStore(...)` — S3/Azure upload helper
+- `pushFileToRemoteStoreWithPresignedUrl(...)` — GCS upload helper
+- `computeDigest(...)`, `compressStreamWithGZIP(...)`,
+  `compressStreamWithGZIPNoDigest(...)` — stream processing helpers
+
+**Note:** Both Iceberg clients and replicated Snowflake clients will coexist.
+`SFSession` references kept temporarily (always null from callers).
+
+---
+
+### Step 8e — Swap ALL imports at once ⬜ TODO
 
-Replace the non-Iceberg upload path. `IcebergFileTransferAgent` IS the
-already-existing verbatim replication of JDBC's upload stack (using
-`IcebergStorageClient` implementations instead of JDBC's `SnowflakeStorageClient`).
+Now that the full JDBC storage stack is replicated (Steps 8a–8d), swap all
+remaining JDBC imports to ingest versions in a single step:
 
-- Route non-Iceberg `uploadWithoutConnection` through `IcebergFileTransferAgent`
-- Remove `SnowflakeFileTransferConfig`, JDBC `SnowflakeFileTransferAgent` import
+- `InternalStage`: swap all 6 JDBC imports, use replicated
+  `SnowflakeFileTransferAgent.uploadWithoutConnection` and
+  `getFileTransferMetadatas`, remove `parseConfigureResponseMapper`
+- `InternalStageManager`: swap `SnowflakeSQLException`
+- `SnowflakeFileTransferMetadataWithAge`: swap `SnowflakeFileTransferMetadataV1`
+- All `fileTransferAgent` files: remove JDBC imports (now same-package)
+- `InternalStageTest`: update imports
 
 **InternalStage now fully free of JDBC imports.**
 
@@ -476,11 +551,13 @@ Remove JDBC imports from the replicated classes themselves:
 
 - `SqlState` in `ErrorCode`, `SnowflakeFileTransferAgent` → already replicated
   in same package, just swap import
-- `SFSession` parameter in `SnowflakeFileTransferAgent.getStageInfo()` → always
-  null from our callers, replace parameter type with stub or remove
+- `SFSession` parameter in `SnowflakeFileTransferAgent.getStageInfo()` and
+  storage clients → always null from our callers, replace with stub or remove
 - `SFException` in `SnowflakeSQLException` → used in one constructor never called
   by ingest; remove constructor or stub
 - `SecretDetector` in `TelemetryData` → already replicated (Step 2b)
+- `HttpUtil.isSocksProxyDisabled()` in `StorageClientFactory` → already in
+  ingest's `HttpUtil`
 
 ---
 

pom.xml

@@ -686,6 +686,22 @@
         </exclusion>
       </exclusions>
     </dependency>
+    <dependency>
+      <groupId>com.google.auth</groupId>
+      <artifactId>google-auth-library-credentials</artifactId>
+      <version>1.29.0</version>
+    </dependency>
+    <dependency>
+      <groupId>com.google.auth</groupId>
+      <artifactId>google-auth-library-oauth2-http</artifactId>
+      <version>1.29.0</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.google.errorprone</groupId>
+          <artifactId>error_prone_annotations</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-core</artifactId>

src/main/java/net/snowflake/ingest/streaming/internal/fileTransferAgent/CommonObjectMetadata.java

@@ -0,0 +1,84 @@
+/*
+ * Replicated from snowflake-jdbc (v3.25.1)
+ * Source: https://github.com/snowflakedb/snowflake-jdbc/blob/v3.25.1/src/main/java/net/snowflake/client/jdbc/cloud/storage/CommonObjectMetadata.java
+ *
+ * Permitted differences: package, SnowflakeUtil.createCaseInsensitiveMap replaced
+ * with StorageClientUtil.createCaseInsensitiveMap.
+ */
+package net.snowflake.ingest.streaming.internal.fileTransferAgent;
+
+import java.util.Map;
+import java.util.TreeMap;
+
+/**
+ * Implements platform-independent interface Azure BLOB and GCS object metadata
+ *
+ * <p>Only the metadata accessors and mutators used by the JDBC client currently are supported,
+ * additional methods should be added as needed
+ */
+public class CommonObjectMetadata implements StorageObjectMetadata {
+  private long contentLength;
+  private final Map<String, String> userDefinedMetadata;
+  private String contentEncoding;
+
+  CommonObjectMetadata() {
+    userDefinedMetadata = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+  }
+
+  /*
+   * Constructs a common metadata object
+   * from the set of parameters that the JDBC client is using
+   */
+  CommonObjectMetadata(
+      long contentLength, String contentEncoding, Map<String, String> userDefinedMetadata) {
+    this.contentEncoding = contentEncoding;
+    this.contentLength = contentLength;
+    this.userDefinedMetadata = StorageClientUtil.createCaseInsensitiveMap(userDefinedMetadata);
+  }
+
+  /**
+   * @return returns a Map/key-value pairs of metadata properties
+   */
+  @Override
+  public Map<String, String> getUserMetadata() {
+    return userDefinedMetadata;
+  }
+
+  /**
+   * @return returns the size of object in bytes
+   */
+  @Override
+  public long getContentLength() {
+    return contentLength;
+  }
+
+  /** Sets size of the associated object in bytes */
+  @Override
+  public void setContentLength(long contentLength) {
+    this.contentLength = contentLength;
+  }
+
+  /** Adds the key value pair of custom user-metadata for the associated object. */
+  @Override
+  public void addUserMetadata(String key, String value) {
+    userDefinedMetadata.put(key, value);
+  }
+
+  /**
+   * Sets the optional Content-Encoding HTTP header specifying what content encodings, have been
+   * applied to the object and what decoding mechanisms must be applied, in order to obtain the
+   * media-type referenced by the Content-Type field.
+   */
+  @Override
+  public void setContentEncoding(String encoding) {
+    contentEncoding = encoding;
+  }
+
+  /*
+   * @return returns the content encoding type
+   */
+  @Override
+  public String getContentEncoding() {
+    return contentEncoding;
+  }
+}

src/main/java/net/snowflake/ingest/streaming/internal/fileTransferAgent/FileCompressionType.java

@@ -0,0 +1,84 @@
+/*
+ * Replicated from snowflake-jdbc-thin (v3.25.1):
+ *   net.snowflake.client.jdbc.internal.snowflake.common.core.FileCompressionType
+ * Originally from net.snowflake:snowflake-common (decompiled from JDBC thin jar).
+ *
+ * Permitted differences: package.
+ */
+package net.snowflake.ingest.streaming.internal.fileTransferAgent;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/** File compression type enum. */
+public enum FileCompressionType {
+  GZIP(".gz", "application", Arrays.asList("gzip", "x-gzip"), true),
+  DEFLATE(".deflate", "application", Arrays.asList("zlib", "deflate"), true),
+  RAW_DEFLATE(".raw_deflate", "application", Arrays.asList("raw_deflate"), true),
+  BZIP2(".bz2", "application", Arrays.asList("bzip2", "x-bzip2", "x-bz2", "x-bzip", "bz2"), true),
+  ZSTD(".zst", "application", Arrays.asList("zstd"), true),
+  BROTLI(".br", "application", Arrays.asList("brotli", "x-brotli"), true),
+  LZIP(".lz", "application", Arrays.asList("lzip", "x-lzip"), false),
+  LZMA(".lzma", "application", Arrays.asList("lzma", "x-lzma"), false),
+  LZO(".lzo", "application", Arrays.asList("lzo", "x-lzop"), false),
+  XZ(".xz", "application", Arrays.asList("xz", "x-xz"), false),
+  COMPRESS(".Z", "application", Arrays.asList("compress", "x-compress"), false),
+  PARQUET(".parquet", "application", Arrays.asList("parquet"), true),
+  ORC(".orc", "application", Arrays.asList("orc"), true);
+
+  private final String fileExtension;
+  private final String mimeType;
+  private final List<String> mimeSubTypes;
+  private final boolean supported;
+
+  static final Map<String, FileCompressionType> mimeSubTypeToCompressionMap;
+
+  static {
+    mimeSubTypeToCompressionMap = new HashMap<>();
+    for (FileCompressionType type : values()) {
+      for (String mimeSubType : type.mimeSubTypes) {
+        mimeSubTypeToCompressionMap.put(mimeSubType, type);
+      }
+    }
+  }
+
+  FileCompressionType(
+      String fileExtension, String mimeType, List<String> mimeSubTypes, boolean supported) {
+    this.fileExtension = fileExtension;
+    this.mimeType = mimeType;
+    this.mimeSubTypes = mimeSubTypes;
+    this.supported = supported;
+  }
+
+  public static Optional<FileCompressionType> lookupByMimeSubType(String mimeSubType) {
+    return Optional.ofNullable(mimeSubTypeToCompressionMap.get(mimeSubType));
+  }
+
+  public static Optional<FileCompressionType> lookupByFileExtension(String fileExtension) {
+    for (FileCompressionType type : values()) {
+      if (type.fileExtension.equalsIgnoreCase(fileExtension)) {
+        return Optional.of(type);
+      }
+    }
+    return Optional.empty();
+  }
+
+  public boolean isSupported() {
+    return supported;
+  }
+
+  public String getFileExtension() {
+    return fileExtension;
+  }
+
+  public String getMimeType() {
+    return mimeType;
+  }
+
+  public List<String> getMimeSubTypes() {
+    return mimeSubTypes;
+  }
+}

src/main/java/net/snowflake/ingest/streaming/internal/fileTransferAgent/GCSAccessStrategy.java

@@ -0,0 +1,50 @@
+/*
+ * Replicated from snowflake-jdbc (v3.25.1)
+ * Source: https://github.com/snowflakedb/snowflake-jdbc/blob/v3.25.1/src/main/java/net/snowflake/client/jdbc/cloud/storage/GCSAccessStrategy.java
+ *
+ * Permitted differences: package, SFSession kept from JDBC temporarily,
+ * SFPair uses ingest version, all storage types use ingest versions (same package).
+ */
+package net.snowflake.ingest.streaming.internal.fileTransferAgent;
+
+import java.io.File;
+import java.io.InputStream;
+import java.util.Map;
+import net.snowflake.client.core.SFSession;
+import net.snowflake.ingest.utils.SFPair;
+
+interface GCSAccessStrategy {
+  StorageObjectSummaryCollection listObjects(String remoteStorageLocation, String prefix);
+
+  StorageObjectMetadata getObjectMetadata(String remoteStorageLocation, String prefix);
+
+  Map<String, String> download(
+      int parallelism, String remoteStorageLocation, String stageFilePath, File localFile)
+      throws InterruptedException;
+
+  SFPair<InputStream, Map<String, String>> downloadToStream(
+      String remoteStorageLocation, String stageFilePath, boolean isEncrypting);
+
+  void uploadWithDownScopedToken(
+      int parallelism,
+      String remoteStorageLocation,
+      String destFileName,
+      String contentEncoding,
+      Map<String, String> metadata,
+      long contentLength,
+      InputStream content,
+      String queryId)
+      throws InterruptedException;
+
+  boolean handleStorageException(
+      Exception ex,
+      int retryCount,
+      String operation,
+      SFSession session,
+      String command,
+      String queryId,
+      SnowflakeGCSClient gcsClient)
+      throws SnowflakeSQLException, net.snowflake.client.jdbc.SnowflakeSQLException;
+
+  void shutdown();
+}