chamikaramj commented on code in PR #38837:
URL: https://github.com/apache/beam/pull/38837#discussion_r3678150399
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtils.java:
##########
@@ -70,6 +70,7 @@
*/
public final class CdcReadUtils {
private static final Logger LOG =
LoggerFactory.getLogger(CdcReadUtils.class);
+ static final int COMPRESSED_TO_DECODED_BYTES_ESTIMATE = 4;
Review Comment:
Probably add a small docstring for the non-private constant.
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/CdcResolver.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiConsumer;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.sdk.values.ValueKind;
+import org.apache.iceberg.data.Record;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * Helper class to reconcile CDC rows. Used by {@link ResolveChanges} (with
Beam {@link Row}s) and
+ * {@link LocalResolveDoFn} (with Iceberg {@link Record}s).
+ *
+ * <p>We determine the output ValueKind as follows:
+ *
+ * <ul>
+ * <li>(delete, insert) pairs become {@code UPDATE_BEFORE} + {@code
UPDATE_AFTER}
Review Comment:
(delete, insert) pairs with the same key ?
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/CdcResolver.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiConsumer;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.sdk.values.ValueKind;
+import org.apache.iceberg.data.Record;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * Helper class to reconcile CDC rows. Used by {@link ResolveChanges} (with
Beam {@link Row}s) and
+ * {@link LocalResolveDoFn} (with Iceberg {@link Record}s).
+ *
+ * <p>We determine the output ValueKind as follows:
+ *
+ * <ul>
+ * <li>(delete, insert) pairs become {@code UPDATE_BEFORE} + {@code
UPDATE_AFTER}
+ * <li>singletons remain {@code DELETE} or {@code INSERT}
+ * <li>matching delete+insert with identical non-PK fields are considered
Copy-on-Write side
+ * effects and are dropped
+ * </ul>
+ *
+ * <p>General implementation:
+ *
+ * <ol>
+ * <li>Hash-index inserts by their non-PK field hash, for efficient
Copy-on-Write detection.
+ * <li>Skip matching (delete, insert) pairs with identical non-PK columns. A
CoW operation deletes
+ * and rewrites the whole file (minus some records that are actually
marked for deletion).
+ * Unchanged records are no-ops and should not be mistaken for updates.
+ * <li>Walk the remaining deletes and inserts, emitting matched pairs as
{@link
+ * ValueKind#UPDATE_BEFORE} / {@link ValueKind#UPDATE_AFTER}.
+ * <li>Emit any unmatched extras as {@link ValueKind#DELETE} / {@link
ValueKind#INSERT}.
+ * </ol>
+ */
+abstract class CdcResolver<T> {
+ /** Hashes the non-PK fields of an element. Used as the index for O(n+m) CoW
deduplication. */
+ protected abstract int nonPkHash(T element);
+
+ /**
+ * Returns true if two records (already known to share a PK) share identical
non-PK fields. Called
+ * only when the two elements collide in the {@link #nonPkHash} index, so
the implementation can
+ * stay simple (linear scan of non-PK fields).
+ */
+ protected abstract boolean nonPkEquals(T delete, T insert);
+
+ /**
+ * Resolves a Primary Key group of deletes and inserts. Caller provides
{@code emit} which decides
+ * how to materialize each output.
+ *
+ * <p>Both input lists are inspected in their given order.
+ */
+ final void resolve(List<T> deletes, List<T> inserts, BiConsumer<ValueKind,
T> emit) {
+ boolean hasDeletes = !deletes.isEmpty();
+ boolean hasInserts = !inserts.isEmpty();
+
+ if (hasInserts && hasDeletes) {
+ // First, check if any (delete, insert) pairs are duplicates that should
not be
+ // included in the output
+ boolean[] dupDeletes = new boolean[deletes.size()];
+ boolean[] dupInserts = new boolean[inserts.size()];
+
+ // Map hash to insert-indices
+ Map<Integer, List<Integer>> insertHashToIdx = new HashMap<>();
+ for (int insertIdx = 0; insertIdx < inserts.size(); insertIdx++) {
+ int insertHash = nonPkHash(inserts.get(insertIdx));
+ insertHashToIdx.computeIfAbsent(insertHash, k -> new
ArrayList<>()).add(insertIdx);
+ }
+ for (int deleteIdx = 0; deleteIdx < deletes.size(); deleteIdx++) {
+ int deleteHash = nonPkHash(deletes.get(deleteIdx));
+ @Nullable List<Integer> candidates = insertHashToIdx.get(deleteHash);
+ if (candidates != null) {
+ // check if candidates are just duplicates (e.g. from CoW)
+ for (int idx = 0; idx < candidates.size(); idx++) {
+ int insertIdx = candidates.get(idx);
+ if (!dupInserts[insertIdx]
+ && nonPkEquals(deletes.get(deleteIdx),
inserts.get(insertIdx))) {
+ // this (delete, insert) pair is a duplicate --> should be
skipped
+ dupDeletes[deleteIdx] = true;
+ dupInserts[insertIdx] = true;
+ candidates.remove(idx);
+ break;
+ }
+ }
+ }
+ }
+
+ // Emit matched pairs as UPDATE_BEFORE / UPDATE_AFTER.
+ int d = 0;
+ int i = 0;
+ while (d < deletes.size() && i < inserts.size()) {
+ // skip duplicates
+ while (d < deletes.size() && dupDeletes[d]) {
+ d++;
+ }
+ while (i < inserts.size() && dupInserts[i]) {
+ i++;
+ }
+
+ if (d < deletes.size() && i < inserts.size()) {
+ emit.accept(ValueKind.UPDATE_BEFORE, deletes.get(d));
Review Comment:
Seems like deletes and inserts have to be in a particular order to match
here correctly ?
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/CdcResolver.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiConsumer;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.sdk.values.ValueKind;
+import org.apache.iceberg.data.Record;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * Helper class to reconcile CDC rows. Used by {@link ResolveChanges} (with
Beam {@link Row}s) and
+ * {@link LocalResolveDoFn} (with Iceberg {@link Record}s).
+ *
+ * <p>We determine the output ValueKind as follows:
+ *
+ * <ul>
+ * <li>(delete, insert) pairs become {@code UPDATE_BEFORE} + {@code
UPDATE_AFTER}
+ * <li>singletons remain {@code DELETE} or {@code INSERT}
+ * <li>matching delete+insert with identical non-PK fields are considered
Copy-on-Write side
+ * effects and are dropped
+ * </ul>
+ *
+ * <p>General implementation:
+ *
+ * <ol>
+ * <li>Hash-index inserts by their non-PK field hash, for efficient
Copy-on-Write detection.
+ * <li>Skip matching (delete, insert) pairs with identical non-PK columns. A
CoW operation deletes
+ * and rewrites the whole file (minus some records that are actually
marked for deletion).
+ * Unchanged records are no-ops and should not be mistaken for updates.
Review Comment:
"Unchanged records are no-ops"
Is this a quirk in Iceberg CDC ? Ideally we never produce such records.
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java:
##########
@@ -471,120 +473,150 @@ private static Object getIcebergTimestampValue(Object
beamValue, boolean shouldA
}
}
+ /** Converts a {@link StructLike} to a Beam {@link Row}. */
+ public static Row structToRow(Schema schema, StructLike struct) {
+ checkState(
+ schema.getFieldCount() == struct.size(),
+ "Struct of size %s does not match expected schema size %s",
+ struct.size(),
+ schema.getFieldCount());
+ Row.Builder rowBuilder = Row.withSchema(schema);
+ for (int i = 0; i < schema.getFieldCount(); i++) {
+ Schema.Field field = schema.getField(i);
+ @Nullable Object icebergValue = struct.get(i, Object.class);
+ addIcebergValue(rowBuilder, field, icebergValue);
+ }
+ return rowBuilder.build();
+ }
+
/** Converts an Iceberg {@link Record} to a Beam {@link Row}. */
public static Row icebergRecordToBeamRow(Schema schema, Record record) {
Row.Builder rowBuilder = Row.withSchema(schema);
for (Schema.Field field : schema.getFields()) {
- boolean isNullable = field.getType().getNullable();
@Nullable Object icebergValue = record.getField(field.getName());
- if (icebergValue == null) {
- if (isNullable) {
- rowBuilder.addValue(null);
- continue;
- }
- throw new RuntimeException(
- String.format("Received null value for required field '%s'.",
field.getName()));
+ addIcebergValue(rowBuilder, field, icebergValue);
+ }
+ return rowBuilder.build();
+ }
+
+ private static void addIcebergValue(
+ Row.Builder rowBuilder, Schema.Field field, @Nullable Object
icebergValue) {
+ boolean isNullable = field.getType().getNullable();
+ if (icebergValue == null) {
+ if (isNullable) {
+ rowBuilder.addValue(null);
+ return;
}
- switch (field.getType().getTypeName()) {
- case BYTE:
- case INT16:
- case INT32:
- case INT64:
- case DECIMAL: // Iceberg and Beam both use BigDecimal
- case FLOAT: // Iceberg and Beam both use float
- case DOUBLE: // Iceberg and Beam both use double
- case STRING: // Iceberg and Beam both use String
- case BOOLEAN: // Iceberg and Beam both use boolean
- rowBuilder.addValue(icebergValue);
- break;
- case ARRAY:
- checkState(
- icebergValue instanceof List,
- "Expected List type for field '%s' but received %s",
- field.getName(),
- icebergValue.getClass());
- List<@NonNull ?> beamList = (List<@NonNull ?>) icebergValue;
- Schema.FieldType collectionType =
- checkStateNotNull(field.getType().getCollectionElementType());
- // recurse on struct types
- if (collectionType.getTypeName().isCompositeType()) {
- Schema innerSchema =
checkStateNotNull(collectionType.getRowSchema());
- beamList =
- beamList.stream()
- .map(v -> icebergRecordToBeamRow(innerSchema, (Record) v))
- .collect(Collectors.toList());
- }
- rowBuilder.addValue(beamList);
- break;
- case ITERABLE:
- checkState(
- icebergValue instanceof Iterable,
- "Expected Iterable type for field '%s' but received %s",
- field.getName(),
- icebergValue.getClass());
- Iterable<@NonNull ?> beamIterable = (Iterable<@NonNull ?>)
icebergValue;
- Schema.FieldType iterableCollectionType =
- checkStateNotNull(field.getType().getCollectionElementType());
- // recurse on struct types
- if (iterableCollectionType.getTypeName().isCompositeType()) {
- Schema innerSchema =
checkStateNotNull(iterableCollectionType.getRowSchema());
- ImmutableList.Builder<Row> builder = ImmutableList.builder();
- for (Record v : (Iterable<@NonNull Record>) icebergValue) {
- builder.add(icebergRecordToBeamRow(innerSchema, v));
- }
- beamIterable = builder.build();
+ throw new RuntimeException(
+ String.format("Received null value for required field '%s'.",
field.getName()));
+ }
+ switch (field.getType().getTypeName()) {
+ case BYTE:
+ case INT16:
+ case INT32:
+ case INT64:
+ case DECIMAL: // Iceberg and Beam both use BigDecimal
+ case FLOAT: // Iceberg and Beam both use float
+ case DOUBLE: // Iceberg and Beam both use double
+ case STRING: // Iceberg and Beam both use String
+ case BOOLEAN: // Iceberg and Beam both use boolean
+ rowBuilder.addValue(icebergValue);
+ break;
+ case ARRAY:
+ checkState(
+ icebergValue instanceof List,
+ "Expected List type for field '%s' but received %s",
+ field.getName(),
+ icebergValue.getClass());
+ List<@NonNull ?> beamList = (List<@NonNull ?>) icebergValue;
+ Schema.FieldType collectionType =
+ checkStateNotNull(field.getType().getCollectionElementType());
+ // recurse on struct types
+ if (collectionType.getTypeName().isCompositeType()) {
+ Schema innerSchema =
checkStateNotNull(collectionType.getRowSchema());
+ beamList =
+ beamList.stream()
+ .map(v -> icebergRecordToBeamRow(innerSchema, (Record) v))
+ .collect(Collectors.toList());
+ }
+ rowBuilder.addValue(beamList);
+ break;
+ case ITERABLE:
+ checkState(
+ icebergValue instanceof Iterable,
+ "Expected Iterable type for field '%s' but received %s",
+ field.getName(),
+ icebergValue.getClass());
+ Iterable<@NonNull ?> beamIterable = (Iterable<@NonNull ?>)
icebergValue;
+ Schema.FieldType iterableCollectionType =
+ checkStateNotNull(field.getType().getCollectionElementType());
+ // recurse on struct types
+ if (iterableCollectionType.getTypeName().isCompositeType()) {
+ Schema innerSchema =
checkStateNotNull(iterableCollectionType.getRowSchema());
+ ImmutableList.Builder<Row> builder = ImmutableList.builder();
+ for (Record v : (Iterable<@NonNull Record>) icebergValue) {
+ builder.add(icebergRecordToBeamRow(innerSchema, v));
}
- rowBuilder.addValue(beamIterable);
- break;
- case MAP:
- checkState(
- icebergValue instanceof Map,
- "Expected Map type for field '%s' but received %s",
- field.getName(),
- icebergValue.getClass());
- Map<?, ?> beamMap = (Map<?, ?>) icebergValue;
- Schema.FieldType valueType =
checkStateNotNull(field.getType().getMapValueType());
- // recurse on struct types
- if (valueType.getTypeName().isCompositeType()) {
- Schema innerSchema = checkStateNotNull(valueType.getRowSchema());
- ImmutableMap.Builder<Object, Row> newMap = ImmutableMap.builder();
- for (Map.Entry<?, ?> entry : ((Map<?, ?>)
icebergValue).entrySet()) {
- Record rec = ((Record) entry.getValue());
- newMap.put(
- checkStateNotNull(entry.getKey()),
- icebergRecordToBeamRow(innerSchema, checkStateNotNull(rec)));
- }
- beamMap = newMap.build();
+ beamIterable = builder.build();
+ }
+ rowBuilder.addValue(beamIterable);
+ break;
+ case MAP:
+ checkState(
+ icebergValue instanceof Map,
+ "Expected Map type for field '%s' but received %s",
+ field.getName(),
+ icebergValue.getClass());
+ Map<?, ?> beamMap = (Map<?, ?>) icebergValue;
+ Schema.FieldType valueType =
checkStateNotNull(field.getType().getMapValueType());
+ // recurse on struct types
+ if (valueType.getTypeName().isCompositeType()) {
+ Schema innerSchema = checkStateNotNull(valueType.getRowSchema());
+ ImmutableMap.Builder<Object, Row> newMap = ImmutableMap.builder();
+ for (Map.Entry<?, ?> entry : ((Map<?, ?>) icebergValue).entrySet()) {
+ Record rec = ((Record) entry.getValue());
+ newMap.put(
+ checkStateNotNull(entry.getKey()),
+ icebergRecordToBeamRow(innerSchema, checkStateNotNull(rec)));
}
- rowBuilder.addValue(beamMap);
- break;
- case DATETIME:
- // Iceberg uses a long for micros.
- // Beam DATETIME uses joda's DateTime, which only supports millis,
- // so we do lose some precision here
- rowBuilder.addValue(getBeamDateTimeValue(icebergValue));
- break;
- case BYTES:
- // Iceberg uses ByteBuffer; Beam uses byte[]
- rowBuilder.addValue(((ByteBuffer) icebergValue).array());
- break;
- case ROW:
- Record nestedRecord = (Record) icebergValue;
- Schema nestedSchema =
- checkArgumentNotNull(
- field.getType().getRowSchema(),
- "Corrupted schema: Row type did not have associated nested
schema.");
- rowBuilder.addValue(icebergRecordToBeamRow(nestedSchema,
nestedRecord));
- break;
- case LOGICAL_TYPE:
- rowBuilder.addValue(getLogicalTypeValue(icebergValue,
field.getType()));
- break;
- default:
+ beamMap = newMap.build();
+ }
+ rowBuilder.addValue(beamMap);
+ break;
+ case DATETIME:
+ // Iceberg uses a long for micros.
+ // Beam DATETIME uses joda's DateTime, which only supports millis,
+ // so we do lose some precision here
Review Comment:
Needs a TODO to fix ?
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/OverlapRange.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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;
+
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
+
+import java.util.Comparator;
+import org.apache.beam.sdk.io.iceberg.IcebergScanConfig;
+import org.apache.beam.sdk.io.iceberg.IcebergUtils;
+import org.apache.beam.sdk.io.iceberg.TableCache;
+import org.apache.beam.sdk.values.Row;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.util.StructProjection;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * Primary-key-projection and overlap-range comparison helper.
+ *
+ * <p>Used by {@link LocalResolveDoFn} and {@link ReadFromChangelogs} to
decide whether a record's
Review Comment:
In general, ranges where both sides are inclusive can lead to errors at the
boundary. For example, this is why our source split ranges are non-inclusive at
the upper bound "[...)". I'm not sure if it's an issue here but if we keep
range as is, let's add assertions/tests to make sure that we do not run into
issues at the boundary.
--
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]
