diff --git a/sdks/java/io/iceberg/build.gradle b/sdks/java/io/iceberg/build.gradle index e2e8a12d01eb..7340d1eeb411 100644 --- a/sdks/java/io/iceberg/build.gradle +++ b/sdks/java/io/iceberg/build.gradle @@ -49,6 +49,7 @@ dependencies { implementation library.java.avro implementation library.java.slf4j_api implementation library.java.joda_time + implementation library.java.guava implementation "org.apache.parquet:parquet-column:$parquet_version" implementation "org.apache.parquet:parquet-hadoop:$parquet_version" implementation "org.apache.parquet:parquet-common:$parquet_version" diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecord.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecord.java new file mode 100644 index 000000000000..1aa82914fde9 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecord.java @@ -0,0 +1,84 @@ +/* + * 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 com.google.common.base.MoreObjects; +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. + * + *

{@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 MoreObjects.toStringHelper(CdcRecord.class) + .add("data", data) + .add("kind", kind) + .add("sequenceNumber", sequenceNumber) + .toString(); + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoder.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoder.java new file mode 100644 index 000000000000..d477fe100509 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoder.java @@ -0,0 +1,99 @@ +/* + * 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.CustomCoder; +import org.apache.beam.sdk.coders.RowCoder; +import org.apache.beam.sdk.coders.ValueKindCoder; +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 { + + private final RowCoder dataCoder; + private final ValueKindCoder kindCoder = ValueKindCoder.of(); + private final VarLongCoder seqCoder = VarLongCoder.of(); + + private CdcRecordCoder(Schema dataSchema) { + this.dataCoder = RowCoder.of(dataSchema); + } + + public static CdcRecordCoder of(Schema dataSchema) { + return new CdcRecordCoder(dataSchema); + } + + public Schema getDataSchema() { + return dataCoder.getSchema(); + } + + @Override + public void encode(CdcRecord value, OutputStream outStream) throws IOException { + dataCoder.encode(value.getData(), outStream); + kindCoder.encode(value.getKind(), outStream); + seqCoder.encode(value.getSequenceNumber(), outStream); + } + + @Override + public CdcRecord decode(InputStream inStream) throws IOException { + Row data = dataCoder.decode(inStream); + ValueKind kind = kindCoder.decode(inStream); + long seq = seqCoder.decode(inStream); + return CdcRecord.of(data, kind, seq); + } + + @Override + public void verifyDeterministic() throws NonDeterministicException { + Coder.verifyDeterministic(this, "Data coder must be deterministic", dataCoder); + } + + @Override + public boolean consistentWithEquals() { + // decode() rebuilds the row with this coder's schema object, which may differ from the + // original row's; false is always safe. + return false; + } + + @Override + public boolean equals(@Nullable Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + return getDataSchema().equals(((CdcRecordCoder) o).getDataSchema()); + } + + @Override + public int hashCode() { + return getDataSchema().hashCode(); + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSortKey.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSortKey.java new file mode 100644 index 000000000000..21a297b38240 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSortKey.java @@ -0,0 +1,77 @@ +/* + * 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.nio.ByteBuffer; +import java.util.Arrays; +import org.apache.beam.sdk.values.ValueKind; + +/** + * Builds the byte-comparable secondary sort key used by {@code SortValues} (extensions/sorter) to + * order each (destination, shard, window) group: one primary key's records come out contiguous, + * ordered by sequence number then {@link #kindRank(ValueKind)} within the key. + * + *

The key is {@code [pkLen:4][pkBytes][seq ^ Long.MIN_VALUE:8][kindRank:1]}, big-endian. + */ +final class CdcSortKey { + + private CdcSortKey() {} + + /** Ranks change kinds so before-images sort before after-images at an equal {@code seq}. */ + public static byte kindRank(ValueKind kind) { + switch (kind) { + case UPDATE_BEFORE: + return 0; + case DELETE: + return 1; + case UPDATE_AFTER: + return 2; + case INSERT: + return 3; + default: + throw new IllegalArgumentException("Unknown ValueKind: " + kind); + } + } + + /** + * Encodes the deterministic, byte-comparable sort key {@code [pkLen:4][pkBytes][seq ^ + * Long.MIN_VALUE:8][kindRank:1]} for one CDC record. + * + *

SortValues compares unsigned lexicographic byte order. The length prefix is needed to + * accurately compare two primary keys of varying byte-lengths. Flipping the sequence number's + * sign bit makes unsigned byte order match signed numeric order. kindRank breaks equal-seq ties. + */ + public static byte[] encode(byte[] pkBytes, long seq, ValueKind kind) { + return ByteBuffer.allocate(4 + pkBytes.length + 9) + .putInt(pkBytes.length) + .put(pkBytes) + .putLong(seq ^ Long.MIN_VALUE) + .put(kindRank(kind)) + .array(); + } + + /** + * Whether two encoded sort keys carry the same primary key, compared on the raw {@code + * [pkLen:4][pkBytes]} prefix. + */ + public static boolean samePk(byte[] a, byte[] b) { + int aPkEnd = 4 + ByteBuffer.wrap(a).getInt(0); + int bPkEnd = 4 + ByteBuffer.wrap(b).getInt(0); + return Arrays.equals(a, 0, aPkEnd, b, 0, bPkEnd); + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/package-info.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/package-info.java new file mode 100644 index 000000000000..2bcd8b5f6222 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/package-info.java @@ -0,0 +1,23 @@ +/* + * 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. + */ + +/** + * A change data capture (CDC) write sink for Apache Iceberg: applies inserts, updates, and deletes + * to a table, identifying rows by equality columns. + */ +package org.apache.beam.sdk.io.iceberg.cdc.sink; diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoderTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoderTest.java new file mode 100644 index 000000000000..a53b98a10aa9 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoderTest.java @@ -0,0 +1,113 @@ +/* + * 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.hamcrest.MatcherAssert.assertThat; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import org.apache.beam.sdk.coders.CoderException; +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.testing.CoderProperties; +import org.apache.beam.sdk.values.Row; +import org.apache.beam.sdk.values.ValueKind; +import org.hamcrest.Matchers; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Tests for {@link CdcRecordCoder}. */ +@RunWith(JUnit4.class) +public class CdcRecordCoderTest { + + private static final Schema DATA_SCHEMA = + Schema.builder().addInt32Field("id").addStringField("name").addStringField("data").build(); + + @Test + public void roundTripsAllValueKinds() throws Exception { + for (ValueKind kind : ValueKind.values()) { + CdcRecord record = + CdcRecord.of(Row.withSchema(DATA_SCHEMA).addValues(1, "a", "x").build(), kind, 12L); + CoderProperties.coderDecodeEncodeEqual(CdcRecordCoder.of(DATA_SCHEMA), record); + } + } + + /** The coder's own properties: deterministic, serializable, schema-keyed equality. */ + @Test + public void coderIsDeterministicSerializableAndSchemaKeyed() throws Exception { + CdcRecordCoder.of(DATA_SCHEMA).verifyDeterministic(); + CoderProperties.coderSerializable(CdcRecordCoder.of(DATA_SCHEMA)); + // decode() rebuilds the row with the coder's schema object, which may differ from the + // original row's, so the coder must not claim consistency with equals. + assertFalse(CdcRecordCoder.of(DATA_SCHEMA).consistentWithEquals()); + assertThat(CdcRecordCoder.of(DATA_SCHEMA).getDataSchema(), Matchers.equalTo(DATA_SCHEMA)); + + Schema otherSchema = Schema.builder().addInt32Field("other").build(); + assertThat(CdcRecordCoder.of(DATA_SCHEMA), Matchers.equalTo(CdcRecordCoder.of(DATA_SCHEMA))); + assertThat( + CdcRecordCoder.of(DATA_SCHEMA).hashCode(), + Matchers.equalTo(CdcRecordCoder.of(DATA_SCHEMA).hashCode())); + assertThat( + CdcRecordCoder.of(DATA_SCHEMA), + Matchers.not(Matchers.equalTo(CdcRecordCoder.of(otherSchema)))); + } + + @Test + public void encodePinnedWireMapping() throws Exception { + // The stream layout is [row bytes][kind VarInt][seq VarLong]. Codes 0-3 each fit in a single + // VarInt byte equal to the code itself, so the byte immediately following the row bytes must + // be the pinned code for that ValueKind. + Row row = Row.withSchema(DATA_SCHEMA).addValues(1, "a", "x").build(); + ByteArrayOutputStream rowOnly = new ByteArrayOutputStream(); + RowCoder.of(DATA_SCHEMA).encode(row, rowOnly); + int rowLen = rowOnly.toByteArray().length; + + assertPinnedKindCode(row, ValueKind.INSERT, rowLen, 0); + assertPinnedKindCode(row, ValueKind.UPDATE_BEFORE, rowLen, 1); + assertPinnedKindCode(row, ValueKind.UPDATE_AFTER, rowLen, 2); + assertPinnedKindCode(row, ValueKind.DELETE, rowLen, 3); + } + + private static void assertPinnedKindCode(Row row, ValueKind kind, int rowLen, int expectedCode) + throws Exception { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + CdcRecordCoder.of(DATA_SCHEMA).encode(CdcRecord.of(row, kind, 1L), out); + byte[] bytes = out.toByteArray(); + assertThat(bytes[rowLen] & 0xFF, Matchers.equalTo(expectedCode)); + } + + @Test + public void decodeRejectsUnknownKindCode() throws Exception { + // Hand-encode a stream with a valid data row and seq, but a ValueKind code (4) at the pinned + // mapping's boundary: one past the highest valid code (3). + ByteArrayOutputStream out = new ByteArrayOutputStream(); + RowCoder.of(DATA_SCHEMA) + .encode(Row.withSchema(DATA_SCHEMA).addValues(1, "a", "x").build(), out); + VarIntCoder.of().encode(4, out); + VarLongCoder.of().encode(1L, out); + + CdcRecordCoder coder = CdcRecordCoder.of(DATA_SCHEMA); + assertThrows( + CoderException.class, () -> coder.decode(new ByteArrayInputStream(out.toByteArray()))); + } +} diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSortKeyTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSortKeyTest.java new file mode 100644 index 000000000000..6e13f3858261 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSortKeyTest.java @@ -0,0 +1,230 @@ +/* + * 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.hamcrest.MatcherAssert.assertThat; +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import org.apache.beam.sdk.values.ValueKind; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.UnsignedBytes; +import org.hamcrest.Matchers; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Tests for {@link CdcSortKey}. */ +@RunWith(JUnit4.class) +public class CdcSortKeyTest { + + // Ascending sequence numbers, and ValueKinds in ascending kindRank order. + private static final long[] SEQS_ASCENDING = {-5L, -1L, 0L, 1L, 5L}; + private static final ValueKind[] KINDS_BY_RANK_ASCENDING = { + ValueKind.UPDATE_BEFORE, ValueKind.DELETE, ValueKind.UPDATE_AFTER, ValueKind.INSERT + }; + + // Primary keys of assorted lengths and leading bytes. + private static final byte[][] PKS = {{0x02}, {0x01, 0x7F}, {(byte) 0xF0, 0x00, 0x10}}; + + @Test + public void kindRankOrdersBeforeImagesFirst() { + assertThat(CdcSortKey.kindRank(ValueKind.UPDATE_BEFORE), Matchers.equalTo((byte) 0)); + assertThat(CdcSortKey.kindRank(ValueKind.DELETE), Matchers.equalTo((byte) 1)); + assertThat(CdcSortKey.kindRank(ValueKind.UPDATE_AFTER), Matchers.equalTo((byte) 2)); + assertThat(CdcSortKey.kindRank(ValueKind.INSERT), Matchers.equalTo((byte) 3)); + } + + /** + * The sorted keys must group each primary key's entries into one unbroken run, ordered by (seq, + * kind) within the run. Which primary key's run comes first is NOT part of the contract and is + * deliberately unasserted. + */ + @Test + public void sortGroupsEachPkContiguouslyOrderedBySeqThenKindWithin() { + // Per pk, its keys in expected within-key order: seq ascending, kindRank breaking ties. + List> expectedByPk = new ArrayList<>(); + for (byte[] pk : PKS) { + List expected = new ArrayList<>(); + for (long seq : SEQS_ASCENDING) { + for (ValueKind kind : KINDS_BY_RANK_ASCENDING) { + expected.add(CdcSortKey.encode(pk, seq, kind)); + } + } + expectedByPk.add(expected); + } + + // Feed the sorter the same keys interleaved across pks (the shape a shard group arrives in); + // the sort must both regroup and reorder. + List sorted = new ArrayList<>(); + for (int i = 0; i < expectedByPk.get(0).size(); i++) { + for (List keys : expectedByPk) { + sorted.add(keys.get(i)); + } + } + sorted.sort(UnsignedBytes.lexicographicalComparator()); + + // Contiguity: each pk appears in exactly one run. + Set seenRuns = new HashSet<>(); + int previous = -1; + for (byte[] key : sorted) { + int pkIndex = pkIndexOf(key); + if (pkIndex != previous) { + assertTrue("entries of pk " + pkIndex + " are split across runs", seenRuns.add(pkIndex)); + previous = pkIndex; + } + } + + // Within each run, the (seq, kind) order pinned above. + for (int pkIndex = 0; pkIndex < PKS.length; pkIndex++) { + List run = new ArrayList<>(); + for (byte[] key : sorted) { + if (pkIndexOf(key) == pkIndex) { + run.add(key); + } + } + List expected = expectedByPk.get(pkIndex); + assertThat(run, Matchers.hasSize(expected.size())); + for (int i = 0; i < expected.size(); i++) { + assertArrayEquals(expected.get(i), run.get(i)); + } + } + } + + /** + * Two primary keys where one's bytes are a strict prefix of the other's, with sequence numbers + * chosen so the unprefixed layout {@code [pkBytes][seq ^ Long.MIN_VALUE:8][kindRank:1]} WOULD + * interleave them; the length prefix must keep the short key's entries adjacent. + */ + @Test + public void lengthPrefixKeepsAPrefixPkContiguous() { + byte[] shortPk = {0x0A}; + byte[] longPk = {0x0A, (byte) 0x80}; + byte[] shortLow = CdcSortKey.encode(shortPk, 0L, ValueKind.INSERT); + byte[] shortHigh = CdcSortKey.encode(shortPk, Long.MAX_VALUE, ValueKind.INSERT); + byte[] longMid = CdcSortKey.encode(longPk, 0L, ValueKind.INSERT); + + // Fixture self-check: with the length prefix stripped, the long pk's entry lands BETWEEN the + // short pk's two entries (its second byte 0x80 ties the flipped seq 0 and loses to the + // flipped Long.MAX_VALUE). + List naive = + new ArrayList<>(Arrays.asList(stripLengthPrefix(shortHigh), stripLengthPrefix(longMid))); + naive.add(stripLengthPrefix(shortLow)); + naive.sort(UnsignedBytes.lexicographicalComparator()); + assertArrayEquals(stripLengthPrefix(shortLow), naive.get(0)); + assertArrayEquals(stripLengthPrefix(longMid), naive.get(1)); + assertArrayEquals(stripLengthPrefix(shortHigh), naive.get(2)); + + // The real keys: the short pk's entries stay adjacent, low seq first. + List sorted = new ArrayList<>(Arrays.asList(shortHigh, longMid, shortLow)); + sorted.sort(UnsignedBytes.lexicographicalComparator()); + int low = indexOfKey(sorted, shortLow); + int high = indexOfKey(sorted, shortHigh); + assertThat(high, Matchers.equalTo(low + 1)); + } + + @Test + public void encodeOrdersAcrossLongExtremes() { + byte[] pk = {0x01}; + byte[] min = CdcSortKey.encode(pk, Long.MIN_VALUE, ValueKind.INSERT); + byte[] negOne = CdcSortKey.encode(pk, -1L, ValueKind.INSERT); + byte[] zero = CdcSortKey.encode(pk, 0L, ValueKind.INSERT); + byte[] one = CdcSortKey.encode(pk, 1L, ValueKind.INSERT); + byte[] max = CdcSortKey.encode(pk, Long.MAX_VALUE, ValueKind.INSERT); + + List sorted = new ArrayList<>(Arrays.asList(max, zero, min, one, negOne)); + sorted.sort(UnsignedBytes.lexicographicalComparator()); + + assertArrayEquals(min, sorted.get(0)); + assertArrayEquals(negOne, sorted.get(1)); + assertArrayEquals(zero, sorted.get(2)); + assertArrayEquals(one, sorted.get(3)); + assertArrayEquals(max, sorted.get(4)); + } + + /** + * The encoding is a frozen wire format: an in-place pipeline update replays in-flight groups + * through it, so changing any of these bytes silently re-orders live data. Update only with a + * migration story. + */ + @Test + public void encodePinnedByteLayout() { + // pk {0x01, 0x02} -> pkLen 2; seq 5 -> flippedSeq 0x8000000000000005; INSERT -> kindRank 3. + assertArrayEquals( + new byte[] {0, 0, 0, 2, 1, 2, (byte) 0x80, 0, 0, 0, 0, 0, 0, 5, 3}, + CdcSortKey.encode(new byte[] {1, 2}, 5L, ValueKind.INSERT)); + // pk {0xAB} -> pkLen 1; seq 0 -> flippedSeq 0x8000000000000000; UPDATE_BEFORE -> kindRank 0. + assertArrayEquals( + new byte[] {0, 0, 0, 1, (byte) 0xAB, (byte) 0x80, 0, 0, 0, 0, 0, 0, 0, 0}, + CdcSortKey.encode(new byte[] {(byte) 0xAB}, 0L, ValueKind.UPDATE_BEFORE)); + } + + /** + * {@code samePk} is the writer's block-boundary test: equal for any two keys of one pk whatever + * their seq/kind bytes, unequal across pks, including the prefix-pk pair whose seq bytes tie. + */ + @Test + public void samePkComparesOnlyThePkPrefix() { + byte[] pk = {0x0A}; + assertTrue( + CdcSortKey.samePk( + CdcSortKey.encode(pk, Long.MIN_VALUE, ValueKind.UPDATE_BEFORE), + CdcSortKey.encode(pk, Long.MAX_VALUE, ValueKind.INSERT))); + assertFalse( + CdcSortKey.samePk( + CdcSortKey.encode(new byte[] {0x01}, 5L, ValueKind.INSERT), + CdcSortKey.encode(new byte[] {0x02}, 5L, ValueKind.INSERT))); + // A strict-prefix pk with seq bytes continuing the longer pk's bytes must still differ. + assertFalse( + CdcSortKey.samePk( + CdcSortKey.encode(pk, 0L, ValueKind.INSERT), + CdcSortKey.encode(new byte[] {0x0A, (byte) 0x80}, 0L, ValueKind.INSERT))); + } + + /** The index in {@link #PKS} of the pk carried in {@code key}'s length-prefixed prefix. */ + private static int pkIndexOf(byte[] key) { + int pkLen = ByteBuffer.wrap(key).getInt(); + byte[] pk = Arrays.copyOfRange(key, 4, 4 + pkLen); + for (int i = 0; i < PKS.length; i++) { + if (Arrays.equals(PKS[i], pk)) { + return i; + } + } + throw new AssertionError("unknown pk " + Arrays.toString(pk)); + } + + /** Drops the 4-byte length prefix, leaving the naive {@code [pkBytes][seq][kind]} layout. */ + private static byte[] stripLengthPrefix(byte[] key) { + return Arrays.copyOfRange(key, 4, key.length); + } + + private static int indexOfKey(List keys, byte[] key) { + for (int i = 0; i < keys.size(); i++) { + if (Arrays.equals(keys.get(i), key)) { + return i; + } + } + throw new AssertionError("key not found: " + Arrays.toString(key)); + } +}