Add new SnowflakeCatalog implementation to enable directly using Snowflake-managed Iceberg tables

.github/labeler.yml

@@ -81,4 +81,6 @@ ALIYUN:
 GCP:
   - gcp/**/*
 DELL:
-  - dell/**/*
+  - dell/**/*
+SNOWFLAKE:
+  - snowflake/**/*

build.gradle

@@ -697,6 +697,24 @@ project(':iceberg-dell') {
   }
 }
 
+project(':iceberg-snowflake') {
+  test {
+    useJUnitPlatform()
+  }
+
+  dependencies {
+    implementation project(':iceberg-core')
+    implementation project(':iceberg-common')
+    implementation project(path: ':iceberg-bundled-guava', configuration: 'shadow')
+    implementation "com.fasterxml.jackson.core:jackson-databind"
+    implementation "com.fasterxml.jackson.core:jackson-core"
+
+    runtimeOnly("net.snowflake:snowflake-jdbc")
+
+    testImplementation project(path: ':iceberg-core', configuration: 'testArtifacts')
+  }
+}
+
 @Memoized
 boolean versionFileExists() {
   return file('version.txt').exists()

core/src/main/java/org/apache/iceberg/jdbc/JdbcClientPool.java

@@ -27,12 +27,12 @@
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.ClientPoolImpl;
 
-class JdbcClientPool extends ClientPoolImpl<Connection, SQLException> {
+public class JdbcClientPool extends ClientPoolImpl<Connection, SQLException> {
 
   private final String dbUrl;
   private final Map<String, String> properties;
 
-  JdbcClientPool(String dbUrl, Map<String, String> props) {
+  public JdbcClientPool(String dbUrl, Map<String, String> props) {
     this(
         Integer.parseInt(
             props.getOrDefault(
@@ -42,7 +42,7 @@ class JdbcClientPool extends ClientPoolImpl<Connection, SQLException> {
         props);
   }
 
-  JdbcClientPool(int poolSize, String dbUrl, Map<String, String> props) {
+  public JdbcClientPool(int poolSize, String dbUrl, Map<String, String> props) {
     super(poolSize, SQLNonTransientConnectionException.class, true);
     properties = props;
     this.dbUrl = dbUrl;

settings.gradle

@@ -34,6 +34,7 @@ include 'hive-metastore'
 include 'nessie'
 include 'gcp'
 include 'dell'
+include 'snowflake'
 
 project(':api').name = 'iceberg-api'
 project(':common').name = 'iceberg-common'
@@ -51,6 +52,7 @@ project(':hive-metastore').name = 'iceberg-hive-metastore'
 project(':nessie').name = 'iceberg-nessie'
 project(':gcp').name = 'iceberg-gcp'
 project(':dell').name = 'iceberg-dell'
+project(':snowflake').name = 'iceberg-snowflake'
 
 if (null != System.getProperty("allVersions")) {
   System.setProperty("flinkVersions", System.getProperty("knownFlinkVersions"))