Address wrap-up items for Iceberg side input table cache

sdks/java/io/iceberg/build.gradle

@@ -50,6 +50,7 @@ dependencies {
     implementation library.java.avro
     implementation library.java.slf4j_api
     implementation library.java.joda_time
+    implementation library.java.jackson_databind
     implementation "org.apache.parquet:parquet-column:$parquet_version"
     implementation "org.apache.parquet:parquet-hadoop:$parquet_version"
     implementation "org.apache.parquet:parquet-common:$parquet_version"

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergCatalogConfig.java

@@ -80,6 +80,21 @@ public Catalog catalog() {
     return CATALOG_CACHE.computeIfAbsent(this, IcebergCatalogConfig::buildCatalog);
   }
 
+  /**
+   * Constructs and returns a new {@link Configuration} populated with properties from {@link
+   * #getConfigProperties()}.
+   */
+  public Configuration getHadoopConfiguration() {
+    Configuration config = new Configuration();
+    Map<String, String> confProps = getConfigProperties();
+    if (confProps != null) {
+      for (Map.Entry<String, String> prop : confProps.entrySet()) {
+        config.set(prop.getKey(), prop.getValue());
+      }
+    }
+    return config;
+  }
+
   private static Catalog buildCatalog(IcebergCatalogConfig catalogConfig) {
     String catalogName = catalogConfig.getCatalogName();
     if (catalogName == null) {
@@ -89,14 +104,7 @@ private static Catalog buildCatalog(IcebergCatalogConfig catalogConfig) {
     if (catalogProps == null) {
       catalogProps = Maps.newHashMap();
     }
-    Map<String, String> confProps = catalogConfig.getConfigProperties();
-    if (confProps == null) {
-      confProps = Maps.newHashMap();
-    }
-    Configuration config = new Configuration();
-    for (Map.Entry<String, String> prop : confProps.entrySet()) {
-      config.set(prop.getKey(), prop.getValue());
-    }
+    Configuration config = catalogConfig.getHadoopConfiguration();
     return CatalogUtil.buildIcebergCatalog(catalogName, catalogProps, config);
   }
 

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriterManager.java

@@ -27,7 +27,6 @@
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -318,9 +317,7 @@ Table getOrCreateTable(
     if (sideInputTableSpecs != null && sideInputTableSpecs.containsKey(tableIdString)) {
       SerializableTableSpec spec = sideInputTableSpecs.get(tableIdString);
       if (spec != null) {
-        Map<String, String> catalogProperties = catalogConfig.getCatalogProperties();
-        return new SideInputTable(
-            spec, catalogProperties != null ? catalogProperties : Collections.emptyMap());
+        return new SideInputTable(spec, catalogConfig);
       }
     }
     return TableCache.getAndRefreshIfStale(

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpec.java

@@ -19,7 +19,9 @@
 
 import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import com.google.auto.value.AutoValue;
+import java.io.IOException;
 import java.io.Serializable;
 import java.util.Collections;
 import java.util.List;
@@ -33,6 +35,8 @@
 import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber;
 import org.apache.beam.sdk.schemas.annotations.SchemaIgnore;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.iceberg.EncryptedKeyParser;
 import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.PartitionSpec;
@@ -47,6 +51,7 @@
 import org.apache.iceberg.encryption.EncryptedKey;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.FileIOParser;
+import org.apache.iceberg.util.JsonUtil;
 import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
@@ -185,7 +190,14 @@ public Map<Integer, SortOrder> getSortOrders() {
         if (local == null) {
           ImmutableMap.Builder<Integer, SortOrder> builder = ImmutableMap.builder();
           for (Map.Entry<Integer, String> entry : getSortOrdersJson().entrySet()) {
-            builder.put(entry.getKey(), SortOrderParser.fromJson(getSchema(), entry.getValue()));
+            try {
+              JsonNode node = JsonUtil.mapper().readTree(entry.getValue());
+              builder.put(
+                  entry.getKey(), SortOrderParser.fromJson(getSchema(), node, getOrderId()));
+            } catch (IOException e) {
+              throw new IllegalArgumentException(
+                  "Failed to parse sort order JSON for orderId " + entry.getKey(), e);
+            }
           }
           cachedSortOrders = local = builder.build();
         }
@@ -224,17 +236,34 @@ public TableIdentifier getTableIdentifier() {
     return local;
   }
 
+  /** Returns a cached {@link FileIO} instance for this table using default configuration. */
   @SchemaIgnore
   public FileIO getFileIO() {
+    return getFileIO(null);
+  }
+
+  /**
+   * Returns a cached {@link FileIO} instance for this table, configured with the provided Hadoop
+   * {@link Configuration} if supported.
+   */
+  @SchemaIgnore
+  public FileIO getFileIO(@Nullable Configuration conf) {
     FileIO local = cachedFileIO;
     if (local == null) {
       synchronized (this) {
         local = cachedFileIO;
         if (local == null) {
-          cachedFileIO = local = FileIOParser.fromJson(getFileIoJson());
+          cachedFileIO =
+              local =
+                  conf != null
+                      ? FileIOParser.fromJson(getFileIoJson(), conf)
+                      : FileIOParser.fromJson(getFileIoJson());
         }
       }
     }
+    if (conf != null && local instanceof Configurable) {
+      ((Configurable) local).setConf(conf);
+    }
     return local;
   }
 

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SideInputTable.java

@@ -25,6 +25,7 @@
 import java.util.Objects;
 import org.apache.beam.sdk.annotations.Internal;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.iceberg.AppendFiles;
 import org.apache.iceberg.DeleteFiles;
 import org.apache.iceberg.ExpireSnapshots;
@@ -61,6 +62,7 @@
 import org.apache.iceberg.encryption.PlaintextEncryptionManager;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.LocationProvider;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
  * A lightweight adapter that implements {@link Table} backed by a {@link SerializableTableSpec}.
@@ -80,14 +82,23 @@ public class SideInputTable implements Table {
   private final SerializableTableSpec spec;
   private final EncryptionManager encryptionManager;
   private final LocationProvider locationProvider;
+  private final @Nullable Configuration hadoopConf;
 
   public SideInputTable(SerializableTableSpec spec) {
-    this(spec, Collections.emptyMap());
+    this(spec, Collections.emptyMap(), null);
   }
 
   public SideInputTable(SerializableTableSpec spec, Map<String, String> catalogProperties) {
+    this(spec, catalogProperties, null);
+  }
+
+  public SideInputTable(
+      SerializableTableSpec spec,
+      Map<String, String> catalogProperties,
+      @Nullable Configuration hadoopConf) {
     this.spec = checkNotNull(spec, "spec must not be null");
     checkNotNull(catalogProperties, "catalogProperties must not be null");
+    this.hadoopConf = hadoopConf;
     this.locationProvider =
         LocationProviders.locationsFor(spec.getLocation(), spec.getProperties());
 
@@ -102,10 +113,27 @@ public SideInputTable(SerializableTableSpec spec, Map<String, String> catalogPro
   }
 
   public SideInputTable(SerializableTableSpec spec, EncryptionManager encryptionManager) {
+    this(spec, encryptionManager, null);
+  }
+
+  public SideInputTable(
+      SerializableTableSpec spec,
+      EncryptionManager encryptionManager,
+      @Nullable Configuration hadoopConf) {
     this.spec = checkNotNull(spec, "spec must not be null");
     this.encryptionManager = checkNotNull(encryptionManager, "encryptionManager must not be null");
     this.locationProvider =
         LocationProviders.locationsFor(spec.getLocation(), spec.getProperties());
+    this.hadoopConf = hadoopConf;
+  }
+
+  public SideInputTable(SerializableTableSpec spec, IcebergCatalogConfig catalogConfig) {
+    this(
+        spec,
+        checkNotNull(catalogConfig, "catalogConfig must not be null").getCatalogProperties() != null
+            ? catalogConfig.getCatalogProperties()
+            : Collections.emptyMap(),
+        catalogConfig.getHadoopConfiguration());
   }
 
   public SerializableTableSpec getTableSpec() {
@@ -164,7 +192,7 @@ public LocationProvider locationProvider() {
 
   @Override
   public FileIO io() {
-    return spec.getFileIO();
+    return spec.getFileIO(hadoopConf);
   }
 
   @Override
@@ -363,6 +391,7 @@ public String toString() {
     return MoreObjects.toStringHelper(this)
         .add("spec", spec)
         .add("encryptionManager", encryptionManager)
+        .add("hadoopConf", hadoopConf)
         .toString();
   }
 }

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/TableMetadataDriver.java

@@ -24,6 +24,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import org.apache.beam.sdk.annotations.Internal;
@@ -252,19 +253,24 @@ public PCollectionView<Map<String, SerializableTableSpec>> expand(PCollection<Ro
 
   @Override
   public PCollection<KV<String, @Nullable SerializableTableSpec>> expand(PCollection<Row> input) {
+    boolean isStreaming = input.isBounded() == PCollection.IsBounded.UNBOUNDED;
+
+    Duration customInterval = getRefreshInterval();
+    Duration interval =
+        checkNotNull(customInterval != null ? customInterval : DEFAULT_REFRESH_INTERVAL);
+
     PCollection<String> tableIds =
         input
-            .apply("ExtractTableIds", ParDo.of(new ExtractTableIdsDoFn(getDynamicDestinations())))
+            .apply(
+                "ExtractTableIds",
+                ParDo.of(
+                    new ExtractTableIdsDoFn(
+                        getDynamicDestinations(), isStreaming ? interval : null)))
             .setCoder(StringUtf8Coder.of())
             .apply("MetadataGlobalWindow", Window.into(new GlobalWindows()));
 
-    boolean isStreaming = input.isBounded() == PCollection.IsBounded.UNBOUNDED;
-
     PCollection<String> distinctTableIds;
     if (isStreaming) {
-      Duration customInterval = getRefreshInterval();
-      Duration interval =
-          checkNotNull(customInterval != null ? customInterval : DEFAULT_REFRESH_INTERVAL);
       distinctTableIds =
           tableIds.apply(
               "DeduplicateTableIds", Deduplicate.<String>values().withDuration(interval));
@@ -323,10 +329,61 @@ public void populateDisplayData(DisplayData.Builder builder) {
   }
 
   static class ExtractTableIdsDoFn extends DoFn<Row, String> {
+    private static final int DEFAULT_LOCAL_CACHE_MAX_SIZE = 10_000;
+
+    private static volatile @Nullable Clock globalTestClock;
+
     private final DynamicDestinations dynamicDestinations;
+    private final @Nullable Duration refreshInterval;
+    private transient @Nullable Clock clock;
+    private transient @Nullable LinkedHashMap<String, Long> lastEmittedCache;
 
     ExtractTableIdsDoFn(DynamicDestinations dynamicDestinations) {
+      this(dynamicDestinations, null);
+    }
+
+    ExtractTableIdsDoFn(
+        DynamicDestinations dynamicDestinations, @Nullable Duration refreshInterval) {
       this.dynamicDestinations = dynamicDestinations;
+      this.refreshInterval = refreshInterval;
+    }
+
+    @VisibleForTesting
+    void setClock(@Nullable Clock clock) {
+      this.clock = clock;
+    }
+
+    @VisibleForTesting
+    static void setGlobalTestClock(@Nullable Clock clock) {
+      globalTestClock = clock;
+    }
+
+    @Setup
+    public void setup() {
+      initCache();
+    }
+
+    private void initCache() {
+      if (lastEmittedCache == null && refreshInterval != null) {
+        this.lastEmittedCache =
+            new LinkedHashMap<String, Long>(16, 0.75f, true) {
+              @Override
+              protected boolean removeEldestEntry(Map.Entry<String, Long> eldest) {
+                return size() > DEFAULT_LOCAL_CACHE_MAX_SIZE;
+              }
+            };
+      }
+    }
+
+    private long getNow() {
+      if (clock != null) {
+        return clock.currentTimeMillis();
+      }
+      Clock global = globalTestClock;
+      if (global != null) {
+        return global.currentTimeMillis();
+      }
+      return System.currentTimeMillis();
     }
 
     @ProcessElement
@@ -340,7 +397,19 @@ public void processElement(
           dynamicDestinations.getTableStringIdentifier(
               ValueInSingleWindow.of(element, timestamp, window, paneInfo));
       if (tableIdentifier != null && !tableIdentifier.trim().isEmpty()) {
-        out.output(tableIdentifier.trim());
+        Map<String, Long> cache = lastEmittedCache;
+        Duration interval = refreshInterval;
+        if (cache != null && interval != null) {
+          long now = getNow();
+          Long lastEmitted = cache.get(tableIdentifier);
+          long minInterval = Math.max(1L, interval.getMillis() / 2);
+          if (lastEmitted == null || (now - lastEmitted) >= minInterval) {
+            cache.put(tableIdentifier, now);
+            out.output(tableIdentifier);
+          }
+        } else {
+          out.output(tableIdentifier);
+        }
       }
     }
   }

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/WritePartitionedRowsToFiles.java

@@ -22,7 +22,6 @@
 import static org.apache.beam.sdk.io.iceberg.RecordWriterManager.getPartitionDataPath;
 import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
 
-import java.util.Collections;
 import java.util.Map;
 import java.util.UUID;
 import org.apache.beam.sdk.coders.IterableCoder;
@@ -223,9 +222,7 @@ Table getOrCreateTable(
       if (sideInputTableSpecs != null && sideInputTableSpecs.containsKey(tableIdString)) {
         SerializableTableSpec spec = sideInputTableSpecs.get(tableIdString);
         if (spec != null) {
-          Map<String, String> catalogProperties = catalogConfig.getCatalogProperties();
-          return new SideInputTable(
-              spec, catalogProperties != null ? catalogProperties : Collections.emptyMap());
+          return new SideInputTable(spec, catalogConfig);
         }
       }
       return TableCache.getAndRefreshIfStale(