[FLINK-39417] Fix GenericRecordData could not be [de]serialized in pipeline

[yuxiqian]

This closes FLINK-39417.

Previously in FLINK-38886, we introduced a GenericRecordData as a replacement of BinaryRecordData so connector and internal operator developers could get rid of the heavy flink-cdc-runtime operator and construct binary-unaware records.

However, lacking corresponding serializers, it could not be used in complete pipelines but only in specific test cases.

[Copilot]

Pull request overview

This PR addresses FLINK-39417 by enabling end-to-end (de)serialization of GenericRecordData in Flink CDC pipelines, complementing the earlier introduction of GenericRecordData as a binary-unaware RecordData implementation.

Changes:

• Extend RecordDataSerializer to support both BinaryRecordData and GenericRecordData.
• Introduce a new self-describing GenericRecordDataSerializer (type-tagged field encoding).
• Broaden runtime/operator/test coverage to exercise GenericRecordData through transforms and Values connector pipelines.

Reviewed changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/data/RecordDataSerializerTest.java	Adds unit tests covering serialization/copy for GenericRecordData and nested structures.
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/BinaryWriter.java	Converts non-binary RecordData rows to BinaryRecordData when writing ROW fields.
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/RecordDataSerializer.java	Adds a type-tagged wrapper format to serialize BinaryRecordData vs GenericRecordData.
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/GenericRecordDataSerializer.java	New self-describing serializer for GenericRecordData fields (incl. nested rows/array/map/variant).
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformProcessor.java	Accepts RecordData input to allow pre-transform projection for generic rows.
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java	Stops casting change-event before/after to BinaryRecordData, enabling generic rows.
flink-cdc-connect/.../ValuesDatabase.java	Builds PK strings via FieldGetter to support non-string PK types and non-binary records.
flink-cdc-composer/src/test/java/.../FlinkPipelineComposerITCase.java	Adds an IT case that runs GenericRecordData end-to-end across many types incl. array/map/row.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.