Add JdbcSnowflakeClientTest using mocks

snowflake/src/main/java/org/apache/iceberg/snowflake/JdbcSnowflakeClient.java

@@ -26,6 +26,7 @@
 import org.apache.iceberg.jdbc.JdbcClientPool;
 import org.apache.iceberg.jdbc.UncheckedInterruptedException;
 import org.apache.iceberg.jdbc.UncheckedSQLException;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.snowflake.entities.SnowflakeSchema;
 import org.apache.iceberg.snowflake.entities.SnowflakeTable;
 import org.apache.iceberg.snowflake.entities.SnowflakeTableMetadata;
@@ -47,11 +48,17 @@ public class JdbcSnowflakeClient implements SnowflakeClient {
 
   private static final Logger LOG = LoggerFactory.getLogger(JdbcSnowflakeClient.class);
   private final JdbcClientPool connectionPool;
+  private QueryRunner queryRunner = new QueryRunner(true);
 
   JdbcSnowflakeClient(JdbcClientPool conn) {
     connectionPool = conn;
   }
 
+  @VisibleForTesting
+  void setQueryRunner(QueryRunner queryRunner) {
+    this.queryRunner = queryRunner;
+  }
+
   @Override
   public List<SnowflakeSchema> listSchemas(Namespace namespace) {
     StringBuilder baseQuery = new StringBuilder("SHOW SCHEMAS");
@@ -68,13 +75,13 @@ public List<SnowflakeSchema> listSchemas(Namespace namespace) {
 
     final String finalQuery = baseQuery.toString();
     final Object[] finalQueryParams = queryParams;
-    QueryRunner run = new QueryRunner(true);
     List<SnowflakeSchema> schemas;
     try {
       schemas =
           connectionPool.run(
               conn ->
-                  run.query(conn, finalQuery, SnowflakeSchema.createHandler(), finalQueryParams));
+                  queryRunner.query(
+                      conn, finalQuery, SnowflakeSchema.createHandler(), finalQueryParams));
     } catch (SQLException e) {
       throw new UncheckedSQLException(
           e,
@@ -111,13 +118,13 @@ public List<SnowflakeTable> listIcebergTables(Namespace namespace) {
 
     final String finalQuery = baseQuery.toString();
     final Object[] finalQueryParams = queryParams;
-    QueryRunner run = new QueryRunner(true);
     List<SnowflakeTable> tables;
     try {
       tables =
           connectionPool.run(
               conn ->
-                  run.query(conn, finalQuery, SnowflakeTable.createHandler(), finalQueryParams));
+                  queryRunner.query(
+                      conn, finalQuery, SnowflakeTable.createHandler(), finalQueryParams));
     } catch (SQLException e) {
       throw new UncheckedSQLException(
           e, "Failed to list tables for namespace %s", namespace.toString());
@@ -129,15 +136,13 @@ public List<SnowflakeTable> listIcebergTables(Namespace namespace) {
 
   @Override
   public SnowflakeTableMetadata getTableMetadata(TableIdentifier tableIdentifier) {
-    QueryRunner run = new QueryRunner(true);
-
     SnowflakeTableMetadata tableMeta;
     try {
       final String finalQuery = "SELECT SYSTEM$GET_ICEBERG_TABLE_INFORMATION(?) AS METADATA";
       tableMeta =
           connectionPool.run(
               conn ->
-                  run.query(
+                  queryRunner.query(
                       conn,
                       finalQuery,
                       SnowflakeTableMetadata.createHandler(),

snowflake/src/main/java/org/apache/iceberg/snowflake/entities/SnowflakeSchema.java

@@ -20,6 +20,7 @@
 
 import java.util.List;
 import org.apache.commons.dbutils.ResultSetHandler;
+import org.apache.iceberg.relocated.com.google.common.base.Objects;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 
 public class SnowflakeSchema {
@@ -39,6 +40,29 @@ public String getDatabase() {
     return databaseName;
   }
 
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    } else if (!(o instanceof SnowflakeSchema)) {
+      return false;
+    }
+
+    SnowflakeSchema that = (SnowflakeSchema) o;
+    return Objects.equal(this.databaseName, that.databaseName)
+        && Objects.equal(this.name, that.name);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(databaseName, name);
+  }
+
+  @Override
+  public String toString() {
+    return String.format("%s.%s", databaseName, name);
+  }
+
   public static ResultSetHandler<List<SnowflakeSchema>> createHandler() {
     return rs -> {
       List<SnowflakeSchema> schemas = Lists.newArrayList();

snowflake/src/main/java/org/apache/iceberg/snowflake/entities/SnowflakeTable.java

@@ -20,6 +20,7 @@
 
 import java.util.List;
 import org.apache.commons.dbutils.ResultSetHandler;
+import org.apache.iceberg.relocated.com.google.common.base.Objects;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 
 public class SnowflakeTable {
@@ -45,6 +46,30 @@ public String getSchemaName() {
     return schemaName;
   }
 
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    } else if (!(o instanceof SnowflakeTable)) {
+      return false;
+    }
+
+    SnowflakeTable that = (SnowflakeTable) o;
+    return Objects.equal(this.databaseName, that.databaseName)
+        && Objects.equal(this.schemaName, that.schemaName)
+        && Objects.equal(this.name, that.name);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(databaseName, schemaName, name);
+  }
+
+  @Override
+  public String toString() {
+    return String.format("%s.%s.%s", databaseName, schemaName, name);
+  }
+
   public static ResultSetHandler<List<SnowflakeTable>> createHandler() {
     return rs -> {
       List<SnowflakeTable> tables = Lists.newArrayList();

snowflake/src/main/java/org/apache/iceberg/snowflake/entities/SnowflakeTableMetadata.java

@@ -23,6 +23,7 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import org.apache.commons.dbutils.ResultSetHandler;
+import org.apache.iceberg.relocated.com.google.common.base.Objects;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.util.JsonUtil;
 
@@ -31,9 +32,12 @@ public class SnowflakeTableMetadata {
       Pattern.compile("azure://([^/]+)/([^/]+)/(.*)");
 
   private String snowflakeMetadataLocation;
-  private String status;
   private String icebergMetadataLocation;
+  private String status;
 
+  // Note: Since not all sources will necessarily come from a raw JSON representation, this raw
+  // JSON should only be considered a convenient debugging field. Equality of two
+  // SnowflakeTableMetadata instances should not depend on equality of this field.
   private String rawJsonVal;
 
   public SnowflakeTableMetadata(
@@ -61,6 +65,34 @@ public String getStatus() {
     return status;
   }
 
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    } else if (!(o instanceof SnowflakeTableMetadata)) {
+      return false;
+    }
+
+    // Only consider parsed fields, not the raw JSON that may or may not be the original source of
+    // this instance.
+    SnowflakeTableMetadata that = (SnowflakeTableMetadata) o;
+    return Objects.equal(this.snowflakeMetadataLocation, that.snowflakeMetadataLocation)
+        && Objects.equal(this.icebergMetadataLocation, that.icebergMetadataLocation)
+        && Objects.equal(this.status, that.status);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(snowflakeMetadataLocation, icebergMetadataLocation, status);
+  }
+
+  @Override
+  public String toString() {
+    return String.format(
+        "snowflakeMetadataLocation: '%s', icebergMetadataLocation: '%s', status: '%s",
+        snowflakeMetadataLocation, icebergMetadataLocation, status);
+  }
+
   /**
    * Translates from Snowflake's path syntax to Iceberg's path syntax for paths matching known
    * non-compatible Snowflake paths. Throws IllegalArgumentException if the prefix of the