Update Snowflake JDBC driver to 4.3.4

.github/workflows/build-report.yml

@@ -0,0 +1,50 @@
+# Copyright © 2024 Cask Data, Inc.
+#  Licensed under the Apache License, Version 2.0 (the "License"); you may not
+#  use this file except in compliance with the License. You may obtain a copy of
+#  the License at
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#  License for the specific language governing permissions and limitations under
+#  the License.
+
+# This workflow will build a Java project with Maven
+# For more information see: https://help.github.com/actions/language-and-framework-guides/building-and-testing-java-with-maven
+# Note: Any changes to this workflow would be used only after merging into develop
+name: Build Unit Tests Report
+
+on:
+  workflow_run:
+    workflows:
+      - Build with unit tests
+    types:
+      - completed
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    if: ${{ github.event.workflow_run.conclusion != 'skipped' }}
+
+    steps:
+      # Pinned 1.0.0 version
+      - uses: marocchino/action-workflow_run-status@54b6e87d6cb552fc5f36dbe9a722a6048725917a
+
+      - name: Download artifact
+        uses: actions/download-artifact@v4
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          run-id: ${{ github.event.workflow_run.id }}
+          path: artifacts/
+
+      - name: Surefire Report
+        # Pinned 3.5.2 version
+        uses: mikepenz/action-junit-report@16a9560bd02f11e7e3bf6b3e2ef6bba6c9d07c32
+        if: always()
+        with:
+          report_paths: '**/target/surefire-reports/TEST-*.xml'
+          github_token: ${{ secrets.GITHUB_TOKEN }}
+          detailed_summary: true
+          commit: ${{ github.event.workflow_run.head_sha }}
+          check_name: Build Test Report

.github/workflows/main.yml

@@ -0,0 +1,59 @@
+# Copyright © 2024 Cask Data, Inc.
+#  Licensed under the Apache License, Version 2.0 (the "License"); you may not
+#  use this file except in compliance with the License. You may obtain a copy of
+#  the License at
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#  License for the specific language governing permissions and limitations under
+#  the License.
+
+# This workflow will build a Java project with Maven
+# For more information see: https://help.github.com/actions/language-and-framework-guides/building-and-testing-java-with-maven
+# Note: Any changes to this workflow would be used only after merging into develop
+name: Build with unit tests
+
+on:
+  push:
+    branches: [ develop, release/** ]
+  pull_request:
+    branches: [ develop, release/** ]
+    types: [opened, synchronize, reopened, labeled]
+
+jobs:
+  build:
+    runs-on: k8s-runner-build
+
+    # We allow builds:
+    # 1) When it's a merge into a branch
+    # 2) For PRs that are labeled as build and
+    #  - It's a code change
+    #  - A build label was just added
+    # A bit complex, but prevents builds when other labels are manipulated
+    if: >
+      github.event_name == 'push'
+      || (contains(github.event.pull_request.labels.*.name, 'build')
+         && (github.event.action != 'labeled' || github.event.label.name == 'build')
+         )
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          ref: ${{ github.event.workflow_run.head_sha }}
+      - name: Cache
+        uses: actions/cache@v3
+        with:
+          path: ~/.m2/repository
+          key: ${{ runner.os }}-maven-${{ github.workflow }}-${{ hashFiles('**/pom.xml') }}
+          restore-keys: |
+            ${{ runner.os }}-maven-${{ github.workflow }}
+      - name: Build with Maven
+        run: mvn clean test -fae -T 2 -B -V -DcloudBuild -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.httpconnectionManager.ttlSeconds=25
+      - name: Archive build artifacts
+        uses: actions/upload-artifact@v4
+        if: always()
+        with:
+          name: reports-${{ github.run_id }}
+          path: |
+            **/target/rat.txt
+            **/target/surefire-reports/*

docs/Snowflake-batchsource.md

@@ -25,7 +25,9 @@ log in to Snowflake, minus the "snowflakecomputing.com"). E.g. "myaccount.us-cen
 
 **Role:** Role to use (e.g. `ACCOUNTADMIN`).
 
-**Import Query:** Query for data import.
+**Import Query Type**  - Method used to retrieve schema from the source.
+* **Table Name**: The name of the table to retrieve the schema.
+* **Import Query**: Query for data import.
 
 ### Credentials
 
@@ -68,7 +70,7 @@ Data Types Mapping
 ----------
 
 | Snowflake Data Types           | CDAP Schema Data Type | Comment                                                   |
-| ------------------------------ | --------------------- | --------------------------------------------------------- |
+| ------------------------------ |-----------------------| --------------------------------------------------------- |
 | NUMBER                         | decimal               | Default precision and scale are (38,0).                   |
 | DECIMAL                        | decimal               | Synonymous with NUMBER.                                   |
 | NUMERIC                        | decimal               | Synonymous with NUMBER.                                   |
@@ -87,10 +89,10 @@ Data Types Mapping
 | DATE                           | date                  |                                                           |
 | DATETIME                       | timestamp             | Alias for TIMESTAMP_NTZ                                   |
 | TIME                           | time                  |                                                           |
-| TIMESTAMP                      | timestamp/string      | Alias for one of the TIMESTAMP variations (TIMESTAMP_NTZ by default).|
+| TIMESTAMP                      | timestamp             | Alias for one of the TIMESTAMP variations (TIMESTAMP_NTZ by default).|
 | TIMESTAMP_LTZ                  | timestamp             | TIMESTAMP with local time zone; time zone, if provided, is not stored.|
 | TIMESTAMP_NTZ                  | timestamp             | TIMESTAMP with no time zone; time zone, if provided, is not stored.|
-| TIMESTAMP_TZ                   | string                | TIMESTAMP with time zone.                                 |
+| TIMESTAMP_TZ                   | timestamp             | TIMESTAMP with time zone.                                 |
 | VARIANT                        | string                | A tagged universal type, which can store values of any other type, including OBJECT and ARRAY, up to a maximum size of 16 MB compressed.                                    |
 | OBJECT                         | string                | This will return a json with the data |
 | ARRAY                          | string                | This will return a json with the data |                                                  |

pom.xml

@@ -21,16 +21,16 @@
 
   <groupId>io.cdap.plugin</groupId>
   <artifactId>snowflake-plugins</artifactId>
-  <version>1.1.0-SNAPSHOT</version>
+  <version>1.1.6-SNAPSHOT</version>
   <packaging>jar</packaging>
   <name>Snowflake plugins</name>
 
   <properties>
     <surefire.redirectTestOutputToFile>true</surefire.redirectTestOutputToFile>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <!-- version properties -->
-    <cdap.version>6.1.0-SNAPSHOT</cdap.version>
-    <hydrator.version>2.3.0-SNAPSHOT</hydrator.version>
+    <cdap.version>6.4.1</cdap.version>
+    <hydrator.version>2.6.1</hydrator.version>
     <commons.csv.version>1.6</commons.csv.version>
     <hadoop.version>2.8.0</hadoop.version>
     <spark2.version>2.3.1</spark2.version>
@@ -47,17 +47,13 @@
     <powermock.version>1.7.4</powermock.version>
     <guice.version>4.0</guice.version>
     <opencsv.version>2.4</opencsv.version>
-    <snowflake-jdbc.version>3.13.24</snowflake-jdbc.version>
+    <snowflake-jdbc.version>4.3.4</snowflake-jdbc.version>
   </properties>
 
   <repositories>
-    <repository>
-      <id>sonatype</id>
-      <url>https://oss.sonatype.org/content/groups/public</url>
-    </repository>
     <repository>
       <id>sonatype-snapshots</id>
-      <url>https://oss.sonatype.org/content/repositories/snapshots</url>
+      <url>https://central.sonatype.com/repository/maven-snapshots</url>
     </repository>
   </repositories>
 
@@ -398,8 +394,8 @@
           <version>1.1.0</version>
           <configuration>
             <cdapArtifacts>
-              <parent>system:cdap-data-pipeline[6.1.0-SNAPSHOT,7.0.0-SNAPSHOT)</parent>
-              <parent>system:cdap-data-streams[6.1.0-SNAPSHOT,7.0.0-SNAPSHOT)</parent>
+              <parent>system:cdap-data-pipeline[6.4.1-SNAPSHOT,7.0.0-SNAPSHOT)</parent>
+              <parent>system:cdap-data-streams[6.4.1-SNAPSHOT,7.0.0-SNAPSHOT)</parent>
             </cdapArtifacts>
           </configuration>
           <executions>

src/main/java/io/cdap/plugin/snowflake/common/BaseSnowflakeConfig.java

@@ -278,7 +278,7 @@ public boolean canConnect() {
       && !containsMacro(PROPERTY_PASSWORD) && !containsMacro(PROPERTY_WAREHOUSE)
       && !containsMacro(PROPERTY_ROLE) && !containsMacro(PROPERTY_CLIENT_ID)
       && !containsMacro(PROPERTY_CLIENT_SECRET) && !containsMacro(PROPERTY_REFRESH_TOKEN)
-      && !containsMacro(PROPERTY_PRIVATE_KEY));
+      && !containsMacro(PROPERTY_PRIVATE_KEY) && !containsMacro(PROPERTY_PASSPHRASE));
   }
 
   protected void validateConnection(FailureCollector collector) {
@@ -299,8 +299,8 @@ protected void validateConnection(FailureCollector collector) {
         .withConfigProperty(PROPERTY_USERNAME);
 
       // TODO: for oauth2
-      if (keyPairEnabled) {
-        failure.withConfigProperty(PROPERTY_PRIVATE_KEY);
+      if (Boolean.TRUE.equals(keyPairEnabled)) {
+        failure.withConfigProperty(PROPERTY_PRIVATE_KEY).withConfigProperty(PROPERTY_PASSPHRASE);
       } else {
         failure.withConfigProperty(PROPERTY_PASSWORD);
       }

src/main/java/io/cdap/plugin/snowflake/common/client/SnowflakeAccessor.java

@@ -23,7 +23,7 @@
 import io.cdap.plugin.snowflake.common.OAuthUtil;
 import io.cdap.plugin.snowflake.common.exception.ConnectionTimeoutException;
 import io.cdap.plugin.snowflake.common.util.QueryUtil;
-import net.snowflake.client.jdbc.SnowflakeBasicDataSource;
+import net.snowflake.client.internal.api.implementation.datasource.SnowflakeBasicDataSource;
 import org.apache.http.impl.client.HttpClients;
 
 import java.io.BufferedWriter;
@@ -32,6 +32,7 @@
 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.sql.Connection;
+import java.sql.DatabaseMetaData;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.ResultSetMetaData;
@@ -66,6 +67,30 @@ public void runSQL(String query) throws IOException {
 
   }
 
+  /**
+   * Returns field descriptors for specified table name
+   * @param schemaName The name of schema containing the table
+   * @param tableName The name of table whose metadata needs to be retrieved
+   * @return list of field descriptors
+   */
+  public List<SnowflakeFieldDescriptor> getFieldDescriptors(String schemaName, String tableName) throws IOException {
+    List<SnowflakeFieldDescriptor> fieldDescriptors = new ArrayList<>();
+    try (Connection connection = dataSource.getConnection()) {
+      DatabaseMetaData dbMetaData = connection.getMetaData();
+      try (ResultSet columns = dbMetaData.getColumns(null, schemaName, tableName, null)) {
+        while (columns.next()) {
+          String columnName = columns.getString("COLUMN_NAME");
+          int columnType = columns.getInt("DATA_TYPE");
+          boolean nullable = columns.getInt("NULLABLE") == DatabaseMetaData.columnNullable;
+          fieldDescriptors.add(new SnowflakeFieldDescriptor(columnName, columnType, nullable));
+        }
+      }
+    } catch (SQLException e) {
+      throw new IOException();
+    }
+    return fieldDescriptors;
+  }
+
   /**
    * Returns field descriptors for specified import query.
    *
@@ -110,7 +135,11 @@ private void initDataSource(SnowflakeBasicDataSource dataSource, BaseSnowflakeCo
 
     if (config.getOauth2Enabled()) {
       String accessToken = OAuthUtil.getAccessTokenByRefreshToken(HttpClients.createDefault(), config);
-      dataSource.setOauthToken(accessToken);
+      // In JDBC 4.x, setOauthToken() was removed. The recommended approach is to explicitly
+      // set the authenticator to "oauth" and pass the access token as the password.
+      // Migration guide: https://docs.snowflake.com/en/developer-guide/jdbc/jdbc-migration
+      dataSource.setAuthenticator("oauth");
+      dataSource.setPassword(accessToken);
     } else if (config.getKeyPairEnabled()) {
       dataSource.setUser(config.getUsername());
 
@@ -171,4 +200,13 @@ private static String writeTextToTmpFile(String text) {
       throw new RuntimeException("Cannot write key to temporary file", e);
     }
   }
+
+  /**
+   * Retrieves schema name from the configuration
+   *
+   * @return The schema name
+   */
+  public String getSchema() {
+    return config.getSchemaName();
+  }
 }

src/main/java/io/cdap/plugin/snowflake/common/util/QueryUtil.java

@@ -16,19 +16,24 @@
 
 package io.cdap.plugin.snowflake.common.util;
 
+import com.google.common.base.Strings;
+
 /**
  * Transforms import query.
  */
 public class QueryUtil {
 
   // Matches "limit <number>". Also "limit $$$$" and "limit ''" which means unlimited in Snowflake.
-  private static final String LIMIT_PATTERN = "(?i)LIMIT (''|\\$\\$\\$\\$|\\d+)";
+  private static final String LIMIT_PATTERN = "(?i)LIMIT (NULL|''|\\$\\$\\$\\$|\\d+)";
   private static final String LIMIT_STRING = "limit %s";
 
   private QueryUtil() {
   }
 
   public static String removeSemicolon(String importQuery) {
+    if (Strings.isNullOrEmpty(importQuery)) {
+      return null;
+    }
     if (importQuery.endsWith(";")) {
       importQuery = importQuery.substring(0, importQuery.length() - 1);
     }