Bump flink to 1 16

pom.xml

@@ -36,9 +36,9 @@ under the License.
 
     <properties>
         <java.version>11</java.version>
-        <flink.version>1.14.6</flink.version>
+        <flink.version>1.16.3</flink.version>
         <log4j.version>2.12.1</log4j.version>
-        <scala.binary.version>2.11</scala.binary.version>
+        <scala.binary.version>2.12</scala.binary.version>
         <ignite.version>2.10.0</ignite.version>
         <maven.compiler.source>${java.version}</maven.compiler.source>
         <maven.compiler.target>${java.version}</maven.compiler.target>
@@ -86,14 +86,14 @@ under the License.
 
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
+            <artifactId>flink-table-api-java-bridge</artifactId>
             <version>${flink.version}</version>
             <scope>provided</scope>
         </dependency>
 
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
+            <artifactId>flink-connector-jdbc</artifactId>
             <version>${flink.version}</version>
             <scope>provided</scope>
         </dependency>
@@ -135,7 +135,7 @@ under the License.
 
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-clients_${scala.binary.version}</artifactId>
+            <artifactId>flink-clients</artifactId>
             <version>${flink.version}</version>
             <scope>test</scope>
         </dependency>

src/main/java/pl/touk/flink/ignite/converter/IgniteRowConverter.java

@@ -1,6 +1,6 @@
 package pl.touk.flink.ignite.converter;
 
-import org.apache.flink.connector.jdbc.internal.converter.AbstractJdbcRowConverter;
+import org.apache.flink.connector.jdbc.converter.AbstractJdbcRowConverter;
 import org.apache.flink.table.types.logical.RowType;
 
 public class IgniteRowConverter extends AbstractJdbcRowConverter {

src/main/java/pl/touk/flink/ignite/dialect/IgniteDialect.java

@@ -1,26 +1,25 @@
 package pl.touk.flink.ignite.dialect;
 
-import org.apache.flink.connector.jdbc.dialect.JdbcDialect;
-import org.apache.flink.connector.jdbc.internal.converter.JdbcRowConverter;
+import org.apache.flink.connector.jdbc.converter.JdbcRowConverter;
+import org.apache.flink.connector.jdbc.dialect.AbstractDialect;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
 import org.apache.flink.table.types.logical.RowType;
 import pl.touk.flink.ignite.converter.IgniteRowConverter;
 
+import java.util.Arrays;
+import java.util.EnumSet;
 import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
 
-public class IgniteDialect implements JdbcDialect {
-
+public class IgniteDialect extends AbstractDialect {
     private static final long serialVersionUID = 1L;
 
     @Override
     public String dialectName() {
         return "Ignite";
     }
 
-    @Override
-    public boolean canHandle(String url) {
-        return url.startsWith("jdbc:ignite:thin");
-    }
-
     @Override
     public JdbcRowConverter getRowConverter(RowType rowType) {
         return new IgniteRowConverter(rowType);
@@ -31,9 +30,72 @@ public String getLimitClause(long l) {
         return "LIMIT " + l;
     }
 
+    @Override
+    public String quoteIdentifier(String identifier) {
+        return "\"" + identifier + "\"";
+    }
+
     @Override
     public Optional<String> defaultDriverName() {
         return Optional.of("org.apache.ignite.IgniteJdbcThinDriver");
     }
 
+    // WARNING code below was taken from PostgresDialect, it can turn out to be wrong
+
+    private static final int MAX_TIMESTAMP_PRECISION = 6;
+    private static final int MIN_TIMESTAMP_PRECISION = 1;
+
+    private static final int MAX_DECIMAL_PRECISION = 1000;
+    private static final int MIN_DECIMAL_PRECISION = 1;
+
+    @Override
+    public Optional<String> getUpsertStatement(
+            String tableName, String[] fieldNames, String[] uniqueKeyFields) {
+        String uniqueColumns =
+                Arrays.stream(uniqueKeyFields)
+                        .map(this::quoteIdentifier)
+                        .collect(Collectors.joining(", "));
+        String updateClause =
+                Arrays.stream(fieldNames)
+                        .map(f -> quoteIdentifier(f) + "=EXCLUDED." + quoteIdentifier(f))
+                        .collect(Collectors.joining(", "));
+        return Optional.of(
+                getInsertIntoStatement(tableName, fieldNames)
+                        + " ON CONFLICT ("
+                        + uniqueColumns
+                        + ")"
+                        + " DO UPDATE SET "
+                        + updateClause);
+    }
+
+    @Override
+    public Optional<Range> decimalPrecisionRange() {
+        return Optional.of(Range.of(MIN_DECIMAL_PRECISION, MAX_DECIMAL_PRECISION));
+    }
+
+    @Override
+    public Optional<Range> timestampPrecisionRange() {
+        return Optional.of(Range.of(MIN_TIMESTAMP_PRECISION, MAX_TIMESTAMP_PRECISION));
+    }
+
+    @Override
+    public Set<LogicalTypeRoot> supportedTypes() {
+        return EnumSet.of(
+                LogicalTypeRoot.CHAR,
+                LogicalTypeRoot.VARCHAR,
+                LogicalTypeRoot.BOOLEAN,
+                LogicalTypeRoot.VARBINARY,
+                LogicalTypeRoot.DECIMAL,
+                LogicalTypeRoot.TINYINT,
+                LogicalTypeRoot.SMALLINT,
+                LogicalTypeRoot.INTEGER,
+                LogicalTypeRoot.BIGINT,
+                LogicalTypeRoot.FLOAT,
+                LogicalTypeRoot.DOUBLE,
+                LogicalTypeRoot.DATE,
+                LogicalTypeRoot.TIME_WITHOUT_TIME_ZONE,
+                LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE,
+                LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE,
+                LogicalTypeRoot.ARRAY);
+    }
 }

src/test/java/pl/touk/flink/ignite/StreamTableEnvironmentUtil.java

@@ -1,10 +1,8 @@
 package pl.touk.flink.ignite;
 
 import org.apache.flink.api.common.RuntimeExecutionMode;
-import org.apache.flink.configuration.ExecutionOptions;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.api.EnvironmentSettings;
-import org.apache.flink.table.api.TableConfig;
 import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl;
 
@@ -13,11 +11,9 @@ public class StreamTableEnvironmentUtil {
     public static StreamTableEnvironment create() {
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
         env.setRuntimeMode(RuntimeExecutionMode.BATCH);
-        TableConfig tableConfig = new TableConfig();
-        tableConfig.getConfiguration().set(ExecutionOptions.RUNTIME_MODE, RuntimeExecutionMode.BATCH);
         EnvironmentSettings settings = EnvironmentSettings.newInstance().inBatchMode().build();
 
-        return StreamTableEnvironmentImpl.create(env, settings, tableConfig);
+        return StreamTableEnvironmentImpl.create(env, settings);
     }
 
 }