diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json b/.github/trigger_files/IO_Iceberg_Integration_Tests.json index b73af5e61a43..7ab7bcd9a9c6 100644 --- a/.github/trigger_files/IO_Iceberg_Integration_Tests.json +++ b/.github/trigger_files/IO_Iceberg_Integration_Tests.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 1 + "modification": 2 } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java index dc8106321db7..010d7f558a12 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java @@ -65,7 +65,6 @@ import org.apache.beam.sdk.transforms.WithKeys; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.PaneInfo; -import org.apache.beam.sdk.util.ShardedKey; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollectionRowTuple; @@ -226,8 +225,8 @@ public PCollectionRowTuple expand(PCollection input) { batchManifestFiles.withMaxBufferingDuration(checkStateNotNull(intervalTrigger)); } - PCollection, Iterable>> groupedFiles = - keyedFiles.apply("GroupDataFilesIntoBatches", batchDataFiles.withShardedKey()); + PCollection>> groupedFiles = + keyedFiles.apply("GroupDataFilesIntoBatches", batchDataFiles); PCollection> manifests = groupedFiles.apply( @@ -660,7 +659,7 @@ static String getPartitionFromMetrics(Metrics metrics, InputFile inputFile, Tabl * downstream {@link CommitManifestFilesDoFn}. */ static class CreateManifests - extends DoFn, Iterable>, KV> { + extends DoFn>, KV> { private final IcebergCatalogConfig catalogConfig; private final String identifier; private transient @MonotonicNonNull Table table; @@ -672,7 +671,7 @@ public CreateManifests(IcebergCatalogConfig catalogConfig, String identifier) { @ProcessElement public void process( - @Element KV, Iterable> batch, + @Element KV> batch, OutputReceiver> output) throws IOException { if (!batch.getValue().iterator().hasNext()) { @@ -682,7 +681,7 @@ public void process( table = catalogConfig.catalog().loadTable(TableIdentifier.parse(identifier)); } - PartitionSpec spec = checkStateNotNull(table.specs().get(batch.getKey().getKey())); + PartitionSpec spec = checkStateNotNull(table.specs().get(batch.getKey())); String manifestPath = String.format(