From 020ea0fd5bd681dd591672176d4f3fde3fdbb8fe Mon Sep 17 00:00:00 2001 From: Ran Tao Date: Wed, 28 Oct 2026 16:05:57 +0800 Subject: [PATCH] Debezium JSON CDC corrupts binary field values --- .../flink/action/cdc/ComputedColumn.java | 34 +- .../cdc/format/AbstractRecordParser.java | 19 +- .../debezium/DebeziumJsonRecordParser.java | 16 + .../format/debezium/DebeziumSchemaUtils.java | 12 +- .../action/cdc/mysql/MySqlRecordParser.java | 28 +- .../paimon/flink/sink/cdc/CdcRecord.java | 54 ++- .../paimon/flink/sink/cdc/CdcRecordUtils.java | 29 +- .../debezium/DebeziumBinaryRecordTest.java | 313 ++++++++++++++++++ .../flink/sink/cdc/CdcRecordUtilsTest.java | 156 +++++++++ 9 files changed, 645 insertions(+), 16 deletions(-) create mode 100644 paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumBinaryRecordTest.java create mode 100644 paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordUtilsTest.java diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/ComputedColumn.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/ComputedColumn.java index 2e4fc4e39d8d..cb4072848562 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/ComputedColumn.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/ComputedColumn.java @@ -23,7 +23,11 @@ import javax.annotation.Nullable; import java.io.Serializable; +import java.nio.charset.Charset; +import java.util.Base64; +import java.util.Collections; import java.util.Map; +import java.util.Set; /** * A Computed column's value is computed from input columns. Only expression with at most two inputs @@ -72,24 +76,46 @@ public String eval(@Nullable String input) { */ @Nullable public String evalFromRecord(Map rowData) { - return eval(referencedValue(rowData)); + return evalFromRecord(rowData, Collections.emptySet()); } + /** Evaluate binary source fields as text, preserving the existing expression semantics. */ @Nullable - private String referencedValue(Map rowData) { + public String evalFromRecord(Map rowData, Set base64EncodedFields) { + return eval(referencedValue(rowData, base64EncodedFields)); + } + + @Nullable + private String referencedValue(Map rowData, Set base64EncodedFields) { String reference = fieldReference(); if (reference == null) { return null; } String value = rowData.get(reference); if (caseSensitive || rowData.containsKey(reference)) { - return value; + return decodeBase64IfNeeded(value, base64EncodedFields.contains(reference)); } for (Map.Entry entry : rowData.entrySet()) { if (reference.equalsIgnoreCase(entry.getKey())) { - return entry.getValue(); + return decodeBase64IfNeeded( + entry.getValue(), base64EncodedFields.contains(entry.getKey())); } } return null; } + + /** + * Decodes a marked Base64 binary value to text for expression evaluation. + * + *

Null and unmarked values are returned unchanged. Marked non-null values must contain valid + * Base64. The default charset preserves the previous {@code new String(bytes)} interpretation + * for computed columns; the record retains its Base64 value so the sink can recover the + * original bytes. + */ + @Nullable + private String decodeBase64IfNeeded(@Nullable String value, boolean base64Encoded) { + return value != null && base64Encoded + ? new String(Base64.getDecoder().decode(value), Charset.defaultCharset()) + : value; + } } diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/AbstractRecordParser.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/AbstractRecordParser.java index d5ea8f9d176e..39c20f2ce561 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/AbstractRecordParser.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/AbstractRecordParser.java @@ -35,9 +35,12 @@ import javax.annotation.Nullable; +import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; /** * Provides a base implementation for parsing messages of various formats into {@link @@ -116,9 +119,11 @@ protected boolean isDDL() { /** generate values for computed columns. */ protected void evalComputedColumns( Map rowData, CdcSchema.Builder schemaBuilder) { + Set base64EncodedFields = + computedColumns.isEmpty() ? Collections.emptySet() : extractBase64EncodedFields(); computedColumns.forEach( computedColumn -> { - String result = computedColumn.evalFromRecord(rowData); + String result = computedColumn.evalFromRecord(rowData, base64EncodedFields); rowData.put(computedColumn.columnName(), result); schemaBuilder.column(computedColumn.columnName(), computedColumn.columnType()); @@ -148,11 +153,21 @@ protected void evalMetadataColumns( protected RichCdcMultiplexRecord createRecord( RowKind rowKind, Map data, CdcSchema.Builder schemaBuilder) { schemaBuilder.primaryKey(extractPrimaryKeys()); + Set base64EncodedFields = new HashSet<>(extractBase64EncodedFields()); + computedColumns.forEach(column -> base64EncodedFields.remove(column.columnName())); + for (CdcMetadataConverter converter : metadataConverters) { + base64EncodedFields.remove(converter.columnName()); + } return new RichCdcMultiplexRecord( getDatabaseName(), getTableName(), schemaBuilder.build(), - new CdcRecord(rowKind, data)); + CdcRecord.withBase64EncodedFields(rowKind, data, base64EncodedFields)); + } + + /** Names of source fields whose non-null values contain Base64-encoded binary data. */ + protected Set extractBase64EncodedFields() { + return Collections.emptySet(); } @Nullable diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumJsonRecordParser.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumJsonRecordParser.java index 9551973f35a8..10333c418f4b 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumJsonRecordParser.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumJsonRecordParser.java @@ -42,6 +42,8 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; import static org.apache.paimon.flink.action.cdc.format.debezium.DebeziumSchemaUtils.FIELD_AFTER; import static org.apache.paimon.flink.action.cdc.format.debezium.DebeziumSchemaUtils.FIELD_BEFORE; @@ -218,6 +220,20 @@ protected Map extractRowData(JsonNode record, CdcSchema.Builder return resultMap; } + @Override + protected Set extractBase64EncodedFields() { + if (!hasSchema) { + return Collections.emptySet(); + } + return debeziumTypes.entrySet().stream() + .filter( + entry -> + DebeziumSchemaUtils.isPlainBinaryType( + entry.getValue(), classNames.get(entry.getKey()))) + .map(Map.Entry::getKey) + .collect(Collectors.toSet()); + } + @Override protected List extractPrimaryKeys() { List primaryKeys = super.extractPrimaryKeys(); diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java index 983b760294af..81e122a2269e 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java @@ -136,6 +136,11 @@ public static String transformAvroRawValue( serverTimeZone); } + /** Plain binary values, excluding logical types such as BIT, DECIMAL and geometry. */ + public static boolean isPlainBinaryType(String debeziumType, @Nullable String schemaName) { + return "bytes".equals(debeziumType) && schemaName == null; + } + /** Transform raw string value according to schema. */ public static String transformRawValue( @Nullable String rawValue, @@ -164,9 +169,10 @@ public static String transformRawValue( } else { transformed = Base64.getEncoder().encodeToString(bigEndian); } - } else if (("bytes".equals(debeziumType) && className == null)) { - // MySQL binary, varbinary, blob - transformed = new String(Base64.getDecoder().decode(rawValue)); + } else if (isPlainBinaryType(debeziumType, className)) { + // Keep binary values in Base64 until the sink converts them to bytes. The parser + // marks these fields in CdcRecord, without changing other CDC string conversions. + transformed = rawValue; } else if ("bytes".equals(debeziumType) && className.endsWith(decimalLogicalName())) { // MySQL numeric, fixed, decimal try { diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlRecordParser.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlRecordParser.java index 9c4caee5ef94..bf58b14ace51 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlRecordParser.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlRecordParser.java @@ -278,8 +278,12 @@ private Map extractRow(JsonNode recordRow) { } // generate values of computed columns + Set base64EncodedFields = + computedColumns.isEmpty() ? Collections.emptySet() : extractBase64EncodedFields(); for (ComputedColumn computedColumn : computedColumns) { - resultMap.put(computedColumn.columnName(), computedColumn.evalFromRecord(resultMap)); + resultMap.put( + computedColumn.columnName(), + computedColumn.evalFromRecord(resultMap, base64EncodedFields)); // remember the computed column data type for later reference by other computed columns schemaBuilder.column(computedColumn.columnName(), computedColumn.columnType()); @@ -294,8 +298,28 @@ private Map extractRow(JsonNode recordRow) { return resultMap; } + private Set extractBase64EncodedFields() { + Set fields = new HashSet<>(); + root.schema() + .beforeAndAfterFields() + .forEach( + (name, field) -> { + if (DebeziumSchemaUtils.isPlainBinaryType(field.type(), field.name())) { + fields.add(name); + } + }); + computedColumns.forEach(column -> fields.remove(column.columnName())); + for (CdcMetadataConverter converter : metadataConverters) { + fields.remove(converter.columnName()); + } + return fields; + } + protected RichCdcMultiplexRecord createRecord(RowKind rowKind, Map data) { return new RichCdcMultiplexRecord( - databaseName, currentTable, null, new CdcRecord(rowKind, data)); + databaseName, + currentTable, + null, + CdcRecord.withBase64EncodedFields(rowKind, data, extractBase64EncodedFields())); } } diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecord.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecord.java index 3093ebbfb54c..6a87e6e0fc0d 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecord.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecord.java @@ -24,9 +24,11 @@ import java.io.Serializable; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.Locale; import java.util.Map; import java.util.Objects; +import java.util.Set; /** A data change message from the CDC source. */ @Experimental @@ -44,6 +46,25 @@ public CdcRecord(RowKind kind, Map data) { this.data = data; } + /** + * Creates a record whose specified fields contain Base64-encoded binary values. + * + *

Non-null values of the specified fields must already be Base64-encoded. This method only + * records their encoding; it does not encode or decode values. Unmarked fields retain the usual + * CDC string conversion, including text mapped to VARBINARY. + */ + public static CdcRecord withBase64EncodedFields( + RowKind kind, Map data, Set base64EncodedFields) { + return base64EncodedFields.isEmpty() + ? new CdcRecord(kind, data) + : new BinaryCdcRecord(kind, data, base64EncodedFields); + } + + /** Names of fields whose non-null values contain Base64-encoded binary data. */ + public Set base64EncodedFields() { + return Collections.emptySet(); + } + public static CdcRecord emptyRecord() { return new CdcRecord(RowKind.INSERT, Collections.emptyMap()); } @@ -64,7 +85,11 @@ public CdcRecord fieldNameLowerCase() { // silently null out the column on the record-schema join newData.put(entry.getKey().toLowerCase(Locale.ROOT), entry.getValue()); } - return new CdcRecord(kind, newData); + Set newBase64EncodedFields = new HashSet<>(); + for (String field : base64EncodedFields()) { + newBase64EncodedFields.add(field.toLowerCase(Locale.ROOT)); + } + return withBase64EncodedFields(kind, newData, newBase64EncodedFields); } @Override @@ -74,12 +99,35 @@ public boolean equals(Object o) { } CdcRecord that = (CdcRecord) o; - return Objects.equals(kind, that.kind) && Objects.equals(data, that.data); + return Objects.equals(kind, that.kind) + && Objects.equals(data, that.data) + && Objects.equals(base64EncodedFields(), that.base64EncodedFields()); } @Override public int hashCode() { - return Objects.hash(kind, data); + return Objects.hash(kind, data, base64EncodedFields()); + } + + // Keep the serialized fields of ordinary CdcRecord unchanged. Only records with explicitly + // encoded binary values need the additional metadata. + private static class BinaryCdcRecord extends CdcRecord { + + private static final long serialVersionUID = 1L; + + // Keep the serialized field name unchanged. + private final Set binaryFields; + + private BinaryCdcRecord( + RowKind kind, Map data, Set base64EncodedFields) { + super(kind, data); + this.binaryFields = new HashSet<>(base64EncodedFields); + } + + @Override + public Set base64EncodedFields() { + return binaryFields; + } } @Override diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordUtils.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordUtils.java index 6fc94eb9e535..b9b9900d1ca8 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordUtils.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordUtils.java @@ -21,12 +21,16 @@ import org.apache.paimon.data.GenericRow; import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypeChecks; +import org.apache.paimon.types.DataTypeRoot; import org.apache.paimon.types.RowKind; import org.apache.paimon.utils.TypeUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.nio.charset.Charset; +import java.util.Base64; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -57,7 +61,7 @@ public static GenericRow projectAsInsert(CdcRecord record, List dataF String fieldValue = record.data().get(dataField.name()); if (fieldValue != null) { genericRow.setField( - i, TypeUtils.castFromCdcValueString(fieldValue, dataField.type())); + i, castFieldValue(record, dataField.name(), fieldValue, dataField.type())); } } return genericRow; @@ -101,7 +105,7 @@ public static Optional toGenericRow( DataType type = dataFields.get(idx).type(); try { - genericRow.setField(idx, TypeUtils.castFromCdcValueString(value, type)); + genericRow.setField(idx, castFieldValue(record, key, value, type)); } catch (Exception e) { LOG.info( "Failed to convert field '{}' value {} to type {}. Waiting for schema update.", @@ -115,6 +119,27 @@ public static Optional toGenericRow( return Optional.of(genericRow); } + private static Object castFieldValue( + CdcRecord record, String fieldName, String value, DataType type) { + if (record.base64EncodedFields().contains(fieldName)) { + byte[] bytes = Base64.getDecoder().decode(value); + if (type.getTypeRoot() == DataTypeRoot.BINARY + || type.getTypeRoot() == DataTypeRoot.VARBINARY) { + int length = DataTypeChecks.getLength(type); + if (bytes.length > length) { + throw new IllegalArgumentException( + String.format( + "Length of type %s is %d, but casting result has a length of %d", + type, length, bytes.length)); + } + return bytes; + } + // Preserve the previous text conversion when an existing target column is not binary. + value = new String(bytes, Charset.defaultCharset()); + } + return TypeUtils.castFromCdcValueString(value, type); + } + public static CdcRecord fromGenericRow(GenericRow row, List fieldNames) { Map data = new HashMap<>(); for (int i = 0; i < row.getFieldCount(); i++) { diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumBinaryRecordTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumBinaryRecordTest.java new file mode 100644 index 000000000000..195c2ccdb190 --- /dev/null +++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumBinaryRecordTest.java @@ -0,0 +1,313 @@ +/* + * 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.paimon.flink.action.cdc.format.debezium; + +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.flink.action.cdc.CdcMetadataConverter; +import org.apache.paimon.flink.action.cdc.CdcSourceRecord; +import org.apache.paimon.flink.action.cdc.ComputedColumn; +import org.apache.paimon.flink.action.cdc.Expression; +import org.apache.paimon.flink.action.cdc.TypeMapping; +import org.apache.paimon.flink.action.cdc.mysql.MySqlRecordParser; +import org.apache.paimon.flink.sink.cdc.CdcRecord; +import org.apache.paimon.flink.sink.cdc.CdcRecordUtils; +import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FileStoreTableFactory; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.BatchWriteBuilder; +import org.apache.paimon.table.source.ReadBuilder; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.CloseableIterator; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ArrayNode; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ObjectNode; + +import org.apache.flink.api.common.functions.FlatMapFunction; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.util.Collector; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Base64; +import java.util.Collections; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Binary values must survive Debezium parsing, key projection and table writes. */ +class DebeziumBinaryRecordTest { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + @TempDir java.nio.file.Path tempDir; + + @ParameterizedTest + @ValueSource(strings = {"json", "mysql"}) + void testBinaryValues(String format) throws Exception { + assertBinaryValues((bytes, computed) -> parse(format, bytes, computed)); + } + + @ParameterizedTest + @ValueSource(strings = {"json", "mysql"}) + void testWriteAndReadBinaryValues(String format) throws Exception { + assertWriteAndReadBinaryValues( + (bytes, computed) -> parse(format, bytes, computed), tempDir.resolve(format)); + } + + @ParameterizedTest + @ValueSource(strings = {"json", "mysql"}) + void testComputedColumnsKeepTextSemantics(String format) throws Exception { + assertComputedColumns((bytes, computed) -> parse(format, bytes, computed)); + } + + @Test + void testBinaryAlongsideLogicalTypes() throws Exception { + byte[] bytes = {0, (byte) 0xff, (byte) 0x80, 65}; + ObjectNode value = (ObjectNode) jsonValue(ByteBuffer.wrap(bytes)); + ArrayNode fields = (ArrayNode) value.get("schema").get("fields").get(0).get("fields"); + ObjectNode bits = + fields.addObject() + .put("field", "bits") + .put("type", "bytes") + .put("name", "io.debezium.data.Bits"); + bits.putObject("parameters").put("length", "16"); + ObjectNode price = + fields.addObject() + .put("field", "price") + .put("type", "bytes") + .put("name", "org.apache.kafka.connect.data.Decimal"); + price.putObject("parameters").put("connect.decimal.precision", "10").put("scale", "2"); + fields.addObject() + .put("field", "location") + .put("type", "struct") + .put("name", "io.debezium.data.geometry.Point"); + ObjectNode after = (ObjectNode) value.get("payload").get("after"); + after.put("bits", "Af8="); + after.put("price", "123.45"); + byte[] wkb = + ByteBuffer.allocate(21) + .order(ByteOrder.LITTLE_ENDIAN) + .put((byte) 1) + .putInt(1) + .putDouble(1) + .putDouble(2) + .array(); + after.putObject("location") + .put("wkb", Base64.getEncoder().encodeToString(wkb)) + .put("srid", 0); + DebeziumJsonRecordParser parser = + new DebeziumJsonRecordParser(TypeMapping.defaultMapping(), Collections.emptyList()); + RichCdcMultiplexRecord rich = collectRecord(parser, value); + CdcRecord record = rich.toRichCdcRecord().toCdcRecord(); + assertThat(record.base64EncodedFields()).containsExactly("payload"); + GenericRow row = + CdcRecordUtils.toGenericRow(record, rich.buildSchema().fields(), true).get(); + RowType rowType = rich.buildSchema().rowType(); + assertThat(row.getBinary(rowType.getFieldIndex("payload"))).containsExactly(bytes); + assertThat(row.getBinary(rowType.getFieldIndex("bits"))) + .containsExactly((byte) 0xff, (byte) 1); + assertThat(row.getDecimal(rowType.getFieldIndex("price"), 10, 2).toBigDecimal()) + .isEqualByComparingTo("123.45"); + JsonNode geometry = + MAPPER.readTree(row.getString(rowType.getFieldIndex("location")).toString()); + assertThat(geometry.get("type").asText()).isEqualTo("Point"); + assertThat(geometry.get("coordinates").get(0).asDouble()).isEqualTo(1); + assertThat(geometry.get("coordinates").get(1).asDouble()).isEqualTo(2); + } + + static void assertBinaryValues(RecordParser parser) throws Exception { + for (byte[] bytes : binaryValues()) { + RichCdcMultiplexRecord rich = + parser.parse( + bytes == null ? null : ByteBuffer.wrap(bytes), Collections.emptyList()); + CdcRecord record = rich.toRichCdcRecord().toCdcRecord(); + List fields = + Arrays.asList( + new DataField(0, "payload", DataTypes.BYTES()), + new DataField(1, "id", DataTypes.INT())); + GenericRow row = CdcRecordUtils.toGenericRow(record, fields, true).get(); + GenericRow key = CdcRecordUtils.projectAsInsert(record, fields); + if (bytes == null) { + assertThat(row.isNullAt(0)).isTrue(); + assertThat(key.isNullAt(0)).isTrue(); + } else { + assertThat(row.getBinary(0)).containsExactly(bytes); + assertThat(key.getBinary(0)).containsExactly(bytes); + } + } + } + + static void assertWriteAndReadBinaryValues(RecordParser parser, java.nio.file.Path tableDir) + throws Exception { + Path path = new Path(tableDir.toUri()); + Schema schema = + Schema.newBuilder() + .column("payload", DataTypes.BYTES()) + .column("id", DataTypes.INT()) + .option("bucket", "-1") + .build(); + LocalFileIO fileIO = LocalFileIO.create(); + new FileSystemSchemaManager(fileIO, path).createTable(schema); + FileStoreTable table = FileStoreTableFactory.create(fileIO, path); + List values = binaryValues(); + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + for (byte[] bytes : values) { + CdcRecord record = + parser.parse( + bytes == null ? null : ByteBuffer.wrap(bytes), + Collections.emptyList()) + .toRichCdcRecord() + .toCdcRecord(); + write.write(CdcRecordUtils.toGenericRow(record, schema.fields(), true).get()); + } + commit.commit(write.prepareCommit()); + } + List expected = new ArrayList<>(); + for (byte[] bytes : values) { + expected.add(bytes == null ? null : Base64.getEncoder().encodeToString(bytes)); + } + List actual = new ArrayList<>(); + ReadBuilder read = table.newReadBuilder(); + try (CloseableIterator rows = + read.newRead().createReader(read.newScan().plan()).toCloseableIterator()) { + while (rows.hasNext()) { + InternalRow row = rows.next(); + actual.add( + row.isNullAt(0) + ? null + : Base64.getEncoder().encodeToString(row.getBinary(0))); + } + } + assertThat(actual).containsExactlyInAnyOrderElementsOf(expected); + } + + static void assertComputedColumns(RecordParser parser) throws Exception { + byte[] bytes = "plain text".getBytes(StandardCharsets.UTF_8); + List computed = + Arrays.asList( + new ComputedColumn( + "prefix", Expression.substring("PAYLOAD", "0", "5"), false), + new ComputedColumn( + "short_prefix", Expression.substring("prefix", "0", "2"), true)); + CdcRecord record = + parser.parse(ByteBuffer.wrap(bytes), computed).toRichCdcRecord().toCdcRecord(); + assertThat(record.data()) + .containsEntry("prefix", "plain") + .containsEntry("short_prefix", "pl"); + assertThat( + CdcRecordUtils.projectAsInsert( + record, + Collections.singletonList( + new DataField(0, "payload", DataTypes.BYTES()))) + .getBinary(0)) + .containsExactly(bytes); + } + + private static List binaryValues() { + byte[] allBytes = new byte[256]; + for (int i = 0; i < allBytes.length; i++) { + allBytes[i] = (byte) i; + } + return Arrays.asList( + null, + new byte[0], + "plain text".getBytes(StandardCharsets.UTF_8), + "中文😀".getBytes(StandardCharsets.UTF_8), + new byte[] {0, (byte) 0xff, (byte) 0x80, 65}, + allBytes); + } + + private static RichCdcMultiplexRecord parse( + String format, ByteBuffer bytes, List computed) throws Exception { + if (format.equals("mysql")) { + return collectRecord( + new MySqlRecordParser( + new Configuration(), + computed, + TypeMapping.defaultMapping(), + new CdcMetadataConverter[0]), + jsonValue(bytes).toString()); + } + return collectRecord( + new DebeziumJsonRecordParser(TypeMapping.defaultMapping(), computed), + jsonValue(bytes)); + } + + @FunctionalInterface + interface RecordParser { + RichCdcMultiplexRecord parse(ByteBuffer bytes, List computed) + throws Exception; + } + + static RichCdcMultiplexRecord collectRecord( + FlatMapFunction parser, Object value) + throws Exception { + List records = new ArrayList<>(); + parser.flatMap( + new CdcSourceRecord("test", null, value), + new Collector() { + @Override + public void collect(RichCdcMultiplexRecord record) { + records.add(record); + } + + @Override + public void close() {} + }); + assertThat(records).hasSize(1); + return records.get(0); + } + + static JsonNode jsonValue(ByteBuffer buffer) throws Exception { + ObjectNode value = + (ObjectNode) + MAPPER.readTree( + "{\"schema\":{\"fields\":[{\"field\":\"after\",\"fields\":[{\"field\":\"payload\",\"type\":\"bytes\",\"optional\":true},{\"field\":\"id\",\"type\":\"int32\",\"optional\":false}]}]},\"payload\":{\"before\":null,\"after\":{},\"source\":{\"db\":\"test\",\"table\":\"binary_table\"},\"op\":\"c\"}}"); + ObjectNode after = (ObjectNode) value.get("payload").get("after"); + after.put("id", 1); + if (buffer == null) { + after.putNull("payload"); + } else { + ByteBuffer copy = buffer.duplicate(); + byte[] bytes = new byte[copy.remaining()]; + copy.get(bytes); + after.put("payload", Base64.getEncoder().encodeToString(bytes)); + } + return value; + } +} diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordUtilsTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordUtilsTest.java new file mode 100644 index 000000000000..f2dad9cb507f --- /dev/null +++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordUtilsTest.java @@ -0,0 +1,156 @@ +/* + * 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.paimon.flink.sink.cdc; + +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowKind; + +import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer; +import org.apache.flink.core.memory.DataInputDeserializer; +import org.apache.flink.core.memory.DataOutputSerializer; +import org.apache.flink.util.InstantiationUtil; +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Base64; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for binary-aware CDC conversion and compatibility with ordinary string values. */ +class CdcRecordUtilsTest { + + private static final byte[] BINARY = {0, (byte) 0xff, (byte) 0x80, 65}; + + @Test + void testBinaryAndTextConversions() { + String encoded = Base64.getEncoder().encodeToString(BINARY); + Map data = new HashMap<>(); + data.put("binary", encoded); + data.put("text", encoded); + data.put("longtext", "中文"); + data.put("bits", "AQI="); + CdcRecord record = + CdcRecord.withBase64EncodedFields( + RowKind.DELETE, data, Collections.singleton("binary")); + List fields = + Arrays.asList( + new DataField(0, "binary", DataTypes.BYTES()), + new DataField(1, "text", DataTypes.BYTES()), + new DataField(2, "longtext", DataTypes.BYTES()), + new DataField(3, "bits", DataTypes.BINARY(2))); + GenericRow row = CdcRecordUtils.toGenericRow(record, fields, true).get(); + assertThat(row.getRowKind()).isEqualTo(RowKind.DELETE); + assertValues(row, encoded); + GenericRow projected = CdcRecordUtils.projectAsInsert(record, fields); + assertThat(projected.getRowKind()).isEqualTo(RowKind.INSERT); + assertValues(projected, encoded); + } + + private static void assertValues(GenericRow row, String encoded) { + assertThat(row.getBinary(0)).containsExactly(BINARY); + // A Base64-looking value from Canal, a custom record, or a text mapping stays text. + assertThat(row.getBinary(1)).containsExactly(encoded.getBytes(StandardCharsets.UTF_8)); + assertThat(row.getBinary(2)).containsExactly("中文".getBytes(StandardCharsets.UTF_8)); + assertThat(row.getBinary(3)).containsExactly((byte) 1, (byte) 2); + } + + @Test + void testBinarySourceIntoExistingTextColumn() { + CdcRecord record = + CdcRecord.withBase64EncodedFields( + RowKind.INSERT, + Collections.singletonMap("payload", "cGxhaW4gdGV4dA=="), + Collections.singleton("payload")); + List fields = + Collections.singletonList(new DataField(0, "payload", DataTypes.STRING())); + assertThat(CdcRecordUtils.toGenericRow(record, fields, true).get().getString(0).toString()) + .isEqualTo("plain text"); + assertThat(CdcRecordUtils.projectAsInsert(record, fields).getString(0).toString()) + .isEqualTo("plain text"); + } + + @Test + void testDecodedLengthControlsSchemaEvolution() { + CdcRecord record = + CdcRecord.withBase64EncodedFields( + RowKind.INSERT, + Collections.singletonMap( + "payload", Base64.getEncoder().encodeToString(BINARY)), + Collections.singleton("payload")); + List fields = + Collections.singletonList(new DataField(0, "payload", DataTypes.VARBINARY(4))); + assertThat(CdcRecordUtils.toGenericRow(record, fields, true).get().getBinary(0)) + .containsExactly(BINARY); + assertThat(CdcRecordUtils.projectAsInsert(record, fields).getBinary(0)) + .containsExactly(BINARY); + List tooShort = + Collections.singletonList(new DataField(0, "payload", DataTypes.VARBINARY(3))); + assertThat(CdcRecordUtils.toGenericRow(record, tooShort, false)).isEmpty(); + assertThatThrownBy(() -> CdcRecordUtils.projectAsInsert(record, tooShort)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("length of 4"); + List fixed = + Collections.singletonList(new DataField(0, "payload", DataTypes.BINARY(4))); + assertThat(CdcRecordUtils.toGenericRow(record, fixed, true).get().getBinary(0)) + .containsExactly(BINARY); + } + + @Test + void testBinaryMetadataSurvivesCaseConversionAndSerialization() throws Exception { + Set base64EncodedFields = new HashSet<>(Collections.singleton("PAYLOAD")); + CdcRecord record = + CdcRecord.withBase64EncodedFields( + RowKind.INSERT, + Collections.singletonMap( + "PAYLOAD", Base64.getEncoder().encodeToString(BINARY)), + base64EncodedFields); + base64EncodedFields + .clear(); // The next parser record must not mutate an emitted record's metadata. + CdcRecord lower = record.fieldNameLowerCase(); + assertThat(lower.base64EncodedFields()).containsExactly("payload"); + assertThat(record.base64EncodedFields()).containsExactly("PAYLOAD"); + List fields = + Collections.singletonList(new DataField(0, "payload", DataTypes.BYTES())); + assertThat(CdcRecordUtils.projectAsInsert(lower, fields).getBinary(0)) + .containsExactly(BINARY); + CdcRecord javaCopy = InstantiationUtil.clone(lower); + assertThat(javaCopy).isEqualTo(lower); + assertThat(CdcRecordUtils.toGenericRow(javaCopy, fields, true).get().getBinary(0)) + .containsExactly(BINARY); + KryoSerializer serializer = + CdcRecordSerializeITCase.createFlinkKryoSerializer(CdcRecord.class); + DataOutputSerializer output = new DataOutputSerializer(128); + serializer.serialize(lower, output); + CdcRecord kryoCopy = + serializer.deserialize(new DataInputDeserializer(output.getCopyOfBuffer())); + assertThat(kryoCopy).isEqualTo(lower); + assertThat(CdcRecordUtils.projectAsInsert(kryoCopy, fields).getBinary(0)) + .containsExactly(BINARY); + } +}