Upgrade to Flink 2.0 (#10)

Author: grzegorz8

.github/workflows/build.yml

@@ -23,7 +23,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        flink: [ "1.17.2", "1.18.1", "1.19.0" ]
+        flink: [ "2.0.0" ]
         jdk: [ "11" ]
     steps:
       - uses: actions/checkout@v3

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## [Unreleased]
 
+-   Upgrade to Flink 2.0
+
 ## [0.2.1] - 2024-04-11
 
 ## [0.2.0] - 2023-11-22

README.md

@@ -1 +1,148 @@
 # flink-connector-jdbc-elasticsearch-dialect
+
+This module contains Flink JDBC dialect for Elasticsearch.
+
+## Elasticsearch Catalog
+
+This is an implementation of
+a [Flink Catalog](https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/catalogs/)
+for [Elastic](https://www.elastic.co/).
+
+---
+
+### Possible Operations
+
+- `listDatabases` Lists Databases in a catalog.
+- `databaseExists` Checks if a database exists.
+- `listTables` Lists Tables in a Database.
+- `tableExists` Checks if a table exists.
+- `getTable` Gets the metadata information about the table. This consists of table schema and table properties. Table
+  properties among others contain `CONNECTOR`, `BASE_URL`, `TABLE_NAME` and `SCAN_PARTITION` options.
+
+---
+
+### Scan options
+
+If we want tables in a catalog to be partitioned by a column we should specify scan options.
+It is possible to set
+up [Scan options](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/jdbc/#scan-partition-column:~:text=than%201%20second.-,scan.partition.column,-optional)
+while defining a catalog.
+
+There are 2 types of scan options for Elastic Catalog:
+
+#### Default scan options for a catalog
+
+We can specify default partitioning options for all tables in a catalog. If no options for a table are specified, these
+options will be used to
+select a column for partitioning and the number of partitions for a table will be calculated based on catalog default
+option.
+
+- `catalog.default.scan.partition.column.name` Specify what column to use for table partitioning by default. The default
+  option will be used
+  for all tables in a catalog. We can overwrite a column to use for partitioning of a table by specifying table specific
+  scan options.
+- `catalog.default.scan.partition.size` Specify how many elements should be placed in a single partition. The number of
+  partitions will be calculated based on the number of elements and the default size of a partition. If we want a
+  particular table
+  to have an exact number of partitions, we can specify that number using table specific scan options.
+
+#### Table specific scan options
+
+These options can be useful if we know that not all tables in a catalog should be partitioned in the same way. Here
+we can specify partitioning options for selected tables.
+
+- `properties.scan.{tablename}.partition.column.name` Specify the name of the column to use for partitioning of a table.
+  Corresponds to the `scan.partition.column` option.
+- `properties.scan.{tablename}.partition.number` Specify the number of partitions for a table. Corresponds to the
+  `scan.partition.num` option.
+
+For both of options specified above we should replace `{tablename}` with the name of the table that we want the options
+to apply to.
+We can provide these options for multiple tables.
+
+#### Index patterns
+
+If we specify an index pattern, a Flink table will be created in Catalog that instead of targeting a single index in
+Elastic will target all indexes that match
+the pattern provided. It is useful to use if we want to write Flink SQL that reads similar data from many similar tables
+instead of a single one.
+The resulting Flink table will contain all columns found in matching tables and will use all the data from matching
+tables.
+This table will have the same name as the pattern.
+
+- `properties.index.patterns` Specify patterns for which we want to create Flink tables. We can specify multiple index
+  patterns by
+  separating them with a comma `,` sign.
+
+The Flink tables created this way can also be partitioned just as other Flink tables by providing default catalog scan
+options or table specific scan options.
+
+#### Time attributes
+
+It is possible to add `proctime` column to each catalog table.
+
+```properties
+catalog.add-proctime-column=true
+```
+
+---
+
+### Rules for overwriting catalog scan options
+
+#### No scan options were provided
+
+There is no necessity to provide either default scan options for a catalog or table specific scan options. If there are
+no scan options provided
+no tables in a catalog will be partitioned.
+
+#### Only default scan options for a catalog were provided
+
+If only default catalog scan options were provided, all tables in a catalog will be partitioned in a similar way. The
+same column name for table partitioning for all tables and
+the number of partitions for tables will be dependant on the number of records in a table. All tables will have the same
+maximum number of elements in a partition.
+
+#### Only table specific scan options were provided
+
+If we want a specific table to be partitioned and leave the rest of tables nonpartitioned we have to provide both table
+specific scan options.
+
+#### We specified both catalog default scan options and table specific scan were options
+
+Table specific scan options have higher priority over catalog default scan properties when deciding how to partition a
+table.
+If we specify catalog default partition column name and a table specific partition column name then table specific
+partition column name is taken into account.
+Similar thing happens when we specify catalog default scan partition size and table specific partition number. Instead
+of calculating the number of partitions for a table
+based on the count of elements, the table will have the number of partitions equal to the one provided for a table.
+
+--- 
+
+### Calculation of scan partition bounds
+
+If a table is partitioned, meaning that we specified catalog default scan options or we specified table specific scan
+options the upper and lower bounds will be calculated.
+As specified in the Flink documentation, the `properties.scan.{tablename}.partition.column.name` option works for
+numeric and temporal data types.
+The `scan.partition.lower-bound` will be calculated as the lowest value in the table.
+The `scan.partition.upper-bound` will be calculated as the highest value in the table.
+
+---
+
+### Note that
+
+If we want a table to be partitioned it is necessary that we provide a catalog default or table specific option for
+partition column to use and
+catalog default or table specific partition number option for deciding how many partitions to use for a table.
+If only 1 option is provided we will receive an error.
+
+---
+
+### Implementation details
+
+`com.getindata.flink.connector.jdbc.elasticsearch.database.catalog.ElasticsearchJdbcCatalogFactory` - has been copied
+because default CatalogFactory does not allow to pass custom catalog properties.
+
+`com.getindata.flink.connector.jdbc.elasticsearch.database.catalog.CopiedAbstractJdbcCatalog` is a copy of
+`org.apache.flink.connector.jdbc.core.database.catalog.AbstractJdbcCatalog` where JDBC validation is modified.

pom.xml

@@ -41,9 +41,9 @@
     <scala.binary.version>2.12</scala.binary.version>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 
-    <elasticsearch.version>8.11.1</elasticsearch.version>
-    <flink.version>1.18.1</flink.version>
-    <flink-connector-jdbc.version>3.1.2-1.18</flink-connector-jdbc.version>
+    <elasticsearch.version>8.19.0</elasticsearch.version>
+    <flink.version>2.0.0</flink.version>
+    <flink-connector-jdbc.version>4.0.0-2.0</flink-connector-jdbc.version>
     <jackson.version>2.15.2</jackson.version>
 
     <!-- Test -->
@@ -60,6 +60,7 @@
     <junit.version>5.9.1</junit.version>
     <logback.version>1.3.5</logback.version>
     <mockito.version>2.21.0</mockito.version>
+    <okhttp.version>4.10.0</okhttp.version>
     <slf4j-api.version>2.0.4</slf4j-api.version>
     <testcontainers.version>1.18.2</testcontainers.version>
   </properties>
@@ -192,7 +193,13 @@
     </dependency>
     <dependency>
       <groupId>org.apache.flink</groupId>
-      <artifactId>flink-connector-jdbc</artifactId>
+      <artifactId>flink-table-api-java</artifactId>
+      <version>${flink.version}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-connector-jdbc-core</artifactId>
       <version>${flink-connector-jdbc.version}</version>
     </dependency>
 
@@ -204,7 +211,21 @@
       <scope>provided</scope>
     </dependency>
 
+    <!-- Logging dependencies -->
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-api</artifactId>
+      <version>${slf4j-api.version}</version>
+    </dependency>
+
     <!-- Test dependencies -->
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-connector-jdbc-core</artifactId>
+      <version>${flink-connector-jdbc.version}</version>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.junit</groupId>
       <artifactId>junit-bom</artifactId>
@@ -251,12 +272,6 @@
       <version>${logback.version}</version>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>org.slf4j</groupId>
-      <artifactId>slf4j-api</artifactId>
-      <version>${slf4j-api.version}</version>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>ch.qos.logback</groupId>
       <artifactId>logback-classic</artifactId>
@@ -343,6 +358,12 @@
       <version>${jackson.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>com.squareup.okhttp3</groupId>
+      <artifactId>okhttp</artifactId>
+      <version>${okhttp.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <profiles>

src/main/java/com/getindata/flink/connector/jdbc/databases/elasticsearch/dialect/ElasticsearchDialectFactory.java renamed to src/main/java/com/getindata/flink/connector/jdbc/elasticsearch/database/ElasticsearchFactory.java

@@ -16,20 +16,34 @@
  * limitations under the License.
  */
 
-package com.getindata.flink.connector.jdbc.databases.elasticsearch.dialect;
+package com.getindata.flink.connector.jdbc.elasticsearch.database;
 
-import org.apache.flink.connector.jdbc.dialect.JdbcDialect;
-import org.apache.flink.connector.jdbc.dialect.JdbcDialectFactory;
+import com.getindata.flink.connector.jdbc.elasticsearch.database.dialect.ElasticsearchDialect;
+import org.apache.flink.connector.jdbc.core.database.JdbcFactory;
+import org.apache.flink.connector.jdbc.core.database.catalog.JdbcCatalog;
+import org.apache.flink.connector.jdbc.core.database.dialect.JdbcDialect;
 
-public class ElasticsearchDialectFactory implements JdbcDialectFactory {
+public class ElasticsearchFactory implements JdbcFactory {
 
     @Override
     public boolean acceptsURL(String url) {
         return url.startsWith("jdbc:elasticsearch:") || url.startsWith("jdbc:es:");
     }
 
     @Override
-    public JdbcDialect create() {
+    public JdbcDialect createDialect() {
         return new ElasticsearchDialect();
     }
+
+    @Override
+    public JdbcCatalog createCatalog(
+            ClassLoader classLoader,
+            String catalogName,
+            String defaultDatabase,
+            String username,
+            String pwd,
+            String baseUrl) {
+        throw new UnsupportedOperationException("Use elasticsearch catalog factory.");
+    }
+
 }