Cherry-picking side-input Table cache renames

CHANGES.md

@@ -71,6 +71,7 @@
 * ClickHouseIO: support writing `Decimal(P, S)` / `Decimal32/64/128/256` columns (Java) ([#39840](https://github.com/apache/beam/issues/39840)).
 * [IcebergIO] AddFiles (`IcebergAddFiles` in YAML) can evolve the table schema before registering files, with `schema_evolution_options`, `required_columns`, `incompatible_schema_handling` and `unverifiable_file_handling` (Java/YAML, batch only) ([#40144](https://github.com/apache/beam/issues/40144)).
 * [IcebergIO] Added batch and streaming CDC writes that applies INSERT/UPDATE_BEFORE/UPDATE_AFTER/DELETE changes to Iceberg V2+ tables by primary key. Invoke with `IcebergIO.writeCdcRows` (Java) or by setting `mode: merge-on-read` on the Managed `ICEBERG` write (Java, Python, YAML) ([#39979](https://github.com/apache/beam/issues/39979)).
+* [IcebergIO] Added an optional side-input table cache for writes to significantly reduce catalog and table requests for large pipelines. A single worker polls the table and broadcasts it to other workers in the pipeline. Enable with `IcebergIO.writeRows(...).withSideInputTableCache()` (Java) or by setting `use_side_input_table_cache: true` on the Managed `ICEBERG` write (Java, Python) ([#39723](https://github.com/apache/beam/issues/39723)).
 
 ## New Features / Improvements
 

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

@@ -393,7 +393,7 @@ public static WriteRows writeRows(IcebergCatalogConfig catalog) {
         .setCatalogConfig(catalog)
         .setDistributionMode(DistributionMode.NONE)
         .setAutoSharding(false)
-        .setUsingSideInputTableCache(false)
+        .setUseSideInputTableCache(false)
         .build();
   }
 
@@ -437,13 +437,13 @@ public abstract static class WriteRows extends PTransform<PCollection<Row>, Iceb
 
     abstract @Nullable List<String> getSortFields();
 
-    abstract boolean getUsingSideInputTableCache();
+    abstract boolean getUseSideInputTableCache();
 
-    abstract @Nullable Integer getMaximumCacheSize();
+    abstract @Nullable Integer getMaximumTableCacheSize();
 
-    abstract @Nullable Duration getTableRefreshInterval();
+    abstract @Nullable Duration getTableCacheRefreshInterval();
 
-    abstract @Nullable Integer getPollingBuckets();
+    abstract @Nullable Integer getTableCachePollingBuckets();
 
     abstract Builder toBuilder();
 
@@ -469,13 +469,13 @@ abstract static class Builder {
 
       abstract Builder setSortFields(List<String> sortFields);
 
-      abstract Builder setUsingSideInputTableCache(boolean usingSideInputTableCache);
+      abstract Builder setUseSideInputTableCache(boolean useSideInputTableCache);
 
-      abstract Builder setMaximumCacheSize(@Nullable Integer maximumCacheSize);
+      abstract Builder setMaximumTableCacheSize(@Nullable Integer maximumTableCacheSize);
 
-      abstract Builder setTableRefreshInterval(@Nullable Duration refreshInterval);
+      abstract Builder setTableCacheRefreshInterval(@Nullable Duration refreshInterval);
 
-      abstract Builder setPollingBuckets(@Nullable Integer pollingBuckets);
+      abstract Builder setTableCachePollingBuckets(@Nullable Integer pollingBuckets);
 
       abstract WriteRows build();
     }
@@ -569,7 +569,7 @@ public WriteRows withSortOrder(List<String> sortFields) {
      * representations without issuing remote catalog RPCs, drastically reducing catalog load.
      */
     public WriteRows withSideInputTableCache() {
-      return toBuilder().setUsingSideInputTableCache(true).build();
+      return toBuilder().setUseSideInputTableCache(true).build();
     }
 
     /**
@@ -579,47 +579,49 @@ public WriteRows withSideInputTableCache() {
      * <p><b>Note:</b> This option is only supported for bounded (batch) pipelines. Calling this on
      * an unbounded streaming pipeline will throw an exception at pipeline construction.
      */
-    public WriteRows withMaximumCacheSize(int maximumCacheSize) {
-      Preconditions.checkArgument(maximumCacheSize > 0, "maximumCacheSize must be greater than 0");
-      return toBuilder().setMaximumCacheSize(maximumCacheSize).build();
+    public WriteRows withMaximumTableCacheSize(int maximumTableCacheSize) {
+      Preconditions.checkArgument(
+          maximumTableCacheSize > 0, "maximumTableCacheSize must be greater than 0");
+      return toBuilder().setMaximumTableCacheSize(maximumTableCacheSize).build();
     }
 
     /**
      * Sets the interval at which table metadata is refreshed from the Iceberg catalog.
      *
      * <p>Applicable for unbounded streaming pipelines. Defaults to 5 minutes.
      */
-    public WriteRows withTableRefreshInterval(Duration refreshInterval) {
+    public WriteRows withTableCacheRefreshInterval(Duration refreshInterval) {
       Preconditions.checkNotNull(refreshInterval, "refreshInterval must not be null");
       Preconditions.checkArgument(
           refreshInterval.isLongerThan(Duration.ZERO), "refreshInterval must be greater than 0");
-      return toBuilder().setTableRefreshInterval(refreshInterval).build();
+      return toBuilder().setTableCacheRefreshInterval(refreshInterval).build();
     }
 
     /**
      * Sets the number of parallel buckets/workers used to query the Iceberg catalog during
      * refreshes. Defaults to 1 to serialize catalog queries and protect catalogs from connection
      * spikes.
      */
-    public WriteRows withPollingBuckets(int pollingBuckets) {
-      Preconditions.checkArgument(pollingBuckets > 0, "pollingBuckets must be greater than 0");
-      return toBuilder().setPollingBuckets(pollingBuckets).build();
+    public WriteRows withTableCachePollingBuckets(int pollingBuckets) {
+      Preconditions.checkArgument(
+          pollingBuckets > 0, "tableCachePollingBuckets must be greater than 0");
+      return toBuilder().setTableCachePollingBuckets(pollingBuckets).build();
     }
 
     @Override
     public void populateDisplayData(DisplayData.Builder builder) {
       super.populateDisplayData(builder);
       builder.add(
-          DisplayData.item("usingSideInputTableCache", getUsingSideInputTableCache())
+          DisplayData.item("useSideInputTableCache", getUseSideInputTableCache())
               .withLabel("Using Side-Input Table Cache"));
       builder.addIfNotNull(
-          DisplayData.item("maximumCacheSize", getMaximumCacheSize())
+          DisplayData.item("maximumTableCacheSize", getMaximumTableCacheSize())
               .withLabel("Maximum Cache Size"));
       builder.addIfNotNull(
-          DisplayData.item("tableRefreshInterval", getTableRefreshInterval())
+          DisplayData.item("tableCacheRefreshInterval", getTableCacheRefreshInterval())
               .withLabel("Table Refresh Interval"));
       builder.addIfNotNull(
-          DisplayData.item("pollingBuckets", getPollingBuckets())
+          DisplayData.item("tableCachePollingBuckets", getTableCachePollingBuckets())
               .withLabel("Catalog Polling Buckets"));
     }
 
@@ -649,29 +651,29 @@ public IcebergWriteResult expand(PCollection<Row> input) {
       }
 
       boolean hasSideInputOptions =
-          getMaximumCacheSize() != null
-              || getTableRefreshInterval() != null
-              || getPollingBuckets() != null;
+          getMaximumTableCacheSize() != null
+              || getTableCacheRefreshInterval() != null
+              || getTableCachePollingBuckets() != null;
       Preconditions.checkArgument(
-          getUsingSideInputTableCache() || !hasSideInputOptions,
-          "Cannot specify side-input cache sub-options (maximumCacheSize, "
-              + "tableRefreshInterval, pollingBuckets) without enabling side-input table cache via withSideInputTableCache().");
+          getUseSideInputTableCache() || !hasSideInputOptions,
+          "Cannot specify side-input cache sub-options (maximumTableCacheSize, "
+              + "tableCacheRefreshInterval, tableCachePollingBuckets) without enabling side-input table cache via withSideInputTableCache().");
 
       PCollectionView<Map<String, SerializableTableSpec>> metadataView = null;
-      if (getUsingSideInputTableCache()) {
+      if (getUseSideInputTableCache()) {
         TableMetadataDriver.Builder driverBuilder =
             TableMetadataDriver.builder()
                 .setCatalogConfig(getCatalogConfig())
                 .setDynamicDestinations(destinations);
 
-        if (getMaximumCacheSize() != null) {
-          driverBuilder.setMaximumCacheSize(getMaximumCacheSize());
+        if (getMaximumTableCacheSize() != null) {
+          driverBuilder.setMaximumCacheSize(getMaximumTableCacheSize());
         }
-        if (getTableRefreshInterval() != null) {
-          driverBuilder.setRefreshInterval(getTableRefreshInterval());
+        if (getTableCacheRefreshInterval() != null) {
+          driverBuilder.setRefreshInterval(getTableCacheRefreshInterval());
         }
-        if (getPollingBuckets() != null) {
-          driverBuilder.setPollingBuckets(getPollingBuckets());
+        if (getTableCachePollingBuckets() != null) {
+          driverBuilder.setPollingBuckets(getTableCachePollingBuckets());
         }
 
         metadataView = input.apply("GenerateTableMetadataView", driverBuilder.build().asView());

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

@@ -212,20 +212,20 @@ public static Builder builder() {
 
     @SchemaFieldDescription(
         "Enables expirable side-input caching of Iceberg table metadata across workers to reduce catalog load.")
-    public abstract @Nullable Boolean getUsingSideInputTableCache();
+    public abstract @Nullable Boolean getUseSideInputTableCache();
 
     @SchemaFieldDescription(
         "For a streaming pipeline, sets the interval in seconds at which table metadata is refreshed from the catalog.")
-    public abstract @Nullable Integer getTableRefreshIntervalSeconds();
+    public abstract @Nullable Integer getTableCacheRefreshIntervalSeconds();
 
     @SchemaFieldDescription(
         "For a batch pipeline, sets the maximum number of table metadata specs to cache in memory. "
             + "Tables exceeding this limit fall back to worker-local catalog loading.")
-    public abstract @Nullable Integer getMaximumCacheSize();
+    public abstract @Nullable Integer getMaximumTableCacheSize();
 
     @SchemaFieldDescription(
         "Sets the number of parallel buckets/workers used to query the Iceberg catalog during refreshes. Defaults to 1.")
-    public abstract @Nullable Integer getPollingBuckets();
+    public abstract @Nullable Integer getTableCachePollingBuckets();
 
     @SchemaFieldDescription(
         "Columns defining row identity (equality-delete fields). Defaults to the destination table's "
@@ -316,13 +316,14 @@ public abstract static class Builder {
 
       public abstract Builder setWriteProperties(Map<String, String> writeProperties);
 
-      public abstract Builder setUsingSideInputTableCache(Boolean usingSideInputTableCache);
+      public abstract Builder setUseSideInputTableCache(Boolean useSideInputTableCache);
 
-      public abstract Builder setTableRefreshIntervalSeconds(Integer tableRefreshIntervalSeconds);
+      public abstract Builder setTableCacheRefreshIntervalSeconds(
+          Integer tableCacheRefreshIntervalSeconds);
 
-      public abstract Builder setMaximumCacheSize(Integer maximumCacheSize);
+      public abstract Builder setMaximumTableCacheSize(Integer maximumTableCacheSize);
 
-      public abstract Builder setPollingBuckets(Integer pollingBuckets);
+      public abstract Builder setTableCachePollingBuckets(Integer pollingBuckets);
 
       public abstract Builder setMode(String mode);
 
@@ -419,14 +420,14 @@ void validateModeOptions() {
       requireMode(Mode.APPEND, "autosharding", getAutosharding(), unsupported);
       requireMode(Mode.APPEND, "write_properties", getWriteProperties(), unsupported);
       requireMode(
-          Mode.APPEND, "using_side_input_table_cache", getUsingSideInputTableCache(), unsupported);
+          Mode.APPEND, "using_side_input_table_cache", getUseSideInputTableCache(), unsupported);
       requireMode(
           Mode.APPEND,
           "table_refresh_interval_seconds",
-          getTableRefreshIntervalSeconds(),
+          getTableCacheRefreshIntervalSeconds(),
           unsupported);
-      requireMode(Mode.APPEND, "maximum_cache_size", getMaximumCacheSize(), unsupported);
-      requireMode(Mode.APPEND, "polling_buckets", getPollingBuckets(), unsupported);
+      requireMode(Mode.APPEND, "maximum_cache_size", getMaximumTableCacheSize(), unsupported);
+      requireMode(Mode.APPEND, "polling_buckets", getTableCachePollingBuckets(), unsupported);
       if (!unsupported.isEmpty()) {
         throw new IllegalArgumentException(
             String.format(
@@ -535,34 +536,32 @@ private PCollectionRowTuple expandAppend(PCollection<Row> rows) {
       }
 
       boolean hasSideInputOptions =
-          configuration.getTableRefreshIntervalSeconds() != null
-              || configuration.getMaximumCacheSize() != null
-              || configuration.getPollingBuckets() != null;
+          configuration.getTableCacheRefreshIntervalSeconds() != null
+              || configuration.getMaximumTableCacheSize() != null
+              || configuration.getTableCachePollingBuckets() != null;
 
-      if (!Boolean.TRUE.equals(configuration.getUsingSideInputTableCache())
-          && hasSideInputOptions) {
+      if (!Boolean.TRUE.equals(configuration.getUseSideInputTableCache()) && hasSideInputOptions) {
         throw new IllegalArgumentException(
             "Cannot specify side-input cache sub-options (table_refresh_interval_seconds, "
                 + "maximum_cache_size, polling_buckets) without explicitly setting using_side_input_table_cache to true.");
       }
 
-      boolean enableSideInputCache =
-          Boolean.TRUE.equals(configuration.getUsingSideInputTableCache());
+      boolean enableSideInputCache = Boolean.TRUE.equals(configuration.getUseSideInputTableCache());
 
       if (enableSideInputCache) {
         writeTransform = writeTransform.withSideInputTableCache();
-        @Nullable Integer refreshSec = configuration.getTableRefreshIntervalSeconds();
+        @Nullable Integer refreshSec = configuration.getTableCacheRefreshIntervalSeconds();
         if (refreshSec != null) {
           writeTransform =
-              writeTransform.withTableRefreshInterval(Duration.standardSeconds(refreshSec));
+              writeTransform.withTableCacheRefreshInterval(Duration.standardSeconds(refreshSec));
         }
-        @Nullable Integer maxCacheSize = configuration.getMaximumCacheSize();
+        @Nullable Integer maxCacheSize = configuration.getMaximumTableCacheSize();
         if (maxCacheSize != null) {
-          writeTransform = writeTransform.withMaximumCacheSize(maxCacheSize);
+          writeTransform = writeTransform.withMaximumTableCacheSize(maxCacheSize);
         }
-        @Nullable Integer pollingBuckets = configuration.getPollingBuckets();
+        @Nullable Integer pollingBuckets = configuration.getTableCachePollingBuckets();
         if (pollingBuckets != null) {
-          writeTransform = writeTransform.withPollingBuckets(pollingBuckets);
+          writeTransform = writeTransform.withTableCachePollingBuckets(pollingBuckets);
         }
       }