This is an automated email from the ASF dual-hosted git repository.
ahmedabu98 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 17bed431e05 remove shardedKey (#39024)
17bed431e05 is described below
commit 17bed431e05578c732d8844b7eede72dc985afe3
Author: Ahmed Abualsaud <[email protected]>
AuthorDate: Thu Jun 18 14:08:05 2026 -0400
remove shardedKey (#39024)
---
.github/trigger_files/IO_Iceberg_Integration_Tests.json | 2 +-
.../main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java | 11 +++++------
2 files changed, 6 insertions(+), 7 deletions(-)
diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json
b/.github/trigger_files/IO_Iceberg_Integration_Tests.json
index b73af5e61a4..7ab7bcd9a9c 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 dc8106321db..010d7f558a1 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.ParDo;
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 class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
batchManifestFiles.withMaxBufferingDuration(checkStateNotNull(intervalTrigger));
}
- PCollection<KV<ShardedKey<Integer>, Iterable<SerializableDataFile>>>
groupedFiles =
- keyedFiles.apply("GroupDataFilesIntoBatches",
batchDataFiles.withShardedKey());
+ PCollection<KV<Integer, Iterable<SerializableDataFile>>> groupedFiles =
+ keyedFiles.apply("GroupDataFilesIntoBatches", batchDataFiles);
PCollection<KV<String, byte[]>> manifests =
groupedFiles.apply(
@@ -660,7 +659,7 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
* downstream {@link CommitManifestFilesDoFn}.
*/
static class CreateManifests
- extends DoFn<KV<ShardedKey<Integer>, Iterable<SerializableDataFile>>,
KV<String, byte[]>> {
+ extends DoFn<KV<Integer, Iterable<SerializableDataFile>>, KV<String,
byte[]>> {
private final IcebergCatalogConfig catalogConfig;
private final String identifier;
private transient @MonotonicNonNull Table table;
@@ -672,7 +671,7 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
@ProcessElement
public void process(
- @Element KV<ShardedKey<Integer>, Iterable<SerializableDataFile>> batch,
+ @Element KV<Integer, Iterable<SerializableDataFile>> batch,
OutputReceiver<KV<String, byte[]>> output)
throws IOException {
if (!batch.getValue().iterator().hasNext()) {
@@ -682,7 +681,7 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
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(
