SNOW-1789749: Support regional GCS endpoints

src/main/java/net/snowflake/client/jdbc/SnowflakeFileTransferAgent.java

@@ -1111,8 +1111,16 @@ static StageInfo getStageInfo(JsonNode jsonNode, SFSession session) throws Snowf
     // specifically
     // for FIPS or VPCE S3 endpoint. SNOW-652696
     String endPoint = null;
-    if ("AZURE".equalsIgnoreCase(stageLocationType) || "S3".equalsIgnoreCase(stageLocationType)) {
+    if ("AZURE".equalsIgnoreCase(stageLocationType)
+        || "S3".equalsIgnoreCase(stageLocationType)
+        || "GCS".equalsIgnoreCase(stageLocationType)) {
       endPoint = jsonNode.path("data").path("stageInfo").findValue("endPoint").asText();
+      if ("GCS".equalsIgnoreCase(stageLocationType)
+          && endPoint != null
+          && (endPoint.trim().isEmpty() || "null".equals(endPoint))) {
+        // setting to null to preserve previous behaviour for GCS
+        endPoint = null;
+      }
     }
 
     String stgAcct = null;
@@ -1179,6 +1187,8 @@ static StageInfo getStageInfo(JsonNode jsonNode, SFSession session) throws Snowf
       }
     }
 
+    setupUseRegionalUrl(jsonNode, stageInfo);
+
     if (stageInfo.getStageType() == StageInfo.StageType.S3) {
       if (session == null) {
         // This node's value is set if PUT is used without Session. (For Snowpipe Streaming, we rely
@@ -1200,6 +1210,18 @@ static StageInfo getStageInfo(JsonNode jsonNode, SFSession session) throws Snowf
     return stageInfo;
   }
 
+  private static void setupUseRegionalUrl(JsonNode jsonNode, StageInfo stageInfo) {
+    if (stageInfo.getStageType() != StageInfo.StageType.GCS
+        && stageInfo.getStageType() != StageInfo.StageType.S3) {
+      return;
+    }
+    JsonNode useRegionalURLNode = jsonNode.path("data").path("stageInfo").path("useRegionalUrl");
+    if (!useRegionalURLNode.isMissingNode()) {
+      boolean useRegionalURL = useRegionalURLNode.asBoolean(false);
+      stageInfo.setUseRegionalUrl(useRegionalURL);
+    }
+  }
+
   /**
    * A helper method to verify if the local file path from GS matches what's parsed locally. This is
    * for security purpose as documented in SNOW-15153.

src/main/java/net/snowflake/client/jdbc/cloud/storage/SnowflakeGCSClient.java

@@ -18,9 +18,11 @@
 import com.google.api.gax.rpc.FixedHeaderProvider;
 import com.google.auth.oauth2.AccessToken;
 import com.google.auth.oauth2.GoogleCredentials;
+import com.google.cloud.NoCredentials;
 import com.google.cloud.storage.Blob;
 import com.google.cloud.storage.BlobId;
 import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.HttpStorageOptions;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobListOption;
 import com.google.cloud.storage.StorageException;
@@ -1312,6 +1314,8 @@ private void setupGCSClient(
       if (accessToken != null) {
         // We are authenticated with an oauth access token.
         StorageOptions.Builder builder = StorageOptions.newBuilder();
+        stage.gcsCustomEndpoint().ifPresent(builder::setHost);
+
         if (areDisabledGcsDefaultCredentials(session)) {
           logger.debug(
               "Adding explicit credentials to avoid default credential lookup by the GCS client");
@@ -1329,7 +1333,10 @@ private void setupGCSClient(
                 .getService();
       } else {
         // Use anonymous authentication.
-        this.gcsClient = StorageOptions.getUnauthenticatedInstance().getService();
+        HttpStorageOptions.Builder builder =
+            HttpStorageOptions.newBuilder().setCredentials(NoCredentials.getInstance());
+        stage.gcsCustomEndpoint().ifPresent(builder::setHost);
+        this.gcsClient = builder.build().getService();
       }
 
       if (encMat != null) {

src/main/java/net/snowflake/client/jdbc/cloud/storage/StageInfo.java

@@ -2,10 +2,17 @@
 
 import java.io.Serializable;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Properties;
+import net.snowflake.client.core.SnowflakeJdbcInternalApi;
 
-/** Encapsulates all the required stage properties used by GET/PUT for Azure and S3 stages */
+/** Encapsulates all the required stage properties used by GET/PUT for Azure, GCS and S3 stages */
 public class StageInfo implements Serializable {
+
+  // me-central2 GCS region always use regional urls
+  // TODO SNOW-1818804: the value is hardcoded now, but it should be server driven
+  private static final String GCS_REGION_ME_CENTRAL_2 = "me-central2";
+
   public enum StageType {
     S3,
     AZURE,
@@ -17,12 +24,18 @@ public enum StageType {
   private StageType stageType; // The stage type
   private String location; // The container or bucket
   private Map<?, ?> credentials; // the credentials required for the  stage
-  private String region; // AWS/S3/GCS region (S3/GCS only)
-  private String endPoint; // The Azure Storage endpoint (Azure only)
+  private String region; // S3/GCS region
+  // An endpoint (Azure, AWS FIPS and GCS custom endpoint override)
+  private String endPoint;
   private String storageAccount; // The Azure Storage account (Azure only)
   private String presignedUrl; // GCS gives us back a presigned URL instead of a cred
   private boolean isClientSideEncrypted; // whether to encrypt/decrypt files on the stage
-  private boolean useS3RegionalUrl; // whether to use s3 regional URL (AWS Only)
+  // whether to use s3 regional URL (AWS Only)
+  // TODO SNOW-1818804: this field will be deprecated when the server returns {@link
+  // #useRegionalUrl}
+  private boolean useS3RegionalUrl;
+  // whether to use regional URL (AWS and GCS only)
+  private boolean useRegionalUrl;
   private Properties proxyProperties;
 
   /*
@@ -166,16 +179,39 @@ public boolean getUseS3RegionalUrl() {
     return useS3RegionalUrl;
   }
 
+  @SnowflakeJdbcInternalApi
+  public void setUseRegionalUrl(boolean useRegionalUrl) {
+    this.useRegionalUrl = useRegionalUrl;
+  }
+
+  @SnowflakeJdbcInternalApi
+  public boolean getUseRegionalUrl() {
+    return useRegionalUrl;
+  }
+
   private static boolean isSpecified(String arg) {
     return !(arg == null || arg.equalsIgnoreCase(""));
   }
 
   public void setProxyProperties(Properties proxyProperties) {
     this.proxyProperties = proxyProperties;
   }
-  ;
 
   public Properties getProxyProperties() {
     return proxyProperties;
   }
+
+  @SnowflakeJdbcInternalApi
+  public Optional<String> gcsCustomEndpoint() {
+    if (stageType != StageType.GCS) {
+      return Optional.empty();
+    }
+    if (endPoint != null && !endPoint.trim().isEmpty() && !"null".equals(endPoint)) {
+      return Optional.of(endPoint);
+    }
+    if (GCS_REGION_ME_CENTRAL_2.equalsIgnoreCase(region) || useRegionalUrl) {
+      return Optional.of(String.format("storage.%s.rep.googleapis.com", region.toLowerCase()));
+    }
+    return Optional.empty();
+  }
 }

src/main/java/net/snowflake/client/jdbc/cloud/storage/StorageClientFactory.java

@@ -59,8 +59,9 @@ public SnowflakeStorageClient createClient(
     switch (stage.getStageType()) {
       case S3:
         boolean useS3RegionalUrl =
-            (stage.getUseS3RegionalUrl()
-                || (session != null && session.getUseRegionalS3EndpointsForPresignedURL()));
+            stage.getUseS3RegionalUrl()
+                || stage.getUseRegionalUrl()
+                || session != null && session.getUseRegionalS3EndpointsForPresignedURL();
         return createS3Client(
             stage.getCredentials(),
             parallel,