[flink][cdc] Preserve binary values in Debezium JSON and MySQL records

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<String, String> 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<String, String> rowData) {
+    public String evalFromRecord(Map<String, String> rowData, Set<String> base64EncodedFields) {
+        return eval(referencedValue(rowData, base64EncodedFields));
+    }
+
+    @Nullable
+    private String referencedValue(Map<String, String> rowData, Set<String> 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<String, String> 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.
+     *
+     * <p>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;
+    }
 }

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<String, String> rowData, CdcSchema.Builder schemaBuilder) {
+        Set<String> 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<String, String> data, CdcSchema.Builder schemaBuilder) {
         schemaBuilder.primaryKey(extractPrimaryKeys());
+        Set<String> 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<String> extractBase64EncodedFields() {
+        return Collections.emptySet();
     }
 
     @Nullable

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<String, String> extractRowData(JsonNode record, CdcSchema.Builder
         return resultMap;
     }
 
+    @Override
+    protected Set<String> 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<String> extractPrimaryKeys() {
         List<String> primaryKeys = super.extractPrimaryKeys();

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 {

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlRecordParser.java

@@ -278,8 +278,12 @@ private Map<String, String> extractRow(JsonNode recordRow) {
         }
 
         // generate values of computed columns
+        Set<String> 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<String, String> extractRow(JsonNode recordRow) {
         return resultMap;
     }
 
+    private Set<String> extractBase64EncodedFields() {
+        Set<String> 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<String, String> data) {
         return new RichCdcMultiplexRecord(
-                databaseName, currentTable, null, new CdcRecord(rowKind, data));
+                databaseName,
+                currentTable,
+                null,
+                CdcRecord.withBase64EncodedFields(rowKind, data, extractBase64EncodedFields()));
     }
 }

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<String, String> data) {
         this.data = data;
     }
 
+    /**
+     * Creates a record whose specified fields contain Base64-encoded binary values.
+     *
+     * <p>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<String, String> data, Set<String> 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<String> 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<String> 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<String> binaryFields;
+
+        private BinaryCdcRecord(
+                RowKind kind, Map<String, String> data, Set<String> base64EncodedFields) {
+            super(kind, data);
+            this.binaryFields = new HashSet<>(base64EncodedFields);
+        }
+
+        @Override
+        public Set<String> base64EncodedFields() {
+            return binaryFields;
+        }
     }
 
     @Override

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<DataField> 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<GenericRow> 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<GenericRow> 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<String> fieldNames) {
         Map<String, String> data = new HashMap<>();
         for (int i = 0; i < row.getFieldCount(); i++) {