[FLINK-33761][Connector/JDBC] support snowflake dialect

Author: boris-snyk

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/databases/snowflake/dialect/SnowflakeDialect.java

@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.flink.connector.jdbc.databases.snowflake.dialect;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.jdbc.converter.JdbcRowConverter;
+import org.apache.flink.connector.jdbc.dialect.AbstractPostgresCompatibleDialect;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.Optional;
+
+/** JDBC dialect for Snowflake. */
+@Internal
+public class SnowflakeDialect extends AbstractPostgresCompatibleDialect {
+    private static final long serialVersionUID = 1L;
+
+    @Override
+    public JdbcRowConverter getRowConverter(RowType rowType) {
+        return new SnowflakeRowConverter(rowType);
+    }
+
+    @Override
+    public Optional<String> defaultDriverName() {
+        return Optional.of("net.snowflake.client.jdbc.SnowflakeDriver");
+    }
+
+    @Override
+    public String dialectName() {
+        return "Snowflake";
+    }
+}

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/databases/snowflake/dialect/SnowflakeDialectFactory.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.flink.connector.jdbc.databases.snowflake.dialect;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.jdbc.dialect.JdbcDialect;
+import org.apache.flink.connector.jdbc.dialect.JdbcDialectFactory;
+
+/** Factory for {@link SnowflakeDialect}. */
+@Internal
+public class SnowflakeDialectFactory implements JdbcDialectFactory {
+    @Override
+    public boolean acceptsURL(String url) {
+        return url.startsWith("jdbc:snowflake:");
+    }
+
+    @Override
+    public JdbcDialect create() {
+        return new SnowflakeDialect();
+    }
+}

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/databases/snowflake/dialect/SnowflakeRowConverter.java

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.flink.connector.jdbc.databases.snowflake.dialect;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.jdbc.converter.AbstractPostgresCompatibleRowConverter;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.postgresql.jdbc.PgArray;
+
+/**
+ * Runtime converter that responsible to convert between JDBC object and Flink internal object for
+ * Snowflake.
+ */
+@Internal
+public class SnowflakeRowConverter extends AbstractPostgresCompatibleRowConverter<PgArray> {
+
+    private static final long serialVersionUID = 1L;
+
+    @Override
+    public String converterName() {
+        return "Snowflake";
+    }
+
+    // https://docs.snowflake.com/en/sql-reference/intro-summary-data-types
+    public SnowflakeRowConverter(RowType rowType) {
+        super(rowType);
+    }
+}

flink-connector-jdbc/src/main/resources/META-INF/services/org.apache.flink.connector.jdbc.dialect.JdbcDialectFactory

@@ -22,3 +22,4 @@ org.apache.flink.connector.jdbc.databases.sqlserver.dialect.SqlServerDialectFact
 org.apache.flink.connector.jdbc.databases.cratedb.dialect.CrateDBDialectFactory
 org.apache.flink.connector.jdbc.databases.db2.dialect.Db2DialectFactory
 org.apache.flink.connector.jdbc.databases.trino.dialect.TrinoDialectFactory
+org.apache.flink.connector.jdbc.databases.snowflake.dialect.SnowflakeDialectFactory