Flink: sync 1.16 with 1.17 for backports missed or not ported identic…

…ally (apache#7403)
haojue · Apr 24, 2023 · 38cf2a1 · 38cf2a1
1 parent 69c3249
commit 38cf2a1
Show file tree

Hide file tree

Showing 7 changed files with 13 additions and 55 deletions.
diff --git a/.../v1.16/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java b/.../v1.16/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java
@@ -18,7 +18,6 @@
  */
 package org.apache.iceberg.flink.source;
 
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import org.apache.flink.annotation.Internal;
@@ -59,14 +58,8 @@ public class RowDataFileScanTaskReader implements FileScanTaskReader<RowData> {
   private final Schema projectedSchema;
   private final String nameMapping;
   private final boolean caseSensitive;
-
   private final FlinkSourceFilter rowFilter;
 
-  public RowDataFileScanTaskReader(
-      Schema tableSchema, Schema projectedSchema, String nameMapping, boolean caseSensitive) {
-    this(tableSchema, projectedSchema, nameMapping, caseSensitive, Collections.emptyList());
-  }
-
   public RowDataFileScanTaskReader(
       Schema tableSchema,
       Schema projectedSchema,
@@ -148,7 +141,6 @@ private CloseableIterable<RowData> newIterable(
     if (rowFilter != null) {
       return CloseableIterable.filter(iter, rowFilter::filter);
     }
-
     return iter;
   }
 

diff --git a/flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/source/RowDataRewriter.java b/flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/source/RowDataRewriter.java
@@ -21,6 +21,7 @@
 import static org.apache.iceberg.TableProperties.DEFAULT_NAME_MAPPING;
 
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
 import org.apache.flink.api.common.functions.RichMapFunction;
@@ -125,7 +126,8 @@ public RewriteMap(
       this.encryptionManager = encryptionManager;
       this.taskWriterFactory = taskWriterFactory;
       this.rowDataReader =
-          new RowDataFileScanTaskReader(schema, schema, nameMapping, caseSensitive);
+          new RowDataFileScanTaskReader(
+              schema, schema, nameMapping, caseSensitive, Collections.emptyList());
     }
 
     @Override

diff --git a/...src/main/java/org/apache/iceberg/flink/source/reader/AvroGenericRecordReaderFunction.java b/...src/main/java/org/apache/iceberg/flink/source/reader/AvroGenericRecordReaderFunction.java
@@ -18,7 +18,6 @@
  */
 package org.apache.iceberg.flink.source.reader;
 
-import java.util.Collections;
 import java.util.List;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.flink.configuration.Configuration;
@@ -58,28 +57,7 @@ public static AvroGenericRecordReaderFunction fromTable(Table table) {
         false,
         table.io(),
         table.encryption(),
-        Collections.emptyList());
-  }
-
-  public AvroGenericRecordReaderFunction(
-      String tableName,
-      ReadableConfig config,
-      Schema tableSchema,
-      Schema projectedSchema,
-      String nameMapping,
-      boolean caseSensitive,
-      FileIO io,
-      EncryptionManager encryption) {
-    this(
-        tableName,
-        config,
-        tableSchema,
-        projectedSchema,
-        nameMapping,
-        caseSensitive,
-        io,
-        encryption,
-        Collections.emptyList());
+        null);
   }
 
   public AvroGenericRecordReaderFunction(

diff --git a/....16/flink/src/main/java/org/apache/iceberg/flink/source/reader/RowDataReaderFunction.java b/....16/flink/src/main/java/org/apache/iceberg/flink/source/reader/RowDataReaderFunction.java
@@ -18,7 +18,6 @@
  */
 package org.apache.iceberg.flink.source.reader;
 
-import java.util.Collections;
 import java.util.List;
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.table.data.RowData;
@@ -41,25 +40,6 @@ public class RowDataReaderFunction extends DataIteratorReaderFunction<RowData> {
   private final EncryptionManager encryption;
   private final List<Expression> filters;
 
-  public RowDataReaderFunction(
-      ReadableConfig config,
-      Schema tableSchema,
-      Schema projectedSchema,
-      String nameMapping,
-      boolean caseSensitive,
-      FileIO io,
-      EncryptionManager encryption) {
-    this(
-        config,
-        tableSchema,
-        projectedSchema,
-        nameMapping,
-        caseSensitive,
-        io,
-        encryption,
-        Collections.emptyList());
-  }
-
   public RowDataReaderFunction(
       ReadableConfig config,
       Schema tableSchema,

diff --git a/flink/v1.16/flink/src/test/java/org/apache/iceberg/flink/source/reader/ReaderUtil.java b/flink/v1.16/flink/src/test/java/org/apache/iceberg/flink/source/reader/ReaderUtil.java
@@ -20,6 +20,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.util.Collections;
 import java.util.List;
 import org.apache.flink.table.data.RowData;
 import org.apache.iceberg.BaseCombinedScanTask;
@@ -83,7 +84,8 @@ public static FileScanTask createFileTask(
 
   public static DataIterator<RowData> createDataIterator(CombinedScanTask combinedTask) {
     return new DataIterator<>(
-        new RowDataFileScanTaskReader(TestFixtures.SCHEMA, TestFixtures.SCHEMA, null, true),
+        new RowDataFileScanTaskReader(
+            TestFixtures.SCHEMA, TestFixtures.SCHEMA, null, true, Collections.emptyList()),
         combinedTask,
         new HadoopFileIO(new org.apache.hadoop.conf.Configuration()),
         new PlaintextEncryptionManager());

diff --git a/...6/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestIcebergSourceReader.java b/...6/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestIcebergSourceReader.java
@@ -19,6 +19,7 @@
 package org.apache.iceberg.flink.source.reader;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import org.apache.flink.api.connector.source.SourceReaderContext;
 import org.apache.flink.configuration.Configuration;
@@ -106,7 +107,8 @@ private IcebergSourceReader createReader(
             null,
             true,
             new HadoopFileIO(new org.apache.hadoop.conf.Configuration()),
-            new PlaintextEncryptionManager());
+            new PlaintextEncryptionManager(),
+            Collections.emptyList());
     return new IcebergSourceReader<>(readerMetrics, readerFunction, readerContext);
   }
 }
diff --git a/...flink/src/test/java/org/apache/iceberg/flink/source/reader/TestRowDataReaderFunction.java b/...flink/src/test/java/org/apache/iceberg/flink/source/reader/TestRowDataReaderFunction.java
@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg.flink.source.reader;
 
+import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
 import org.apache.flink.configuration.Configuration;
@@ -55,7 +56,8 @@ protected ReaderFunction<RowData> readerFunction() {
         null,
         true,
         new HadoopFileIO(new org.apache.hadoop.conf.Configuration()),
-        new PlaintextEncryptionManager());
+        new PlaintextEncryptionManager(),
+        Collections.emptyList());
   }
 
   @Override