kennknowles commented on code in PR #39981: URL: https://github.com/apache/beam/pull/39981#discussion_r3925670340
########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecord.java: ########## @@ -0,0 +1,86 @@ +/* + * 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 java.util.Objects; +import org.apache.beam.sdk.values.Row; +import org.apache.beam.sdk.values.ValueKind; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * One change record carried through the CDC sink's shuffle. + * + * <p>{@link ValueKind} is reified because it's not preserved across a {@code GroupByKey}. + */ +final class CdcRecord { + + private final Row data; + private final ValueKind kind; + private final long sequenceNumber; + + private CdcRecord(Row data, ValueKind kind, long sequenceNumber) { + this.data = data; + this.kind = kind; + this.sequenceNumber = sequenceNumber; + } + + public static CdcRecord of(Row data, ValueKind kind, long sequenceNumber) { + return new CdcRecord(data, kind, sequenceNumber); + } + + public Row getData() { + return data; + } + + public ValueKind getKind() { + return kind; + } + + public long getSequenceNumber() { + return sequenceNumber; + } + + @Override + public boolean equals(@Nullable Object o) { + if (this == o) { + return true; + } + if (!(o instanceof CdcRecord)) { + return false; + } + CdcRecord that = (CdcRecord) o; + return sequenceNumber == that.sequenceNumber && kind == that.kind && data.equals(that.data); + } + + @Override + public int hashCode() { + return Objects.hash(data, kind, sequenceNumber); + } + + @Override + public String toString() { + return "CdcRecord{" Review Comment: ToStringHelper is a nice way to standardize this ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecord.java: ########## @@ -0,0 +1,86 @@ +/* + * 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 java.util.Objects; +import org.apache.beam.sdk.values.Row; +import org.apache.beam.sdk.values.ValueKind; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * One change record carried through the CDC sink's shuffle. + * + * <p>{@link ValueKind} is reified because it's not preserved across a {@code GroupByKey}. Review Comment: Also, once you are "inside" a sink, you don't (necessarily) need the implicit propagation of ValueKind as metadata, since it is more like an explicit field you will write to the sink. ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoder.java: ########## @@ -0,0 +1,130 @@ +/* + * 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 java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.coders.CoderException; +import org.apache.beam.sdk.coders.CustomCoder; +import org.apache.beam.sdk.coders.RowCoder; +import org.apache.beam.sdk.coders.VarIntCoder; +import org.apache.beam.sdk.coders.VarLongCoder; +import org.apache.beam.sdk.schemas.Schema; +import org.apache.beam.sdk.values.Row; +import org.apache.beam.sdk.values.ValueKind; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * {@link CdcRecord} carries a {@link Row} field whose schema is known only at pipeline-construction + * time. We need a custom coder because {@code AutoValueSchema} only infers schemas at the class + * level and cannot infer a dynamic {@link Row} field, so {@code @DefaultSchema} alone cannot + * produce a working coder for {@link CdcRecord}. + */ +final class CdcRecordCoder extends CustomCoder<CdcRecord> { + + private final RowCoder dataCoder; + private final VarIntCoder kindCoder = VarIntCoder.of(); Review Comment: Could probably add a global ValueKindCoder with a registered URN since it will come up a lot. it is just an enum so not sure if this is overkill or what. ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoder.java: ########## @@ -0,0 +1,130 @@ +/* + * 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 java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.coders.CoderException; +import org.apache.beam.sdk.coders.CustomCoder; +import org.apache.beam.sdk.coders.RowCoder; +import org.apache.beam.sdk.coders.VarIntCoder; +import org.apache.beam.sdk.coders.VarLongCoder; +import org.apache.beam.sdk.schemas.Schema; +import org.apache.beam.sdk.values.Row; +import org.apache.beam.sdk.values.ValueKind; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * {@link CdcRecord} carries a {@link Row} field whose schema is known only at pipeline-construction + * time. We need a custom coder because {@code AutoValueSchema} only infers schemas at the class + * level and cannot infer a dynamic {@link Row} field, so {@code @DefaultSchema} alone cannot + * produce a working coder for {@link CdcRecord}. + */ +final class CdcRecordCoder extends CustomCoder<CdcRecord> { Review Comment: Is there any world in which Python will want to directly send these `CdcRecord` things over the wire? Just curious. You would want a `StructuredCoder<CdcRecord>` not a `CustomCoder` which I think translates to `beam:coder:javasdk` with java serialized payload, aka not intelligible to runners or other languages. -- 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]
