diff --git a/CHANGES.md b/CHANGES.md index b389823101ba..1801a2cf9db4 100644 --- a/CHANGES.md +++ b/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 diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergIO.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergIO.java index bbcaef7a0a6f..d98d9944bf3c 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergIO.java +++ b/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, Iceb abstract @Nullable List 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 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 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,9 +579,10 @@ public WriteRows withSideInputTableCache() { *

Note: 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(); } /** @@ -589,11 +590,11 @@ public WriteRows withMaximumCacheSize(int maximumCacheSize) { * *

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(); } /** @@ -601,25 +602,26 @@ public WriteRows withTableRefreshInterval(Duration refreshInterval) { * 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 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> 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()); diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergWriteSchemaTransformProvider.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergWriteSchemaTransformProvider.java index 4f568fb65620..91d0ed73835e 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergWriteSchemaTransformProvider.java +++ b/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 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 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); } } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergIOSideInputTableCacheTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergIOSideInputTableCacheTest.java index 289fabc69918..957dafa78479 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergIOSideInputTableCacheTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergIOSideInputTableCacheTest.java @@ -159,7 +159,7 @@ public void testBatchSingleTableWithSideInputCache() throws Exception { IcebergIO.writeRows(catalogConfig) .to(tableId) .withSideInputTableCache() - .withPollingBuckets(1); + .withTableCachePollingBuckets(1); input.apply("WriteToTable", applyDistribution(write)); PipelineResult result = testPipeline.run(); @@ -251,7 +251,7 @@ public String getTableStringIdentifier(ValueInSingleWindow element) { IcebergIO.writeRows(catalogConfig) .to(dynamicDestinations) .withSideInputTableCache() - .withPollingBuckets(1); + .withTableCachePollingBuckets(1); input.apply("WriteDynamic", applyDistribution(write)); PipelineResult result = testPipeline.run(); @@ -352,8 +352,8 @@ public String getTableStringIdentifier(ValueInSingleWindow element) { IcebergIO.writeRows(catalogConfig) .to(dynamicDestinations) .withSideInputTableCache() - .withMaximumCacheSize(2) - .withPollingBuckets(1); + .withMaximumTableCacheSize(2) + .withTableCachePollingBuckets(1); input.apply("WriteWithSampleCap", applyDistribution(write)); PipelineResult result = testPipeline.run(); @@ -397,8 +397,8 @@ public void testStreamingWithSideInputCache() throws Exception { .to(tableId) .withSideInputTableCache() .withTriggeringFrequency(Duration.standardSeconds(1)) - .withTableRefreshInterval(Duration.standardSeconds(2)) - .withPollingBuckets(1); + .withTableCacheRefreshInterval(Duration.standardSeconds(2)) + .withTableCachePollingBuckets(1); input.apply("StreamingWrite", applyDistribution(write)); PipelineResult result = testPipeline.run(); @@ -467,8 +467,8 @@ public void testStreamingSpecEvolutionWithoutPipelineRestart() throws Exception .to(tableId) .withSideInputTableCache() .withTriggeringFrequency(Duration.standardSeconds(1)) - .withTableRefreshInterval(Duration.standardSeconds(1)) - .withPollingBuckets(1); + .withTableCacheRefreshInterval(Duration.standardSeconds(1)) + .withTableCachePollingBuckets(1); input.apply("StreamingWriteEvolved", write); PipelineResult result = testPipeline.run(); @@ -522,16 +522,16 @@ public void testPreconditionsAndValidation() { TableIdentifier tableId = TableIdentifier.of("default_side_input", "validation_table"); IcebergIO.WriteRows write = IcebergIO.writeRows(catalogConfig).to(tableId); - assertThrows(IllegalArgumentException.class, () -> write.withMaximumCacheSize(0)); - assertThrows(IllegalArgumentException.class, () -> write.withMaximumCacheSize(-1)); + assertThrows(IllegalArgumentException.class, () -> write.withMaximumTableCacheSize(0)); + assertThrows(IllegalArgumentException.class, () -> write.withMaximumTableCacheSize(-1)); assertThrows( - IllegalArgumentException.class, () -> write.withTableRefreshInterval(Duration.ZERO)); + IllegalArgumentException.class, () -> write.withTableCacheRefreshInterval(Duration.ZERO)); - assertThrows(IllegalArgumentException.class, () -> write.withPollingBuckets(0)); - assertThrows(IllegalArgumentException.class, () -> write.withPollingBuckets(-1)); + assertThrows(IllegalArgumentException.class, () -> write.withTableCachePollingBuckets(0)); + assertThrows(IllegalArgumentException.class, () -> write.withTableCachePollingBuckets(-1)); - // Unbounded streaming pipeline with maximumCacheSize must fail at expand + // Unbounded streaming pipeline with maximumTableCacheSize must fail at expand Pipeline p = Pipeline.create(); Schema schema = Schema.builder().addInt64Field("id").build(); TestStream testStream = @@ -543,18 +543,18 @@ public void testPreconditionsAndValidation() { // Sub-options specified without withSideInputTableCache() must fail at expand IcebergIO.WriteRows writeWithMaxCacheOnly = - IcebergIO.writeRows(catalogConfig).to(tableId).withMaximumCacheSize(5); + IcebergIO.writeRows(catalogConfig).to(tableId).withMaximumTableCacheSize(5); assertThrows(IllegalArgumentException.class, () -> streamInput.apply(writeWithMaxCacheOnly)); IcebergIO.WriteRows writeWithRefreshIntervalOnly = IcebergIO.writeRows(catalogConfig) .to(tableId) - .withTableRefreshInterval(Duration.standardMinutes(1)); + .withTableCacheRefreshInterval(Duration.standardMinutes(1)); assertThrows( IllegalArgumentException.class, () -> streamInput.apply(writeWithRefreshIntervalOnly)); IcebergIO.WriteRows writeWithPollingBucketsOnly = - IcebergIO.writeRows(catalogConfig).to(tableId).withPollingBuckets(2); + IcebergIO.writeRows(catalogConfig).to(tableId).withTableCachePollingBuckets(2); assertThrows( IllegalArgumentException.class, () -> streamInput.apply(writeWithPollingBucketsOnly)); @@ -562,7 +562,7 @@ public void testPreconditionsAndValidation() { IcebergIO.writeRows(catalogConfig) .to(tableId) .withSideInputTableCache() - .withMaximumCacheSize(5); + .withMaximumTableCacheSize(5); assertThrows(IllegalArgumentException.class, () -> streamInput.apply(streamWrite)); } @@ -574,9 +574,9 @@ public void testDisplayData() { IcebergIO.writeRows(catalogConfig) .to(tableId) .withSideInputTableCache() - .withMaximumCacheSize(100) - .withTableRefreshInterval(Duration.standardMinutes(10)) - .withPollingBuckets(3); + .withMaximumTableCacheSize(100) + .withTableCacheRefreshInterval(Duration.standardMinutes(10)) + .withTableCachePollingBuckets(3); DisplayData displayData = DisplayData.from(write); Map items = new HashMap<>(); @@ -584,10 +584,10 @@ public void testDisplayData() { items.put(item.getKey(), item.getValue() != null ? item.getValue().toString() : ""); } - assertEquals("true", items.get("usingSideInputTableCache")); - assertEquals("100", items.get("maximumCacheSize")); - assertEquals("600000", items.get("tableRefreshInterval")); - assertEquals("3", items.get("pollingBuckets")); + assertEquals("true", items.get("useSideInputTableCache")); + assertEquals("100", items.get("maximumTableCacheSize")); + assertEquals("600000", items.get("tableCacheRefreshInterval")); + assertEquals("3", items.get("tableCachePollingBuckets")); } private static class EvolveSpecMidExecutionDoFn extends DoFn { diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergWriteSchemaTransformProviderTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergWriteSchemaTransformProviderTest.java index c4ac1e22cbf6..b408445053f2 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergWriteSchemaTransformProviderTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergWriteSchemaTransformProviderTest.java @@ -130,10 +130,10 @@ public void testBuildTransformWithRowAndSideInputCache() { .withFieldValue("table", "test_table_identifier") .withFieldValue("catalog_name", "test-name") .withFieldValue("catalog_properties", properties) - .withFieldValue("using_side_input_table_cache", true) - .withFieldValue("table_refresh_interval_seconds", 60) - .withFieldValue("maximum_cache_size", 100) - .withFieldValue("polling_buckets", 2) + .withFieldValue("use_side_input_table_cache", true) + .withFieldValue("table_cache_refresh_interval_seconds", 60) + .withFieldValue("maximum_table_cache_size", 100) + .withFieldValue("table_cache_polling_buckets", 2) .build(); new IcebergWriteSchemaTransformProvider().from(transformConfigRow); @@ -196,9 +196,9 @@ public void testSimpleAppendWithSideInputCache() { .setCatalogName("name") .setCatalogProperties(properties) .setDistributionMode(distributionMode.name()) - .setUsingSideInputTableCache(true) - .setTableRefreshIntervalSeconds(60) - .setPollingBuckets(1) + .setUseSideInputTableCache(true) + .setTableCacheRefreshIntervalSeconds(60) + .setTableCachePollingBuckets(1) .build(); PCollectionRowTuple input = @@ -274,10 +274,10 @@ public void testWriteUsingManagedTransformWithSideInputCache() { "table: %s\n" + "catalog_name: test-name\n" + "distribution_mode: %s\n" - + "using_side_input_table_cache: true\n" - + "table_refresh_interval_seconds: 60\n" - + "maximum_cache_size: 50\n" - + "polling_buckets: 1\n" + + "use_side_input_table_cache: true\n" + + "table_cache_refresh_interval_seconds: 60\n" + + "maximum_table_cache_size: 50\n" + + "table_cache_polling_buckets: 1\n" + "catalog_properties: \n" + " type: %s\n" + " warehouse: %s", @@ -316,14 +316,14 @@ public void testSideInputCacheSubOptionsRequireExplicitEnablement() { p.apply("DummyInput", Create.of(TestFixtures.asRows(TestFixtures.FILE1SNAPSHOT1))) .setRowSchema(IcebergUtils.icebergSchemaToBeamSchema(TestFixtures.SCHEMA))); - // Setting sub-options when using_side_input_table_cache is not set (null) must throw + // Setting sub-options when use_side_input_table_cache is not set (null) must throw // IllegalArgumentException Configuration configWithMaxCacheSizeOnly = Configuration.builder() .setTable("default.table_max_cache") .setCatalogName("name") .setCatalogProperties(Collections.singletonMap("type", "hadoop")) - .setMaximumCacheSize(50) + .setMaximumTableCacheSize(50) .build(); assertThrows( IllegalArgumentException.class, @@ -334,7 +334,7 @@ public void testSideInputCacheSubOptionsRequireExplicitEnablement() { .setTable("default.table_refresh") .setCatalogName("name") .setCatalogProperties(Collections.singletonMap("type", "hadoop")) - .setTableRefreshIntervalSeconds(60) + .setTableCacheRefreshIntervalSeconds(60) .build(); assertThrows( IllegalArgumentException.class, @@ -345,36 +345,36 @@ public void testSideInputCacheSubOptionsRequireExplicitEnablement() { .setTable("default.table_buckets") .setCatalogName("name") .setCatalogProperties(Collections.singletonMap("type", "hadoop")) - .setPollingBuckets(2) + .setTableCachePollingBuckets(2) .build(); assertThrows( IllegalArgumentException.class, () -> dummyInput.apply(provider.from(configWithPollingBucketsOnly))); - // Setting using_side_input_table_cache to false while setting sub-options must throw + // Setting use_side_input_table_cache to false while setting sub-options must throw // IllegalArgumentException Configuration invalidConfigWithFalse = Configuration.builder() .setTable("default.table_invalid") .setCatalogName("name") .setCatalogProperties(Collections.singletonMap("type", "hadoop")) - .setUsingSideInputTableCache(false) - .setMaximumCacheSize(50) + .setUseSideInputTableCache(false) + .setMaximumTableCacheSize(50) .build(); assertThrows( IllegalArgumentException.class, () -> dummyInput.apply(provider.from(invalidConfigWithFalse))); - // Explicitly setting using_side_input_table_cache to true with sub-options succeeds + // Explicitly setting use_side_input_table_cache to true with sub-options succeeds Configuration validConfig = Configuration.builder() .setTable("default.table_valid") .setCatalogName("name") .setCatalogProperties(Collections.singletonMap("type", "hadoop")) - .setUsingSideInputTableCache(true) - .setMaximumCacheSize(50) - .setTableRefreshIntervalSeconds(60) - .setPollingBuckets(2) + .setUseSideInputTableCache(true) + .setMaximumTableCacheSize(50) + .setTableCacheRefreshIntervalSeconds(60) + .setTableCachePollingBuckets(2) .build(); assertNotNull(dummyInput.apply(provider.from(validConfig))); }