chamikaramj commented on code in PR #40006: URL: https://github.com/apache/beam/pull/40006#discussion_r3964943728
########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitWindows.java: ########## @@ -0,0 +1,224 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg.cdc.sink; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; + +import java.util.Map; +import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.coders.KvCoder; +import org.apache.beam.sdk.coders.RowCoder; +import org.apache.beam.sdk.extensions.sorter.BufferedExternalSorter; +import org.apache.beam.sdk.extensions.sorter.SortValues; +import org.apache.beam.sdk.schemas.Schema; +import org.apache.beam.sdk.transforms.GroupByKey; +import org.apache.beam.sdk.transforms.PTransform; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.transforms.windowing.AfterPane; +import org.apache.beam.sdk.transforms.windowing.AfterWatermark; +import org.apache.beam.sdk.transforms.windowing.DefaultTrigger; +import org.apache.beam.sdk.transforms.windowing.FixedWindows; +import org.apache.beam.sdk.transforms.windowing.GlobalWindows; +import org.apache.beam.sdk.transforms.windowing.Window; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.PCollection; +import org.apache.beam.sdk.values.PCollection.IsBounded; +import org.apache.beam.sdk.values.PCollectionTuple; +import org.apache.beam.sdk.values.PInput; +import org.apache.beam.sdk.values.POutput; +import org.apache.beam.sdk.values.PValue; +import org.apache.beam.sdk.values.Row; +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.collect.ImmutableMap; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.joda.time.Duration; + +/** + * Windows sharded, sort-keyed records into event-time commit windows, grouped by the {@code + * KV<destination, shard>} key. Each {@code (destination, shard, window)} becomes one commit unit + * group and the event-time watermark is the commit barrier. Each group is sorted by the byte sort + * key applied from {@link AssignCdcKeys}: one primary key's records contiguous, in {@code (seq, + * kind)} order within the key. + * + * <p>Late panes are routed to the DLQ (via {@link SplitLateData}) before any ordering happens. This + * is necessary because the downstream commit step skips panes if their window token is present in + * an already committed snapshot. If late panes are let through, their contents will never get to + * the table. + * + * <p>Windowing by input mode: + * + * <ul> + * <li><b>Bounded</b>: a single {@link GlobalWindows} window. One commit per destination, no late + * data possible. + * <li><b>Unbounded</b>: event-time {@link FixedWindows} of {@code triggeringFrequency}, firing on + * the watermark with late firings per element, discarding fired panes. + * </ul> + */ +final class CommitWindows + extends PTransform< + PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>>, CommitWindows.Result> { + + private static final TupleTag<KV<KV<String, Integer>, Iterable<KV<byte[], CdcRecord>>>> + ON_TIME_TAG = new TupleTag<>("onTime"); + private static final TupleTag<Row> DEAD_LETTER_TAG = new TupleTag<>("deadLetter"); + + private final CdcWriteConfig config; + private final @Nullable Duration triggeringFrequency; + private final Duration allowedLateness; + + CommitWindows( + CdcWriteConfig config, @Nullable Duration triggeringFrequency, Duration allowedLateness) { + this.config = config; + this.triggeringFrequency = triggeringFrequency; + this.allowedLateness = allowedLateness; + } + + @Override + public Result expand(PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>> input) { + Schema deadLetterSchema = SplitLateData.deadLetterSchema(dataSchemaOf(input.getCoder())); + + PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>> windowed = applyCommitWindow(input); + + // Exactly one group per (destination, shard, window) + PCollection<KV<KV<String, Integer>, Iterable<KV<byte[], CdcRecord>>>> grouped = + windowed.apply("GroupByShardKey", GroupByKey.create()); + + // Late-data split before sorting anything + PCollectionTuple split = + grouped.apply( + "SplitLateData", + ParDo.of(new SplitLateData(deadLetterSchema, ON_TIME_TAG, DEAD_LETTER_TAG)) + .withOutputTags(ON_TIME_TAG, TupleTagList.of(DEAD_LETTER_TAG))); + PCollection<KV<KV<String, Integer>, Iterable<KV<byte[], CdcRecord>>>> onTimeUnsorted = + split.get(ON_TIME_TAG).setCoder(grouped.getCoder()); + PCollection<Row> deadLetter = + split.get(DEAD_LETTER_TAG).setCoder(RowCoder.of(deadLetterSchema)); + + // Sort each surviving group's records by the byte sort key. The secondary key is byte[] + + // ByteArrayCoder, so SortValues compares the raw CdcSortKey bytes (no coder framing): + // each primary key's records come out contiguous, in (seq, kind) order within the key. + PCollection<KV<KV<String, Integer>, Iterable<KV<byte[], CdcRecord>>>> sorted = + onTimeUnsorted.apply( + "SortBySeqKind", + SortValues.create( + BufferedExternalSorter.options().withMemoryMB(config.getSorterMemoryMB()))); + + return new Result(input.getPipeline(), sorted, deadLetter, deadLetterSchema); + } + + /** Applies the commit-window assignment for the input's boundedness; see the class Javadoc. */ Review Comment: `for the input's boundedness` -> `based on whether input is bounded or not` ? ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/SplitLateData.java: ########## @@ -0,0 +1,102 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg.cdc.sink; + +import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.metrics.Metrics; +import org.apache.beam.sdk.schemas.Schema; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.transforms.windowing.PaneInfo; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.Row; +import org.apache.beam.sdk.values.TupleTag; + +/** + * Splits late data in {@link CommitWindows}: every {@linkplain PaneInfo.Timing#LATE late} pane is + * diverted to a DLQ side output; on-time and early panes pass through unchanged. Review Comment: We can't actually commit late data instead of sending to DLQ ? (probably reasonable but just to double check). ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitWindows.java: ########## @@ -0,0 +1,224 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg.cdc.sink; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; + +import java.util.Map; +import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.coders.KvCoder; +import org.apache.beam.sdk.coders.RowCoder; +import org.apache.beam.sdk.extensions.sorter.BufferedExternalSorter; +import org.apache.beam.sdk.extensions.sorter.SortValues; +import org.apache.beam.sdk.schemas.Schema; +import org.apache.beam.sdk.transforms.GroupByKey; +import org.apache.beam.sdk.transforms.PTransform; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.transforms.windowing.AfterPane; +import org.apache.beam.sdk.transforms.windowing.AfterWatermark; +import org.apache.beam.sdk.transforms.windowing.DefaultTrigger; +import org.apache.beam.sdk.transforms.windowing.FixedWindows; +import org.apache.beam.sdk.transforms.windowing.GlobalWindows; +import org.apache.beam.sdk.transforms.windowing.Window; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.PCollection; +import org.apache.beam.sdk.values.PCollection.IsBounded; +import org.apache.beam.sdk.values.PCollectionTuple; +import org.apache.beam.sdk.values.PInput; +import org.apache.beam.sdk.values.POutput; +import org.apache.beam.sdk.values.PValue; +import org.apache.beam.sdk.values.Row; +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.collect.ImmutableMap; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.joda.time.Duration; + +/** + * Windows sharded, sort-keyed records into event-time commit windows, grouped by the {@code + * KV<destination, shard>} key. Each {@code (destination, shard, window)} becomes one commit unit + * group and the event-time watermark is the commit barrier. Each group is sorted by the byte sort + * key applied from {@link AssignCdcKeys}: one primary key's records contiguous, in {@code (seq, + * kind)} order within the key. + * + * <p>Late panes are routed to the DLQ (via {@link SplitLateData}) before any ordering happens. This + * is necessary because the downstream commit step skips panes if their window token is present in + * an already committed snapshot. If late panes are let through, their contents will never get to + * the table. + * + * <p>Windowing by input mode: + * + * <ul> + * <li><b>Bounded</b>: a single {@link GlobalWindows} window. One commit per destination, no late + * data possible. + * <li><b>Unbounded</b>: event-time {@link FixedWindows} of {@code triggeringFrequency}, firing on + * the watermark with late firings per element, discarding fired panes. + * </ul> + */ +final class CommitWindows + extends PTransform< + PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>>, CommitWindows.Result> { + + private static final TupleTag<KV<KV<String, Integer>, Iterable<KV<byte[], CdcRecord>>>> + ON_TIME_TAG = new TupleTag<>("onTime"); + private static final TupleTag<Row> DEAD_LETTER_TAG = new TupleTag<>("deadLetter"); + + private final CdcWriteConfig config; + private final @Nullable Duration triggeringFrequency; + private final Duration allowedLateness; + + CommitWindows( + CdcWriteConfig config, @Nullable Duration triggeringFrequency, Duration allowedLateness) { + this.config = config; + this.triggeringFrequency = triggeringFrequency; + this.allowedLateness = allowedLateness; + } + + @Override + public Result expand(PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>> input) { + Schema deadLetterSchema = SplitLateData.deadLetterSchema(dataSchemaOf(input.getCoder())); + + PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>> windowed = applyCommitWindow(input); + + // Exactly one group per (destination, shard, window) + PCollection<KV<KV<String, Integer>, Iterable<KV<byte[], CdcRecord>>>> grouped = + windowed.apply("GroupByShardKey", GroupByKey.create()); + + // Late-data split before sorting anything + PCollectionTuple split = + grouped.apply( + "SplitLateData", + ParDo.of(new SplitLateData(deadLetterSchema, ON_TIME_TAG, DEAD_LETTER_TAG)) + .withOutputTags(ON_TIME_TAG, TupleTagList.of(DEAD_LETTER_TAG))); + PCollection<KV<KV<String, Integer>, Iterable<KV<byte[], CdcRecord>>>> onTimeUnsorted = + split.get(ON_TIME_TAG).setCoder(grouped.getCoder()); + PCollection<Row> deadLetter = + split.get(DEAD_LETTER_TAG).setCoder(RowCoder.of(deadLetterSchema)); + + // Sort each surviving group's records by the byte sort key. The secondary key is byte[] + + // ByteArrayCoder, so SortValues compares the raw CdcSortKey bytes (no coder framing): + // each primary key's records come out contiguous, in (seq, kind) order within the key. + PCollection<KV<KV<String, Integer>, Iterable<KV<byte[], CdcRecord>>>> sorted = + onTimeUnsorted.apply( + "SortBySeqKind", + SortValues.create( + BufferedExternalSorter.options().withMemoryMB(config.getSorterMemoryMB()))); + + return new Result(input.getPipeline(), sorted, deadLetter, deadLetterSchema); + } + + /** Applies the commit-window assignment for the input's boundedness; see the class Javadoc. */ + private PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>> applyCommitWindow( + PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>> input) { + if (input.isBounded() == IsBounded.BOUNDED) { + return input.apply( + "GlobalWindows", + Window.<KV<KV<String, Integer>, KV<byte[], CdcRecord>>>into(new GlobalWindows()) + .triggering(DefaultTrigger.of()) + .discardingFiredPanes()); + } + return input.apply( + "EventTimeWindows", + Window.<KV<KV<String, Integer>, KV<byte[], CdcRecord>>>into( + FixedWindows.of( + checkStateNotNull( + triggeringFrequency, + "triggeringFrequency is required for unbounded input"))) + .triggering( + AfterWatermark.pastEndOfWindow().withLateFirings(AfterPane.elementCountAtLeast(1))) + .withAllowedLateness(allowedLateness) + .discardingFiredPanes()); + } + + /** Extracts the CDC data schema carried by the input's nested {@link CdcRecordCoder}. */ + private static Schema dataSchemaOf(Coder<?> inputCoder) { + checkArgument( + inputCoder instanceof KvCoder, + "expected a KvCoder input element coder, got %s", + inputCoder); + Coder<?> valueCoder = ((KvCoder<?, ?>) inputCoder).getValueCoder(); + checkArgument( + valueCoder instanceof KvCoder, "expected a KvCoder input value coder, got %s", valueCoder); + Coder<?> recordCoder = ((KvCoder<?, ?>) valueCoder).getValueCoder(); Review Comment: This kind of casts can be brittel and can be avoided by introducing new types as mentioned in the other comment I believe. ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitWindows.java: ########## @@ -0,0 +1,224 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg.cdc.sink; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; + +import java.util.Map; +import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.coders.KvCoder; +import org.apache.beam.sdk.coders.RowCoder; +import org.apache.beam.sdk.extensions.sorter.BufferedExternalSorter; +import org.apache.beam.sdk.extensions.sorter.SortValues; +import org.apache.beam.sdk.schemas.Schema; +import org.apache.beam.sdk.transforms.GroupByKey; +import org.apache.beam.sdk.transforms.PTransform; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.transforms.windowing.AfterPane; +import org.apache.beam.sdk.transforms.windowing.AfterWatermark; +import org.apache.beam.sdk.transforms.windowing.DefaultTrigger; +import org.apache.beam.sdk.transforms.windowing.FixedWindows; +import org.apache.beam.sdk.transforms.windowing.GlobalWindows; +import org.apache.beam.sdk.transforms.windowing.Window; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.PCollection; +import org.apache.beam.sdk.values.PCollection.IsBounded; +import org.apache.beam.sdk.values.PCollectionTuple; +import org.apache.beam.sdk.values.PInput; +import org.apache.beam.sdk.values.POutput; +import org.apache.beam.sdk.values.PValue; +import org.apache.beam.sdk.values.Row; +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.collect.ImmutableMap; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.joda.time.Duration; + +/** + * Windows sharded, sort-keyed records into event-time commit windows, grouped by the {@code + * KV<destination, shard>} key. Each {@code (destination, shard, window)} becomes one commit unit + * group and the event-time watermark is the commit barrier. Each group is sorted by the byte sort + * key applied from {@link AssignCdcKeys}: one primary key's records contiguous, in {@code (seq, + * kind)} order within the key. + * + * <p>Late panes are routed to the DLQ (via {@link SplitLateData}) before any ordering happens. This + * is necessary because the downstream commit step skips panes if their window token is present in + * an already committed snapshot. If late panes are let through, their contents will never get to + * the table. + * + * <p>Windowing by input mode: + * + * <ul> + * <li><b>Bounded</b>: a single {@link GlobalWindows} window. One commit per destination, no late + * data possible. + * <li><b>Unbounded</b>: event-time {@link FixedWindows} of {@code triggeringFrequency}, firing on + * the watermark with late firings per element, discarding fired panes. + * </ul> + */ +final class CommitWindows + extends PTransform< + PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>>, CommitWindows.Result> { Review Comment: `PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>>` is pretty hard to follow. Shall we introduce new types to simplify this ? For example, `ShardIdentifier ` -> `KV<String, Integer>` SortKeyAndRecord -> `KV<byte[], CdcRecord>` Alternatively we can use Beam Rows or AutoValue classes with schema here. ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitWindows.java: ########## @@ -0,0 +1,224 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg.cdc.sink; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; + +import java.util.Map; +import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.coders.KvCoder; +import org.apache.beam.sdk.coders.RowCoder; +import org.apache.beam.sdk.extensions.sorter.BufferedExternalSorter; +import org.apache.beam.sdk.extensions.sorter.SortValues; +import org.apache.beam.sdk.schemas.Schema; +import org.apache.beam.sdk.transforms.GroupByKey; +import org.apache.beam.sdk.transforms.PTransform; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.transforms.windowing.AfterPane; +import org.apache.beam.sdk.transforms.windowing.AfterWatermark; +import org.apache.beam.sdk.transforms.windowing.DefaultTrigger; +import org.apache.beam.sdk.transforms.windowing.FixedWindows; +import org.apache.beam.sdk.transforms.windowing.GlobalWindows; +import org.apache.beam.sdk.transforms.windowing.Window; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.PCollection; +import org.apache.beam.sdk.values.PCollection.IsBounded; +import org.apache.beam.sdk.values.PCollectionTuple; +import org.apache.beam.sdk.values.PInput; +import org.apache.beam.sdk.values.POutput; +import org.apache.beam.sdk.values.PValue; +import org.apache.beam.sdk.values.Row; +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.collect.ImmutableMap; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.joda.time.Duration; + +/** + * Windows sharded, sort-keyed records into event-time commit windows, grouped by the {@code + * KV<destination, shard>} key. Each {@code (destination, shard, window)} becomes one commit unit + * group and the event-time watermark is the commit barrier. Each group is sorted by the byte sort + * key applied from {@link AssignCdcKeys}: one primary key's records contiguous, in {@code (seq, + * kind)} order within the key. + * + * <p>Late panes are routed to the DLQ (via {@link SplitLateData}) before any ordering happens. This + * is necessary because the downstream commit step skips panes if their window token is present in + * an already committed snapshot. If late panes are let through, their contents will never get to + * the table. + * + * <p>Windowing by input mode: + * + * <ul> + * <li><b>Bounded</b>: a single {@link GlobalWindows} window. One commit per destination, no late + * data possible. + * <li><b>Unbounded</b>: event-time {@link FixedWindows} of {@code triggeringFrequency}, firing on + * the watermark with late firings per element, discarding fired panes. + * </ul> + */ +final class CommitWindows + extends PTransform< + PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>>, CommitWindows.Result> { + + private static final TupleTag<KV<KV<String, Integer>, Iterable<KV<byte[], CdcRecord>>>> + ON_TIME_TAG = new TupleTag<>("onTime"); + private static final TupleTag<Row> DEAD_LETTER_TAG = new TupleTag<>("deadLetter"); + + private final CdcWriteConfig config; + private final @Nullable Duration triggeringFrequency; + private final Duration allowedLateness; + + CommitWindows( + CdcWriteConfig config, @Nullable Duration triggeringFrequency, Duration allowedLateness) { + this.config = config; + this.triggeringFrequency = triggeringFrequency; + this.allowedLateness = allowedLateness; + } + + @Override + public Result expand(PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>> input) { + Schema deadLetterSchema = SplitLateData.deadLetterSchema(dataSchemaOf(input.getCoder())); + + PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>> windowed = applyCommitWindow(input); + + // Exactly one group per (destination, shard, window) + PCollection<KV<KV<String, Integer>, Iterable<KV<byte[], CdcRecord>>>> grouped = + windowed.apply("GroupByShardKey", GroupByKey.create()); + + // Late-data split before sorting anything + PCollectionTuple split = + grouped.apply( + "SplitLateData", + ParDo.of(new SplitLateData(deadLetterSchema, ON_TIME_TAG, DEAD_LETTER_TAG)) + .withOutputTags(ON_TIME_TAG, TupleTagList.of(DEAD_LETTER_TAG))); + PCollection<KV<KV<String, Integer>, Iterable<KV<byte[], CdcRecord>>>> onTimeUnsorted = + split.get(ON_TIME_TAG).setCoder(grouped.getCoder()); + PCollection<Row> deadLetter = + split.get(DEAD_LETTER_TAG).setCoder(RowCoder.of(deadLetterSchema)); + + // Sort each surviving group's records by the byte sort key. The secondary key is byte[] + + // ByteArrayCoder, so SortValues compares the raw CdcSortKey bytes (no coder framing): + // each primary key's records come out contiguous, in (seq, kind) order within the key. + PCollection<KV<KV<String, Integer>, Iterable<KV<byte[], CdcRecord>>>> sorted = + onTimeUnsorted.apply( + "SortBySeqKind", + SortValues.create( + BufferedExternalSorter.options().withMemoryMB(config.getSorterMemoryMB()))); Review Comment: This could lead to OOMs since we are sorting entire windows per shard. Let's make sure there's enough memory for the default case. ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitWindows.java: ########## @@ -0,0 +1,224 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg.cdc.sink; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; + +import java.util.Map; +import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.coders.KvCoder; +import org.apache.beam.sdk.coders.RowCoder; +import org.apache.beam.sdk.extensions.sorter.BufferedExternalSorter; +import org.apache.beam.sdk.extensions.sorter.SortValues; +import org.apache.beam.sdk.schemas.Schema; +import org.apache.beam.sdk.transforms.GroupByKey; +import org.apache.beam.sdk.transforms.PTransform; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.transforms.windowing.AfterPane; +import org.apache.beam.sdk.transforms.windowing.AfterWatermark; +import org.apache.beam.sdk.transforms.windowing.DefaultTrigger; +import org.apache.beam.sdk.transforms.windowing.FixedWindows; +import org.apache.beam.sdk.transforms.windowing.GlobalWindows; +import org.apache.beam.sdk.transforms.windowing.Window; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.PCollection; +import org.apache.beam.sdk.values.PCollection.IsBounded; +import org.apache.beam.sdk.values.PCollectionTuple; +import org.apache.beam.sdk.values.PInput; +import org.apache.beam.sdk.values.POutput; +import org.apache.beam.sdk.values.PValue; +import org.apache.beam.sdk.values.Row; +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.collect.ImmutableMap; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.joda.time.Duration; + +/** + * Windows sharded, sort-keyed records into event-time commit windows, grouped by the {@code + * KV<destination, shard>} key. Each {@code (destination, shard, window)} becomes one commit unit + * group and the event-time watermark is the commit barrier. Each group is sorted by the byte sort + * key applied from {@link AssignCdcKeys}: one primary key's records contiguous, in {@code (seq, + * kind)} order within the key. + * + * <p>Late panes are routed to the DLQ (via {@link SplitLateData}) before any ordering happens. This + * is necessary because the downstream commit step skips panes if their window token is present in + * an already committed snapshot. If late panes are let through, their contents will never get to + * the table. + * + * <p>Windowing by input mode: + * + * <ul> + * <li><b>Bounded</b>: a single {@link GlobalWindows} window. One commit per destination, no late + * data possible. + * <li><b>Unbounded</b>: event-time {@link FixedWindows} of {@code triggeringFrequency}, firing on + * the watermark with late firings per element, discarding fired panes. + * </ul> + */ +final class CommitWindows + extends PTransform< + PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>>, CommitWindows.Result> { + + private static final TupleTag<KV<KV<String, Integer>, Iterable<KV<byte[], CdcRecord>>>> + ON_TIME_TAG = new TupleTag<>("onTime"); + private static final TupleTag<Row> DEAD_LETTER_TAG = new TupleTag<>("deadLetter"); + + private final CdcWriteConfig config; + private final @Nullable Duration triggeringFrequency; + private final Duration allowedLateness; + + CommitWindows( + CdcWriteConfig config, @Nullable Duration triggeringFrequency, Duration allowedLateness) { + this.config = config; + this.triggeringFrequency = triggeringFrequency; + this.allowedLateness = allowedLateness; + } + + @Override + public Result expand(PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>> input) { + Schema deadLetterSchema = SplitLateData.deadLetterSchema(dataSchemaOf(input.getCoder())); + + PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>> windowed = applyCommitWindow(input); + + // Exactly one group per (destination, shard, window) + PCollection<KV<KV<String, Integer>, Iterable<KV<byte[], CdcRecord>>>> grouped = + windowed.apply("GroupByShardKey", GroupByKey.create()); + + // Late-data split before sorting anything + PCollectionTuple split = + grouped.apply( + "SplitLateData", + ParDo.of(new SplitLateData(deadLetterSchema, ON_TIME_TAG, DEAD_LETTER_TAG)) + .withOutputTags(ON_TIME_TAG, TupleTagList.of(DEAD_LETTER_TAG))); + PCollection<KV<KV<String, Integer>, Iterable<KV<byte[], CdcRecord>>>> onTimeUnsorted = + split.get(ON_TIME_TAG).setCoder(grouped.getCoder()); + PCollection<Row> deadLetter = + split.get(DEAD_LETTER_TAG).setCoder(RowCoder.of(deadLetterSchema)); + + // Sort each surviving group's records by the byte sort key. The secondary key is byte[] + + // ByteArrayCoder, so SortValues compares the raw CdcSortKey bytes (no coder framing): + // each primary key's records come out contiguous, in (seq, kind) order within the key. + PCollection<KV<KV<String, Integer>, Iterable<KV<byte[], CdcRecord>>>> sorted = + onTimeUnsorted.apply( + "SortBySeqKind", + SortValues.create( + BufferedExternalSorter.options().withMemoryMB(config.getSorterMemoryMB()))); + + return new Result(input.getPipeline(), sorted, deadLetter, deadLetterSchema); + } + + /** Applies the commit-window assignment for the input's boundedness; see the class Javadoc. */ + private PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>> applyCommitWindow( + PCollection<KV<KV<String, Integer>, KV<byte[], CdcRecord>>> input) { + if (input.isBounded() == IsBounded.BOUNDED) { + return input.apply( + "GlobalWindows", + Window.<KV<KV<String, Integer>, KV<byte[], CdcRecord>>>into(new GlobalWindows()) + .triggering(DefaultTrigger.of()) + .discardingFiredPanes()); + } + return input.apply( + "EventTimeWindows", + Window.<KV<KV<String, Integer>, KV<byte[], CdcRecord>>>into( + FixedWindows.of( + checkStateNotNull( Review Comment: Do this check early ? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
