From e3b71944e0e72c4599b42aac1f69b3c929169262 Mon Sep 17 00:00:00 2001 From: ntopousis <83309247+ntopousis@users.noreply.github.com> Date: Fri, 17 Jul 2026 13:07:50 -0400 Subject: [PATCH 1/5] Avoid side input when gathering WriteFiles results --- .../org/apache/beam/sdk/io/WriteFiles.java | 66 ++++++++++++------- .../apache/beam/sdk/io/WriteFilesTest.java | 21 ++++++ 2 files changed, 62 insertions(+), 25 deletions(-) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java index b0b5051f3210..2215af5d2d4b 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java @@ -25,6 +25,7 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.UUID; @@ -40,6 +41,7 @@ import org.apache.beam.sdk.coders.ShardedKeyCoder; import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.coders.VarIntCoder; +import org.apache.beam.sdk.coders.VoidCoder; import org.apache.beam.sdk.io.FileBasedSink.DynamicDestinations; import org.apache.beam.sdk.io.FileBasedSink.FileResult; import org.apache.beam.sdk.io.FileBasedSink.FileResultCoder; @@ -49,6 +51,7 @@ import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.ValueProvider; import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider; +import org.apache.beam.sdk.transforms.Create; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.DoFn.MultiOutputReceiver; import org.apache.beam.sdk.transforms.Flatten; @@ -57,11 +60,9 @@ import org.apache.beam.sdk.transforms.MapElements; import org.apache.beam.sdk.transforms.PTransform; import org.apache.beam.sdk.transforms.ParDo; -import org.apache.beam.sdk.transforms.Reify; import org.apache.beam.sdk.transforms.Reshuffle; import org.apache.beam.sdk.transforms.SimpleFunction; import org.apache.beam.sdk.transforms.Values; -import org.apache.beam.sdk.transforms.View; import org.apache.beam.sdk.transforms.WithKeys; import org.apache.beam.sdk.transforms.display.DisplayData; import org.apache.beam.sdk.transforms.errorhandling.BadRecord; @@ -553,29 +554,44 @@ public PCollection> expand(PCollection input) { // Reshuffle one more time to stabilize the contents of the bundle lists to finalize. .apply(Reshuffle.viaRandomKey()); } else { - // Pass results via a side input rather than reshuffle, because we need to get an empty - // iterable to finalize if there are no results. - return input - .getPipeline() - .apply( - "AsPossiblyEmptyList", - Reify.viewInGlobalWindow( - // Insert a reshuffle before taking the view to consolidate the (typically) - // one-output-per-bundle writes. - // This avoids producing a huge number of tiny files in the case that side - // inputs are materialized to disk bundle-by-bundle. - input.apply("Consolidate", Reshuffle.viaRandomKey()).apply(View.asIterable()), - IterableCoder.of(resultCoder))) - // View.asIterable() can be (significantly) cheaper than View.asList(), as it does not - // create a backing indexable view, but we must return a list to maintain update - // compatibility for consumers that are shared between this path and the streaming one. - .apply( - "IterableToList", - MapElements.via( - new SimpleFunction, List>( - x -> ImmutableList.copyOf(x)) {})) - .setCoder(ListCoder.of(resultCoder)); + Coder> resultListCoder = ListCoder.of(resultCoder); + PCollection> resultBundles = + input + .apply("Gather bundles", ParDo.of(new GatherBundlesPerWindowFn<>())) + .setCoder(resultListCoder); + + // Add an empty list so empty, unwindowed writes still reach finalization. Grouping the + // bundle-sized lists on the main-input path also checkpoints the FileResults against + // retries without materializing and reading them back as a global side input. + PCollection> emptyResultList = + input + .getPipeline() + .apply( + "CreateEmptyResultList", + Create.>of(Collections.singletonList(Collections.emptyList())) + .withCoder(resultListCoder)); + + return PCollectionList.of(resultBundles) + .and(emptyResultList) + .apply("EnsureNonEmpty", Flatten.pCollections()) + .apply("Add void key", WithKeys.of((Void) null)) + .setCoder(KvCoder.of(VoidCoder.of(), resultListCoder)) + .apply("Gather all results", GroupByKey.create()) + .apply("Extract results", ParDo.of(new ExtractGatheredResultsFn<>())) + .setCoder(resultListCoder); + } + } + } + + private static class ExtractGatheredResultsFn + extends DoFn>>, List> { + @ProcessElement + public void process(ProcessContext c) { + List results = new ArrayList<>(); + for (List bundleResults : c.element().getValue()) { + results.addAll(bundleResults); } + c.output(results); } } @@ -1350,7 +1366,7 @@ public void process(ProcessContext c) throws Exception { } else { fixedNumShards = null; } - List> fileResults = Lists.newArrayList(c.element()); + List> fileResults = c.element(); LOG.info("Finalizing {} file results", fileResults.size()); if (fileResults.isEmpty() && getSkipIfEmpty()) { return; diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteFilesTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteFilesTest.java index cc174002bb46..643b1dd934b9 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteFilesTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteFilesTest.java @@ -61,6 +61,7 @@ import org.apache.beam.sdk.options.Description; import org.apache.beam.sdk.options.PipelineOptionsFactoryTest.TestPipelineOptions; import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider; +import org.apache.beam.sdk.runners.TransformHierarchy; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -223,6 +224,26 @@ public void testEmptyWrite() throws IOException { true /* expectRemovedTempDirectory */); } + @Test + public void testUnwindowedGatherDoesNotMaterializeResultsAsSideInput() { + Pipeline pipeline = Pipeline.create(); + pipeline.apply(Create.of("foo")).apply(WriteFiles.to(makeSimpleSink())); + + List resultViews = new ArrayList<>(); + pipeline.traverseTopologically( + new Pipeline.PipelineVisitor.Defaults() { + @Override + public void visitPrimitiveTransform(TransformHierarchy.Node node) { + if (node.getFullName().contains("GatherTempFileResults") + && node.getTransform() instanceof View.CreatePCollectionView) { + resultViews.add(node.getFullName()); + } + } + }); + + assertThat(resultViews, Matchers.empty()); + } + /** * Test that WriteFiles with a configured number of shards produces the desired number of shards * even when there are many elements. From 069c48269c129f664f1f804e1787f5c07fa20ae2 Mon Sep 17 00:00:00 2001 From: ntopousis <83309247+ntopousis@users.noreply.github.com> Date: Fri, 17 Jul 2026 15:29:58 -0400 Subject: [PATCH 2/5] Stabilize WriteFiles finalization across retries (#39370) --- .../org/apache/beam/sdk/io/WriteFiles.java | 21 +++++++- .../apache/beam/sdk/io/WriteFilesTest.java | 48 +++++++++++++++++++ 2 files changed, 68 insertions(+), 1 deletion(-) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java index 2215af5d2d4b..156488293435 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java @@ -26,6 +26,7 @@ import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.Comparator; import java.util.List; import java.util.Map; import java.util.UUID; @@ -74,6 +75,7 @@ import org.apache.beam.sdk.transforms.windowing.GlobalWindow; import org.apache.beam.sdk.transforms.windowing.GlobalWindows; import org.apache.beam.sdk.transforms.windowing.PaneInfo; +import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; import org.apache.beam.sdk.transforms.windowing.Window; import org.apache.beam.sdk.util.CoderUtils; import org.apache.beam.sdk.util.MoreFutures; @@ -89,6 +91,7 @@ import org.apache.beam.sdk.values.ShardedKey; import org.apache.beam.sdk.values.TupleTag; import org.apache.beam.sdk.values.TupleTagList; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Objects; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ArrayListMultimap; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; @@ -576,6 +579,12 @@ public PCollection> expand(PCollection input) { .apply("EnsureNonEmpty", Flatten.pCollections()) .apply("Add void key", WithKeys.of((Void) null)) .setCoder(KvCoder.of(VoidCoder.of(), resultListCoder)) + // The old side-input path emitted its list at the minimum timestamp. Keep that + // observable timestamp by combining with the minimum-timestamp empty marker. + .apply( + "Preserve minimum timestamp", + Window.>>configure() + .withTimestampCombiner(TimestampCombiner.EARLIEST)) .apply("Gather all results", GroupByKey.create()) .apply("Extract results", ParDo.of(new ExtractGatheredResultsFn<>())) .setCoder(resultListCoder); @@ -1366,7 +1375,12 @@ public void process(ProcessContext c) throws Exception { } else { fixedNumShards = null; } - List> fileResults = c.element(); + List> fileResults = Lists.newArrayList(c.element()); + if (!getWindowedWrites() && fixedNumShards == null) { + // GroupByKey preserves the set of results across retries, but not their iteration order. + // Runner-determined shard numbers are assigned by position, so make that mapping stable. + sortByTempFilename(fileResults); + } LOG.info("Finalizing {} file results", fileResults.size()); if (fileResults.isEmpty() && getSkipIfEmpty()) { return; @@ -1386,6 +1400,11 @@ public void process(ProcessContext c) throws Exception { } } + @VisibleForTesting + static void sortByTempFilename(List> fileResults) { + fileResults.sort(Comparator.comparing(result -> result.getTempFilename().toString())); + } + private List, ResourceId>> finalizeAllDestinations( List> fileResults, @Nullable Integer fixedNumShards) throws Exception { diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteFilesTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteFilesTest.java index 643b1dd934b9..4eb4342fba21 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteFilesTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteFilesTest.java @@ -51,6 +51,7 @@ import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.io.DefaultFilenamePolicy.Params; import org.apache.beam.sdk.io.FileBasedSink.DynamicDestinations; +import org.apache.beam.sdk.io.FileBasedSink.FileResult; import org.apache.beam.sdk.io.FileBasedSink.FilenamePolicy; import org.apache.beam.sdk.io.FileBasedSink.OutputFileHints; import org.apache.beam.sdk.io.SimpleSink.SimpleWriter; @@ -85,6 +86,7 @@ import org.apache.beam.sdk.transforms.errorhandling.ErrorHandlingTestUtils.ErrorSinkTransform; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.FixedWindows; +import org.apache.beam.sdk.transforms.windowing.GlobalWindow; import org.apache.beam.sdk.transforms.windowing.IntervalWindow; import org.apache.beam.sdk.transforms.windowing.PaneInfo; import org.apache.beam.sdk.transforms.windowing.Sessions; @@ -190,6 +192,13 @@ public PDone expand(PCollection> input) { } } + private static class ExtractTimestampFn extends DoFn { + @ProcessElement + public void process(ProcessContext c) { + c.output(c.timestamp().getMillis()); + } + } + private String getBaseOutputFilename() { return getBaseOutputDirectory().resolve("file", StandardResolveOptions.RESOLVE_FILE).toString(); } @@ -244,6 +253,45 @@ public void visitPrimitiveTransform(TransformHierarchy.Node node) { assertThat(resultViews, Matchers.empty()); } + @Test + public void testSortByTempFilename() { + FileResult laterResult = + new FileResult<>( + getBaseOutputDirectory().resolve("z", StandardResolveOptions.RESOLVE_FILE), + WriteFiles.UNKNOWN_SHARDNUM, + GlobalWindow.INSTANCE, + PaneInfo.NO_FIRING, + null); + FileResult earlierResult = + new FileResult<>( + getBaseOutputDirectory().resolve("a", StandardResolveOptions.RESOLVE_FILE), + WriteFiles.UNKNOWN_SHARDNUM, + GlobalWindow.INSTANCE, + PaneInfo.NO_FIRING, + null); + List> results = Lists.newArrayList(laterResult, earlierResult); + + WriteFiles.sortByTempFilename(results); + + assertThat(results, equalTo(Arrays.asList(earlierResult, laterResult))); + } + + @Test + @Category(NeedsRunner.class) + public void testUnwindowedOutputFilenamesKeepMinimumTimestamp() { + WriteFilesResult result = + p.apply(Create.of("foo")).apply(WriteFiles.to(makeSimpleSink()).withNumShards(1)); + + PCollection outputTimestamps = + result + .getPerDestinationOutputFilenames() + .apply("Extract output timestamps", ParDo.of(new ExtractTimestampFn<>())); + + PAssert.that(outputTimestamps) + .containsInAnyOrder(BoundedWindow.TIMESTAMP_MIN_VALUE.getMillis()); + p.run(); + } + /** * Test that WriteFiles with a configured number of shards produces the desired number of shards * even when there are many elements. From a46eaed283a36e505a9fb199f53b167b27f3079e Mon Sep 17 00:00:00 2001 From: ntopousis <83309247+ntopousis@users.noreply.github.com> Date: Fri, 17 Jul 2026 15:30:16 -0400 Subject: [PATCH 3/5] Document WriteFiles finalization fix (#39370) --- CHANGES.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGES.md b/CHANGES.md index a6fd20ad34dd..9f204fbf51ee 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -82,6 +82,7 @@ ## Bugfixes * Fixed unbounded checkpoint state growth for splittable DoFns that self-checkpoint on the portable Flink runner (Java) ([#27648](https://github.com/apache/beam/issues/27648)). +* Fixed bounded, unwindowed Java `WriteFiles` finalization to gather temporary-file results through a main-input shuffle instead of a global side input ([#39370](https://github.com/apache/beam/issues/39370)). * Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)). ## Security Fixes From ec9f997e46c42944a7f75f7d5ca83c4dba3d2c54 Mon Sep 17 00:00:00 2001 From: ntopousis <83309247+ntopousis@users.noreply.github.com> Date: Fri, 17 Jul 2026 16:27:53 -0400 Subject: [PATCH 4/5] Preserve WriteFiles finalization compatibility (#39370) --- .../org/apache/beam/sdk/io/FileBasedSink.java | 9 +++- .../org/apache/beam/sdk/io/WriteFiles.java | 18 ++------ .../apache/beam/sdk/io/FileBasedSinkTest.java | 46 +++++++++++++++++++ .../apache/beam/sdk/io/WriteFilesTest.java | 31 ++++--------- 4 files changed, 68 insertions(+), 36 deletions(-) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java index bba9b1f82f5b..5845d80ae3a6 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java @@ -34,6 +34,7 @@ import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.Comparator; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -662,8 +663,14 @@ protected final List, ResourceId>> finalizeDestinati if (numShards != null) { resultsWithShardNumbers = Lists.newArrayList(completeResults); } else { + List> orderedResults = Lists.newArrayList(completeResults); + if (!windowedWrites) { + // Runner-determined shard numbers are positional. Use a stable order so retries after a + // partial rename cannot map the same temporary file to a different final shard. + orderedResults.sort(Comparator.comparing(result -> result.getTempFilename().toString())); + } int i = 0; - for (FileResult res : completeResults) { + for (FileResult res : orderedResults) { resultsWithShardNumbers.add(res.withShard(i++)); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java index 156488293435..45158e24b2f4 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java @@ -26,7 +26,6 @@ import java.util.ArrayList; import java.util.Collection; import java.util.Collections; -import java.util.Comparator; import java.util.List; import java.util.Map; import java.util.UUID; @@ -91,7 +90,7 @@ import org.apache.beam.sdk.values.ShardedKey; import org.apache.beam.sdk.values.TupleTag; import org.apache.beam.sdk.values.TupleTagList; -import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; +import org.apache.beam.sdk.values.WindowingStrategy; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Objects; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ArrayListMultimap; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; @@ -587,7 +586,10 @@ public PCollection> expand(PCollection input) { .withTimestampCombiner(TimestampCombiner.EARLIEST)) .apply("Gather all results", GroupByKey.create()) .apply("Extract results", ParDo.of(new ExtractGatheredResultsFn<>())) - .setCoder(resultListCoder); + .setCoder(resultListCoder) + // EARLIEST is internal to gathering. The old side-input path exposed the global + // default strategy, so restore it before finalization and filename output. + .setWindowingStrategyInternal(WindowingStrategy.globalDefault()); } } } @@ -1376,11 +1378,6 @@ public void process(ProcessContext c) throws Exception { fixedNumShards = null; } List> fileResults = Lists.newArrayList(c.element()); - if (!getWindowedWrites() && fixedNumShards == null) { - // GroupByKey preserves the set of results across retries, but not their iteration order. - // Runner-determined shard numbers are assigned by position, so make that mapping stable. - sortByTempFilename(fileResults); - } LOG.info("Finalizing {} file results", fileResults.size()); if (fileResults.isEmpty() && getSkipIfEmpty()) { return; @@ -1400,11 +1397,6 @@ public void process(ProcessContext c) throws Exception { } } - @VisibleForTesting - static void sortByTempFilename(List> fileResults) { - fileResults.sort(Comparator.comparing(result -> result.getTempFilename().toString())); - } - private List, ResourceId>> finalizeAllDestinations( List> fileResults, @Nullable Integer fixedNumShards) throws Exception { diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileBasedSinkTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileBasedSinkTest.java index c4f83954e66c..63f7ac870dc3 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileBasedSinkTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileBasedSinkTest.java @@ -205,6 +205,52 @@ public void testFinalizeWithIntermediateState() throws Exception { runFinalize(writeOp, files); } + @Test + public void testRunnerDeterminedShardAssignmentIsStableAcrossResultOrder() throws Exception { + SimpleSink.SimpleWriteOperation writeOp = buildWriteOperation(); + ResourceId earlierTempFilename = + getBaseTempDirectory().resolve("a", StandardResolveOptions.RESOLVE_FILE); + ResourceId laterTempFilename = + getBaseTempDirectory().resolve("z", StandardResolveOptions.RESOLVE_FILE); + FileResult earlierResult = + new FileResult<>( + earlierTempFilename, + UNKNOWN_SHARDNUM, + GlobalWindow.INSTANCE, + PaneInfo.ON_TIME_AND_ONLY_FIRING, + null); + FileResult laterResult = + new FileResult<>( + laterTempFilename, + UNKNOWN_SHARDNUM, + GlobalWindow.INSTANCE, + PaneInfo.ON_TIME_AND_ONLY_FIRING, + null); + + List, ResourceId>> forward = + writeOp.finalizeDestination( + null, GlobalWindow.INSTANCE, null, Arrays.asList(earlierResult, laterResult)); + List, ResourceId>> reversed = + writeOp.finalizeDestination( + null, GlobalWindow.INSTANCE, null, Arrays.asList(laterResult, earlierResult)); + + assertEquals(tempToFinalFilenameMappings(forward), tempToFinalFilenameMappings(reversed)); + assertEquals(earlierTempFilename, forward.get(0).getKey().getTempFilename()); + assertEquals(0, forward.get(0).getKey().getShard()); + assertEquals(laterTempFilename, forward.get(1).getKey().getTempFilename()); + assertEquals(1, forward.get(1).getKey().getShard()); + } + + private static List tempToFinalFilenameMappings( + List, ResourceId>> resultsToFinalFilenames) { + List mappings = new ArrayList<>(); + for (KV, ResourceId> entry : resultsToFinalFilenames) { + mappings.add(entry.getKey().getTempFilename() + " -> " + entry.getValue()); + } + Collections.sort(mappings); + return mappings; + } + /** Generate n temporary files using the temporary file pattern of Writer. */ private List generateTemporaryFilesForFinalize(int numFiles) throws Exception { List temporaryFiles = new ArrayList<>(); diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteFilesTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteFilesTest.java index 4eb4342fba21..ae4a7e07e961 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteFilesTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteFilesTest.java @@ -51,7 +51,6 @@ import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.io.DefaultFilenamePolicy.Params; import org.apache.beam.sdk.io.FileBasedSink.DynamicDestinations; -import org.apache.beam.sdk.io.FileBasedSink.FileResult; import org.apache.beam.sdk.io.FileBasedSink.FilenamePolicy; import org.apache.beam.sdk.io.FileBasedSink.OutputFileHints; import org.apache.beam.sdk.io.SimpleSink.SimpleWriter; @@ -86,7 +85,6 @@ import org.apache.beam.sdk.transforms.errorhandling.ErrorHandlingTestUtils.ErrorSinkTransform; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.FixedWindows; -import org.apache.beam.sdk.transforms.windowing.GlobalWindow; import org.apache.beam.sdk.transforms.windowing.IntervalWindow; import org.apache.beam.sdk.transforms.windowing.PaneInfo; import org.apache.beam.sdk.transforms.windowing.Sessions; @@ -97,6 +95,7 @@ import org.apache.beam.sdk.values.PCollectionView; import org.apache.beam.sdk.values.PDone; import org.apache.beam.sdk.values.ShardedKey; +import org.apache.beam.sdk.values.WindowingStrategy; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Optional; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; @@ -254,26 +253,14 @@ public void visitPrimitiveTransform(TransformHierarchy.Node node) { } @Test - public void testSortByTempFilename() { - FileResult laterResult = - new FileResult<>( - getBaseOutputDirectory().resolve("z", StandardResolveOptions.RESOLVE_FILE), - WriteFiles.UNKNOWN_SHARDNUM, - GlobalWindow.INSTANCE, - PaneInfo.NO_FIRING, - null); - FileResult earlierResult = - new FileResult<>( - getBaseOutputDirectory().resolve("a", StandardResolveOptions.RESOLVE_FILE), - WriteFiles.UNKNOWN_SHARDNUM, - GlobalWindow.INSTANCE, - PaneInfo.NO_FIRING, - null); - List> results = Lists.newArrayList(laterResult, earlierResult); - - WriteFiles.sortByTempFilename(results); - - assertThat(results, equalTo(Arrays.asList(earlierResult, laterResult))); + public void testUnwindowedOutputFilenamesKeepGlobalDefaultWindowingStrategy() { + Pipeline pipeline = Pipeline.create(); + WriteFilesResult result = + pipeline.apply(Create.of("foo")).apply(WriteFiles.to(makeSimpleSink())); + + assertThat( + result.getPerDestinationOutputFilenames().getWindowingStrategy(), + equalTo(WindowingStrategy.globalDefault())); } @Test From e8484c6063ef4365559ad2167b8f634cf05e3612 Mon Sep 17 00:00:00 2001 From: ntopousis <83309247+ntopousis@users.noreply.github.com> Date: Fri, 17 Jul 2026 16:32:33 -0400 Subject: [PATCH 5/5] Apply CHANGES formatter (#39370) --- CHANGES.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGES.md b/CHANGES.md index 9f204fbf51ee..4448661ce44f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -2508,4 +2508,4 @@ Schema Options, it will be removed in version `2.23.0`. ([BEAM-9704](https://iss ## Highlights -- For versions 2.19.0 and older release notes are available on [Apache Beam Blog](https://beam.apache.org/blog/). +- For versions 2.19.0 and older release notes are available on [Apache Beam Blog](https://beam.apache.org/blog/). \ No newline at end of file