ahmedabu98 commented on code in PR #38837: URL: https://github.com/apache/beam/pull/38837#discussion_r3707830517
########## 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: A primary key can have identical (delete, insert) pairs in two scenarios: 1. There is a Copy-on-Write, where a new DataFile is added to replace an old DataFile now marked deleted. The new DataFile's rows are identical to the old DataFile, except for some missing rows that are interpreted to be deleted. The overlapping rows, on the surface, are identical (delete, insert) pairs. 2. Some records are inserted and then subsequently deleted in the same snapshot. Reading this produces identical (delete, insert) pairs. Our policy is to only look at the delta so we ignore them ########## 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: Only when a PK group has multiple inserts or deletes. This would be unusual because it would mean one commit inserted or deleted a record multiple times. It's still possible though because Iceberg doesn't enforce uniqueness. We unfortunately can't gather any further insight on intended record ordering within a commit, so pairing can be arbitrary. I've documented the case in the javadoc and now sort both sides by hash before pairing so the result is at least deterministic regardless of input order. ########## 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: Can't really fix this, it's a known shortcoming of Joda which is what Beam DATETIME uses. The workaround is to use the new Timestamp.MICROS logical type (backed by Java time), which has just been made the new default (https://github.com/apache/beam/pull/39344). -- 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]
