Skip to content

[Iceberg CDC] Add Changelog readers and update resolver #38837

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ahmedabu98 wants to merge 11 commits into
base: master
Choose a base branch
from
Open

[Iceberg CDC] Add Changelog readers and update resolver #38837

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
013caa0
add changelog readers
ahmedabu98 Jun 5, 2026
5519b31
Merge branch 'master' of https://github.com/ahmedabu98/beam into iceb…
ahmedabu98 Jun 25, 2026
3921fbd
Merge branch 'master' of https://github.com/ahmedabu98/beam into iceb…
Jul 12, 2026
d05c37b
trigger ITs
Jul 12, 2026
f9da952
address comments
Jul 12, 2026
1f19913
Merge branch 'master' of https://github.com/ahmedabu98/beam into iceb…
Jul 13, 2026
ce33c3a
use getLength for size estimate; use Locale.English; cdc resolver PK …
Jul 14, 2026
5ae689e
add scaling factor to byte size threshold
Jul 15, 2026
acfb572
Merge branch 'master' of https://github.com/ahmedabu98/beam into iceb…
Jul 15, 2026
f4df841
Merge branch 'master' of https://github.com/ahmedabu98/beam into iceb…
Jul 15, 2026
d75bf69
sync
Jul 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
107 changes: 96 additions & 11 deletions sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergScanConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,27 +19,33 @@

import static org.apache.beam.sdk.io.iceberg.IcebergUtils.icebergSchemaToBeamSchema;
import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull;
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 static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Sets.newHashSet;
import static org.apache.iceberg.types.Type.TypeID.LONG;
import static org.apache.iceberg.types.Type.TypeID.TIMESTAMP;

import com.google.auto.value.AutoValue;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.beam.sdk.io.iceberg.IcebergIO.ReadRows.StartingStrategy;
import org.apache.beam.sdk.io.iceberg.cdc.IcebergCdcMetadataColumns;
import org.apache.beam.sdk.schemas.Schema;
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.MoreObjects;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableUtil;
Expand All @@ -48,6 +54,7 @@
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.types.Comparators;
import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types.NestedField;
import org.apache.iceberg.util.SnapshotUtil;
import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
import org.checkerframework.checker.nullness.qual.Nullable;
Expand Down Expand Up @@ -102,9 +109,9 @@ static org.apache.iceberg.Schema resolveSchema(
@Nullable List<String> keep,
@Nullable List<String> drop,
@Nullable Set<String> fieldsInFilter) {
ImmutableList.Builder<String> selectedFieldsBuilder = ImmutableList.builder();
Set<String> selectedFields = new LinkedHashSet<>();
if (keep != null && !keep.isEmpty()) {
selectedFieldsBuilder.addAll(keep);
selectedFields.addAll(keep);
} else if (drop != null && !drop.isEmpty()) {
List<String> paths = new ArrayList<>(TypeUtil.indexNameById(schema.asStruct()).values());
Collections.sort(paths);
Expand All @@ -113,7 +120,7 @@ static org.apache.iceberg.Schema resolveSchema(
boolean isParent = i + 1 < paths.size() && paths.get(i + 1).startsWith(path + ".");
boolean isDrop = drop.stream().anyMatch(d -> path.equals(d) || path.startsWith(d + "."));
if (!isParent && !isDrop) {
selectedFieldsBuilder.add(path);
selectedFields.add(path);
}
}
} else {
Expand All @@ -124,9 +131,8 @@ static org.apache.iceberg.Schema resolveSchema(
if (fieldsInFilter != null && !fieldsInFilter.isEmpty()) {
fieldsInFilter.stream()
.map(f -> schema.caseInsensitiveFindField(f).name())
.forEach(selectedFieldsBuilder::add);
.forEach(selectedFields::add);
}
ImmutableList<String> selectedFields = selectedFieldsBuilder.build();
return selectedFields.isEmpty() ? schema : schema.select(selectedFields);
}

Expand Down Expand Up @@ -255,6 +261,12 @@ public Expression getFilter() {
@Pure
public abstract @Nullable List<String> getDropFields();

@Pure
public abstract @Nullable String getWatermarkColumn();

@Pure
public abstract @Nullable String getWatermarkColumnTimeUnit();

@Pure
public abstract @Nullable Duration getMaxSnapshotDiscoveryDelay();

Expand Down Expand Up @@ -284,6 +296,7 @@ public static Builder builder() {
.setStartingStrategy(null)
.setTag(null)
.setBranch(null)
.setWatermarkColumn(null)
.setMetadataColumns(ImmutableList.of());
}

Expand Down Expand Up @@ -347,6 +360,10 @@ public Builder setTableIdentifier(String... names) {

public abstract Builder setDropFields(@Nullable List<String> fields);

public abstract Builder setWatermarkColumn(@Nullable String watermarkColumn);

public abstract Builder setWatermarkColumnTimeUnit(@Nullable String timeUnit);

public abstract Builder setMaxSnapshotDiscoveryDelay(@Nullable Duration delay);

public abstract Builder setMetadataColumns(List<String> metadataColumns);
Expand All @@ -357,6 +374,7 @@ public Builder setTableIdentifier(String... names) {
@VisibleForTesting
abstract Builder toBuilder();

@SuppressWarnings("ReturnValueIgnored")
void validate(Table table) {
@Nullable List<String> keep = getKeepFields();
@Nullable List<String> drop = getDropFields();
Expand All @@ -368,16 +386,19 @@ void validate(Table table) {
String param;
if (keep != null) {
param = "keep";
fieldsSpecified = newHashSet(checkNotNull(keep));
fieldsSpecified = newHashSet(checkArgumentNotNull(keep));
} else { // drop != null
param = "drop";
fieldsSpecified = newHashSet(checkNotNull(drop));
fieldsSpecified = newHashSet(checkArgumentNotNull(drop));
}
fieldsSpecified.removeIf(name -> table.schema().findField(name) != null);

checkArgument(
fieldsSpecified.isEmpty(),
error(String.format("'%s' specifies unknown field(s): %s", param, fieldsSpecified)));
fieldsSpecified.isEmpty()
|| fieldsSpecified.stream().allMatch(MetadataColumns::isMetadataColumn),
error("'%s' specifies unknown field(s): %s"),
param,
fieldsSpecified);
}

// TODO(#34168, ahmedabu98): fill these gaps for the existing batch source
Expand Down Expand Up @@ -441,7 +462,6 @@ void validate(Table table) {
checkArgument(
getToTimestamp() == null || getToSnapshot() == null,
error("only one of 'to_timestamp' or 'to_snapshot' can be set"));

@Nullable Long fromSnapshotId = ReadUtils.getFromSnapshotInclusive(table, this);
@Nullable Long toSnapshotId = ReadUtils.getToSnapshot(table, this);
if (fromSnapshotId != null) {
Expand All @@ -464,11 +484,76 @@ void validate(Table table) {
toSnapshotId);
}

if (fromSnapshotId != null) {
checkArgumentNotNull(
table.snapshot(fromSnapshotId),
error("configured starting snapshot does not exist: '%s'"),
fromSnapshotId);
}
if (toSnapshotId != null) {
checkArgumentNotNull(
table.snapshot(toSnapshotId),
error("configured end snapshot does not exist: '%s'"),
toSnapshotId);
}
if (fromSnapshotId != null && toSnapshotId != null) {
checkArgument(
SnapshotUtil.isAncestorOf(table, toSnapshotId, fromSnapshotId),
error("fromSnapshot '%s' is not an ancestor of toSnapshot '%s'"),
fromSnapshotId,
toSnapshotId);
}

if (getPollInterval() != null) {
checkArgument(
Boolean.TRUE.equals(getStreaming()),
error("'poll_interval_seconds' can only be set when streaming is true"));
}

@Nullable String watermarkColumn = getWatermarkColumn();
if (watermarkColumn != null) {
checkArgument(getUseCdc(), error("'watermark_column' is only supported in CDC mode"));
NestedField field = table.schema().findField(watermarkColumn);
checkArgument(
field != null, error("'watermark_column' refers to unknown column: %s"), watermarkColumn);
checkArgument(
field.isRequired(),
error("'watermark_column' needs to be a non-nullable column: %s"),
watermarkColumn);
checkArgument(
field.type().typeId() == TIMESTAMP || field.type().typeId() == LONG,
error("'watermark_column' must be a timestamp-typed column, but '%s' has type %s"),
watermarkColumn,
field.type().typeId());
checkArgumentNotNull(
getProjectedSchema().findField(watermarkColumn),
"'watermark_column' column should not be dropped.");
}

@Nullable String watermarkColumnTimeUnit = getWatermarkColumnTimeUnit();
if (watermarkColumnTimeUnit != null) {
checkArgument(
table
.schema()
.findField(
checkStateNotNull(
watermarkColumn,
"watermark_column_time_unit is configured without a specified watermark_column"))
.type()
.typeId()
== LONG,
error("watermark_column_time_unit is only applicable for LONG columns."));
try {
TimeUnit.valueOf(watermarkColumnTimeUnit.toUpperCase(Locale.ENGLISH));
} catch (IllegalArgumentException e) {
throw new IllegalArgumentException(
error(
String.format(
"watermark_column_time_unit '%s' is invalid. Please choose one of: %s",
watermarkColumnTimeUnit, Arrays.toString(TimeUnit.values()))),
e);
}
}
}

private void validateMetadataColumns(Table table) {
Expand Down
Loading
Loading