Flink: table supplier in writer operator for optional table refresh

core/src/main/java/org/apache/iceberg/io/OutputFileFactory.java

@@ -23,6 +23,7 @@
 
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.StructLike;
@@ -35,7 +36,7 @@ public class OutputFileFactory {
   private final PartitionSpec defaultSpec;
   private final FileFormat format;
   private final LocationProvider locations;
-  private final FileIO io;
+  private final Supplier<FileIO> ioSupplier;
   private final EncryptionManager encryptionManager;
   private final int partitionId;
   private final long taskId;
@@ -56,7 +57,7 @@ public class OutputFileFactory {
    * @param spec Partition specification used by the location provider
    * @param format File format used for the extension
    * @param locations Location provider used for generating locations
-   * @param io FileIO to store the files
+   * @param ioSupplier Supplier of FileIO to store the files
    * @param encryptionManager Encryption manager used for encrypting the files
    * @param partitionId First part of the file name
    * @param taskId Second part of the file name
@@ -67,7 +68,7 @@ private OutputFileFactory(
       PartitionSpec spec,
       FileFormat format,
       LocationProvider locations,
-      FileIO io,
+      Supplier<FileIO> ioSupplier,
       EncryptionManager encryptionManager,
       int partitionId,
       long taskId,
@@ -76,7 +77,7 @@ private OutputFileFactory(
     this.defaultSpec = spec;
     this.format = format;
     this.locations = locations;
-    this.io = io;
+    this.ioSupplier = ioSupplier;
     this.encryptionManager = encryptionManager;
     this.partitionId = partitionId;
     this.taskId = taskId;
@@ -101,7 +102,7 @@ private String generateFilename() {
 
   /** Generates an {@link EncryptedOutputFile} for unpartitioned writes. */
   public EncryptedOutputFile newOutputFile() {
-    OutputFile file = io.newOutputFile(locations.newDataLocation(generateFilename()));
+    OutputFile file = ioSupplier.get().newOutputFile(locations.newDataLocation(generateFilename()));
     return encryptionManager.encrypt(file);
   }
 
@@ -113,7 +114,7 @@ public EncryptedOutputFile newOutputFile(StructLike partition) {
   /** Generates an {@link EncryptedOutputFile} for partitioned writes in a given spec. */
   public EncryptedOutputFile newOutputFile(PartitionSpec spec, StructLike partition) {
     String newDataLocation = locations.newDataLocation(spec, partition, generateFilename());
-    OutputFile rawOutputFile = io.newOutputFile(newDataLocation);
+    OutputFile rawOutputFile = ioSupplier.get().newOutputFile(newDataLocation);
     return encryptionManager.encrypt(rawOutputFile);
   }
 
@@ -125,6 +126,7 @@ public static class Builder {
     private String operationId;
     private FileFormat format;
     private String suffix;
+    private Supplier<FileIO> ioSupplier;
 
     private Builder(Table table, int partitionId, long taskId) {
       this.table = table;
@@ -136,6 +138,7 @@ private Builder(Table table, int partitionId, long taskId) {
       String formatAsString =
           table.properties().getOrDefault(DEFAULT_FILE_FORMAT, DEFAULT_FILE_FORMAT_DEFAULT);
       this.format = FileFormat.fromString(formatAsString);
+      this.ioSupplier = table::io;
     }
 
     public Builder defaultSpec(PartitionSpec newDefaultSpec) {
@@ -158,12 +161,24 @@ public Builder suffix(String newSuffix) {
       return this;
     }
 
+    public Builder ioSupplier(Supplier<FileIO> newIoSupplier) {
+      this.ioSupplier = newIoSupplier;
+      return this;
+    }
+
     public OutputFileFactory build() {
       LocationProvider locations = table.locationProvider();
-      FileIO io = table.io();
       EncryptionManager encryption = table.encryption();
       return new OutputFileFactory(
-          defaultSpec, format, locations, io, encryption, partitionId, taskId, operationId, suffix);
+          defaultSpec,
+          format,
+          locations,
+          ioSupplier,
+          encryption,
+          partitionId,
+          taskId,
+          operationId,
+          suffix);
     }
   }
 }

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/TableLoader.java

@@ -38,6 +38,8 @@ public interface TableLoader extends Closeable, Serializable, Cloneable {
 
   void open();
 
+  boolean isOpen();
+
   Table loadTable();
 
   /** Clone a TableLoader */
@@ -75,6 +77,11 @@ public void open() {
       tables = new HadoopTables(hadoopConf.get());
     }
 
+    @Override
+    public boolean isOpen() {
+      return tables != null;
+    }
+
     @Override
     public Table loadTable() {
       FlinkEnvironmentContext.init();
@@ -115,6 +122,11 @@ public void open() {
       catalog = catalogLoader.loadCatalog();
     }
 
+    @Override
+    public boolean isOpen() {
+      return catalog != null;
+    }
+
     @Override
     public Table loadTable() {
       FlinkEnvironmentContext.init();
@@ -126,6 +138,7 @@ public void close() throws IOException {
       if (catalog instanceof Closeable) {
         ((Closeable) catalog).close();
       }
+      catalog = null;
     }
 
     @Override

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkManifestUtil.java

@@ -24,13 +24,11 @@
 import java.util.function.Supplier;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
-import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.ManifestFile;
 import org.apache.iceberg.ManifestFiles;
 import org.apache.iceberg.ManifestWriter;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Table;
-import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.OutputFile;
@@ -64,16 +62,13 @@ static List<DataFile> readDataFiles(
   }
 
   static ManifestOutputFileFactory createOutputFileFactory(
-      Table table, String flinkJobId, String operatorUniqueId, int subTaskId, long attemptNumber) {
-    TableOperations ops = ((HasTableOperations) table).operations();
+      Supplier<Table> tableSupplier,
+      String flinkJobId,
+      String operatorUniqueId,
+      int subTaskId,
+      long attemptNumber) {
     return new ManifestOutputFileFactory(
-        ops,
-        table.io(),
-        table.properties(),
-        flinkJobId,
-        operatorUniqueId,
-        subTaskId,
-        attemptNumber);
+        tableSupplier, flinkJobId, operatorUniqueId, subTaskId, attemptNumber);
   }
 
   /**

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java

@@ -32,6 +32,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.function.Function;
+import org.apache.calcite.linq4j.function.Experimental;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeinfo.Types;
@@ -67,6 +68,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.util.SerializableSupplier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -138,6 +140,7 @@ public static class Builder {
     private ReadableConfig readableConfig = new Configuration();
     private final Map<String, String> writeOptions = Maps.newHashMap();
     private FlinkWriteConf flinkWriteConf = null;
+    private long reloadIntervalMs;
 
     private Builder() {}
 
@@ -266,6 +269,20 @@ public Builder upsert(boolean enabled) {
       return this;
     }
 
+    /**
+     * Sets the interval for reloading of the Iceberg {@link Table} instance in the subtasks. When
+     * set, the table will be periodically reloaded using the {@link TableLoader} specified for the
+     * builder. Default is to not reload the table.
+     *
+     * @param intervalMs the interval for periodically reloading the table.
+     * @return {@link Builder} to connect the iceberg table.
+     */
+    @Experimental
+    public Builder reloadIntervalMs(long intervalMs) {
+      this.reloadIntervalMs = intervalMs;
+      return this;
+    }
+
     /**
      * Configuring the equality field columns for iceberg table that accept CDC or UPSERT events.
      *
@@ -429,7 +446,8 @@ private SingleOutputStreamOperator<Void> appendCommitter(
               snapshotProperties,
               flinkWriteConf.workerPoolSize(),
               flinkWriteConf.branch(),
-              table.spec());
+              table.spec(),
+              reloadIntervalMs);
       SingleOutputStreamOperator<Void> committerStream =
           writerStream
               .transform(operatorName(ICEBERG_FILES_COMMITTER_NAME), Types.VOID, filesCommitter)
@@ -463,7 +481,8 @@ private SingleOutputStreamOperator<WriteResult> appendWriter(
       }
 
       IcebergStreamWriter<RowData> streamWriter =
-          createStreamWriter(table, flinkWriteConf, flinkRowType, equalityFieldIds);
+          createStreamWriter(
+              table, tableLoader, flinkWriteConf, flinkRowType, equalityFieldIds, reloadIntervalMs);
 
       int parallelism =
           flinkWriteConf.writeParallelism() == null
@@ -581,16 +600,25 @@ static RowType toFlinkRowType(Schema schema, TableSchema requestedSchema) {
 
   static IcebergStreamWriter<RowData> createStreamWriter(
       Table table,
+      TableLoader tableLoader,
       FlinkWriteConf flinkWriteConf,
       RowType flinkRowType,
-      List<Integer> equalityFieldIds) {
+      List<Integer> equalityFieldIds,
+      long reloadIntervalMs) {
     Preconditions.checkArgument(table != null, "Iceberg table shouldn't be null");
 
     Table serializableTable = SerializableTable.copyOf(table);
+    SerializableSupplier<Table> tableSupplier;
+    if (reloadIntervalMs > 0) {
+      tableSupplier = new ReloadingTableSupplier(serializableTable, tableLoader, reloadIntervalMs);
+    } else {
+      tableSupplier = () -> serializableTable;
+    }
+
     FileFormat format = flinkWriteConf.dataFileFormat();
     TaskWriterFactory<RowData> taskWriterFactory =
         new RowDataTaskWriterFactory(
-            serializableTable,
+            tableSupplier,
             flinkRowType,
             flinkWriteConf.targetDataFileSize(),
             format,