[HUDI-5205] Support flink 1.16.0 (apache#7584)

Co-authored-by: wuzhiping <wuzhiping.007@bytedance.com>
fengjian428 · Apr 5, 2023 · bc6ca1c · bc6ca1c
1 parent ba66bdf
commit bc6ca1c
Show file tree

Hide file tree

Showing 49 changed files with 4,416 additions and 20 deletions.
diff --git a/.github/workflows/bot.yml b/.github/workflows/bot.yml
@@ -69,6 +69,7 @@ jobs:
           - flinkProfile: "flink1.13"
           - flinkProfile: "flink1.14"
           - flinkProfile: "flink1.15"
+          - flinkProfile: "flink1.16"
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK 8
@@ -95,6 +96,8 @@ jobs:
     strategy:
       matrix:
         include:
+          - flinkProfile: 'flink1.16'
+            sparkProfile: 'spark3.3'
           - flinkProfile: 'flink1.15'
             sparkProfile: 'spark3.3'
           - flinkProfile: 'flink1.14'

diff --git a/...asource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java b/...asource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java
@@ -18,6 +18,7 @@
 
 package org.apache.hudi.sink;
 
+import org.apache.hudi.adapter.OperatorCoordinatorAdapter;
 import org.apache.hudi.client.HoodieFlinkWriteClient;
 import org.apache.hudi.client.WriteStatus;
 import org.apache.hudi.common.config.SerializableConfiguration;
@@ -48,10 +49,11 @@
 import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
 import org.apache.flink.runtime.operators.coordination.OperatorEvent;
 import org.apache.flink.runtime.operators.coordination.TaskNotRunningException;
-import org.jetbrains.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.Arrays;
@@ -79,7 +81,7 @@
  * @see StreamWriteFunction for the work flow and semantics
  */
 public class StreamWriteOperatorCoordinator
-    implements OperatorCoordinator {
+    implements OperatorCoordinatorAdapter {
   private static final Logger LOG = LoggerFactory.getLogger(StreamWriteOperatorCoordinator.class);
 
   /**

diff --git a/...k-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/SortOperatorGen.java b/...k-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/sort/SortOperatorGen.java
@@ -18,6 +18,8 @@
 
 package org.apache.hudi.sink.bulk.sort;
 
+import org.apache.hudi.adapter.SortCodeGeneratorAdapter;
+
 import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
 import org.apache.flink.table.api.TableConfig;
 import org.apache.flink.table.data.RowData;
@@ -52,6 +54,6 @@ public SortCodeGenerator createSortCodeGenerator() {
     for (int sortIndex : sortIndices) {
       builder.addField(sortIndex, true, true);
     }
-    return new SortCodeGenerator(tableConfig, rowType, builder.build());
+    return new SortCodeGeneratorAdapter(tableConfig, rowType, builder.build());
   }
 }
diff --git a/...-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieHiveCatalog.java b/...-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieHiveCatalog.java
@@ -746,22 +746,22 @@ public void alterTable(
   @Override
   public List<CatalogPartitionSpec> listPartitions(ObjectPath tablePath)
       throws TableNotExistException, TableNotPartitionedException, CatalogException {
-    throw new HoodieCatalogException("Not supported.");
+    return Collections.emptyList();
   }
 
   @Override
   public List<CatalogPartitionSpec> listPartitions(
       ObjectPath tablePath, CatalogPartitionSpec partitionSpec)
       throws TableNotExistException, TableNotPartitionedException,
       PartitionSpecInvalidException, CatalogException {
-    throw new HoodieCatalogException("Not supported.");
+    return Collections.emptyList();
   }
 
   @Override
   public List<CatalogPartitionSpec> listPartitionsByFilter(
       ObjectPath tablePath, List<Expression> expressions)
       throws TableNotExistException, TableNotPartitionedException, CatalogException {
-    throw new HoodieCatalogException("Not supported.");
+    return Collections.emptyList();
   }
 
   @Override

diff --git a/...ink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java b/...ink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
@@ -1535,18 +1535,18 @@ void testBuiltinFunctionWithHMSCatalog() {
         .end();
     tableEnv.executeSql(hoodieTableDDL);
 
-    String insertSql = "insert into t1 values (1, TO_DATE('2022-02-02'), '1'), (2, DATE '2022-02-02', '1')";
+    String insertSql = "insert into t1 values (1, TO_DATE('2022-02-02'), '1'), (2, DATE '2022-02-02', '2')";
     execInsertSql(tableEnv, insertSql);
 
     List<Row> result = CollectionUtil.iterableToList(
         () -> tableEnv.sqlQuery("select * from t1").execute().collect());
     final String expected = "["
         + "+I[1, 2022-02-02, 1], "
-        + "+I[2, 2022-02-02, 1]]";
+        + "+I[2, 2022-02-02, 2]]";
     assertRowsEquals(result, expected);
 
     List<Row> partitionResult = CollectionUtil.iterableToList(
-        () -> tableEnv.sqlQuery("select * from t1 where f_int = 1").execute().collect());
+        () -> tableEnv.sqlQuery("select * from t1 where f_par = '1'").execute().collect());
     assertRowsEquals(partitionResult, "[+I[1, 2022-02-02, 1]]");
   }
 

diff --git a/...atasource/hudi-flink/src/test/java/org/apache/hudi/utils/source/ContinuousFileSource.java b/...atasource/hudi-flink/src/test/java/org/apache/hudi/utils/source/ContinuousFileSource.java
@@ -20,6 +20,8 @@
 
 import org.apache.hudi.adapter.DataStreamScanProviderAdapter;
 
+import org.apache.flink.api.common.functions.AbstractRichFunction;
+import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.common.state.CheckpointListener;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.fs.Path;
@@ -49,7 +51,7 @@
  * A continuous file source that can trigger checkpoints continuously.
  *
  * <p>It loads the data in the specified file and split the data into number of checkpoints batches.
- * Say, if you want 4 checkpoints and there are 8 records in the file, the emit strategy is:
+ * Say, if you want 4 checkpoints and there are 8 records in the file, the emission strategy is:
  *
  * <pre>
  *   | 2 records | 2 records | 2 records | 2 records |
@@ -85,6 +87,7 @@ public boolean isBounded() {
       @Override
       public DataStream<RowData> produceDataStream(StreamExecutionEnvironment execEnv) {
         final RowType rowType = (RowType) tableSchema.toSourceRowDataType().getLogicalType();
+
         JsonRowDataDeserializationSchema deserializationSchema = new JsonRowDataDeserializationSchema(
             rowType,
             InternalTypeInfo.of(rowType),
@@ -95,7 +98,7 @@ public DataStream<RowData> produceDataStream(StreamExecutionEnvironment execEnv)
         return execEnv.addSource(new BoundedSourceFunction(path, conf.getInteger(CHECKPOINTS)))
             .name("continuous_file_source")
             .setParallelism(1)
-            .map(record -> deserializationSchema.deserialize(record.getBytes(StandardCharsets.UTF_8)),
+            .map(new JsonDeserializationFunction(deserializationSchema),
                 InternalTypeInfo.of(rowType));
       }
     };
@@ -183,4 +186,28 @@ public void notifyCheckpointComplete(long l) {
       this.currentCP.incrementAndGet();
     }
   }
+
+  /**
+   * Wrapper function that manages the lifecycle of the JSON deserialization schema.
+   */
+  public static class JsonDeserializationFunction
+      extends AbstractRichFunction
+      implements MapFunction<String, RowData> {
+    private final JsonRowDataDeserializationSchema deserializationSchema;
+
+    public JsonDeserializationFunction(JsonRowDataDeserializationSchema deserializationSchema) {
+      this.deserializationSchema = deserializationSchema;
+    }
+
+    @Override
+    public void open(Configuration parameters) throws Exception {
+      super.open(parameters);
+      this.deserializationSchema.open(null);
+    }
+
+    @Override
+    public RowData map(String record) throws Exception {
+      return deserializationSchema.deserialize(record.getBytes(StandardCharsets.UTF_8));
+    }
+  }
 }
diff --git a/hudi-flink-datasource/hudi-flink1.13.x/pom.xml b/hudi-flink-datasource/hudi-flink1.13.x/pom.xml
@@ -59,6 +59,8 @@
             <version>${hadoop.version}</version>
             <scope>provided</scope>
         </dependency>
+
+        <!-- Flink -->
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-table-runtime-blink_${scala.binary.version}</artifactId>
@@ -89,15 +91,21 @@
             <version>${flink1.13.version}</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
+            <version>${flink1.13.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- Test dependencies -->
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-runtime_${scala.binary.version}</artifactId>
             <version>${flink1.13.version}</version>
             <scope>test</scope>
             <type>test-jar</type>
         </dependency>
-
-        <!-- Test dependencies -->
         <dependency>
             <groupId>org.apache.hudi</groupId>
             <artifactId>hudi-tests-common</artifactId>

diff --git a/...ce/hudi-flink1.13.x/src/main/java/org/apache/hudi/adapter/OperatorCoordinatorAdapter.java b/...ce/hudi-flink1.13.x/src/main/java/org/apache/hudi/adapter/OperatorCoordinatorAdapter.java
@@ -0,0 +1,27 @@
+/*
+ * 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.hudi.adapter;
+
+import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
+
+/**
+ * Adapter clazz for {@code OperatorCoordinator}.
+ */
+public interface OperatorCoordinatorAdapter extends OperatorCoordinator {
+}
diff --git a/...urce/hudi-flink1.13.x/src/main/java/org/apache/hudi/adapter/SortCodeGeneratorAdapter.java b/...urce/hudi-flink1.13.x/src/main/java/org/apache/hudi/adapter/SortCodeGeneratorAdapter.java
@@ -0,0 +1,33 @@
+/*
+ * 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.hudi.adapter;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.codegen.sort.SortCodeGenerator;
+import org.apache.flink.table.planner.plan.nodes.exec.spec.SortSpec;
+import org.apache.flink.table.types.logical.RowType;
+
+/**
+ * Adapter clazz for {@code SortCodeGenerator}.
+ */
+public class SortCodeGeneratorAdapter extends SortCodeGenerator {
+  public SortCodeGeneratorAdapter(TableConfig conf, RowType input, SortSpec sortSpec) {
+    super(conf, input, sortSpec);
+  }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.14.x/pom.xml b/hudi-flink-datasource/hudi-flink1.14.x/pom.xml
@@ -59,6 +59,8 @@
             <version>${hadoop.version}</version>
             <scope>provided</scope>
         </dependency>
+
+        <!-- Flink -->
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-table-api-java</artifactId>
@@ -107,15 +109,21 @@
             <version>${flink1.14.version}</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+            <version>${flink1.14.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- Test dependencies -->
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-runtime</artifactId>
             <version>${flink1.14.version}</version>
             <scope>test</scope>
             <type>test-jar</type>
         </dependency>
-
-        <!-- Test dependencies -->
         <dependency>
             <groupId>org.apache.hudi</groupId>
             <artifactId>hudi-tests-common</artifactId>

diff --git a/...ce/hudi-flink1.14.x/src/main/java/org/apache/hudi/adapter/OperatorCoordinatorAdapter.java b/...ce/hudi-flink1.14.x/src/main/java/org/apache/hudi/adapter/OperatorCoordinatorAdapter.java
@@ -0,0 +1,27 @@
+/*
+ * 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.hudi.adapter;
+
+import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
+
+/**
+ * Adapter clazz for {@code OperatorCoordinator}.
+ */
+public interface OperatorCoordinatorAdapter extends OperatorCoordinator {
+}
diff --git a/...urce/hudi-flink1.14.x/src/main/java/org/apache/hudi/adapter/SortCodeGeneratorAdapter.java b/...urce/hudi-flink1.14.x/src/main/java/org/apache/hudi/adapter/SortCodeGeneratorAdapter.java
@@ -0,0 +1,33 @@
+/*
+ * 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.hudi.adapter;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.codegen.sort.SortCodeGenerator;
+import org.apache.flink.table.planner.plan.nodes.exec.spec.SortSpec;
+import org.apache.flink.table.types.logical.RowType;
+
+/**
+ * Adapter clazz for {@code SortCodeGenerator}.
+ */
+public class SortCodeGeneratorAdapter extends SortCodeGenerator {
+  public SortCodeGeneratorAdapter(TableConfig tableConfig, RowType input, SortSpec sortSpec) {
+    super(tableConfig, input, sortSpec);
+  }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.15.x/pom.xml b/hudi-flink-datasource/hudi-flink1.15.x/pom.xml
@@ -59,6 +59,8 @@
             <version>${hadoop.version}</version>
             <scope>provided</scope>
         </dependency>
+
+        <!-- Flink -->
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-table-api-java</artifactId>
@@ -107,15 +109,21 @@
             <version>${flink1.15.version}</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-planner_2.12</artifactId>
+            <version>${flink1.15.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- Test dependencies -->
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-runtime</artifactId>
             <version>${flink1.15.version}</version>
             <scope>test</scope>
             <type>test-jar</type>
         </dependency>
-
-        <!-- Test dependencies -->
         <dependency>
             <groupId>org.apache.hudi</groupId>
             <artifactId>hudi-tests-common</artifactId>