Skip to content

[FLINK-38886] Introduce GenericRecordData and Internal / External converters in flink-cdc-common #4218

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lvyanquan merged 4 commits into from
Jan 23, 2026
Merged

[FLINK-38886] Introduce GenericRecordData and Internal / External converters in flink-cdc-common #4218

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d55ee43
[FLINK-38886] Introduce `GenericRecordData` and Internal / External c…
yuxiqian Jan 12, 2026
a0b2d04
add: variant support
yuxiqian Jan 15, 2026
58d5f51
Fix classpath in docs
yuxiqian Jan 16, 2026
7731e01
Remove stale variant tests, moved to VariantConvertingTest class
yuxiqian Jan 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 62 additions & 0 deletions docs/content.zh/docs/core-concept/type-mappings.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
---
title: "Type Mappings"
weight: 8
type: docs
aliases:
- /core-concept/type-mappings/
---
<!--
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.
-->

# 类型映射

对于每个 CDC 数据类型（`org.apache.flink.cdc.common.types.DataType` 的子类），我们规定了用于内部序列化和反序列化的 CDC 内部类型，以及用于类型合并、类型转换和 UDF 求值的 Java 外部类型。

## 内部类型和外部类型

一些基本类型对于内部类型和 Java 类可能具有相同的表示形式（例如，`DataTypes.INT()` 使用 `java.lang.Integer` 同时作为内部和外部类型）。
其他类型可能使用不同的表示形式，例如，`DataTypes.TIMESTAMP` 在内部表示中使用 `org.apache.flink.cdc.common.data.TimestampData`，在外部操作中使用 `java.time.LocalDateTime`。

如果您正在编写 YAML Pipeline 连接器，`DataChangeEvent` 应当携带内部类型 `RecordData`，并且其所有字段都是内部类型的实例。

如果您正在编写 Transform UDF，则其参数和返回值类型应定义为其外部 Java 类型。

## 完整类型列表

| CDC 数据类型 | CDC 内部类型 | Java 外部类型 |
|--------------------------------|------------------------------------------------------------|-----------------------------------------------------|
| BOOLEAN | `java.lang.Boolean` | `java.lang.Boolean` |
| TINYINT | `java.lang.Byte` | `java.lang.Byte` |
| SMALLINT | `java.lang.Short` | `java.lang.Short` |
| INTEGER | `java.lang.Integer` | `java.lang.Integer` |
| BIGINT | `java.lang.Long` | `java.lang.Long` |
| FLOAT | `java.lang.Float` | `java.lang.Float` |
| DOUBLE | `java.lang.Double` | `java.lang.Double` |
| DECIMAL | `org.apache.flink.cdc.common.data.DecimalData` | `java.math.BigDecimal` |
| DATE | `org.apache.flink.cdc.common.data.DateData` | `java.time.LocalDate` |
| TIME | `org.apache.flink.cdc.common.data.TimeData` | `java.time.LocalTime` |
| TIMESTAMP | `org.apache.flink.cdc.common.data.TimestampData` | `java.time.LocalDateTime` |
| TIMESTAMP_TZ | `org.apache.flink.cdc.common.data.ZonedTimestampData` | `java.time.ZonedDateTime` |
| TIMESTAMP_LTZ | `org.apache.flink.cdc.common.data.LocalZonedTimestampData` | `java.time.Instant` |
| CHAR<br/>VARCHAR<br/>STRING | `org.apache.flink.cdc.common.data.StringData` | `java.lang.String` |
| BINARY<br/>VARBINARY<br/>BYTES | `byte[]` | `byte[]` |
| ARRAY | `org.apache.flink.cdc.common.data.ArrayData` | `java.util.List<T>` |
| MAP | `org.apache.flink.cdc.common.data.MapData` | `java.util.Map<K, V>` |
| ROW | `org.apache.flink.cdc.common.data.RecordData` | `java.util.List<Object>` |
| VARIANT | `org.apache.flink.cdc.common.types.variant.Variant` | `org.apache.flink.cdc.common.types.variant.Variant` |
62 changes: 62 additions & 0 deletions docs/content/docs/core-concept/type-mappings.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
---
title: "Type Mappings"
weight: 8
type: docs
aliases:
- /core-concept/type-mappings/
---
<!--
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.
-->

# Type Mappings

For each CDC DataType (subclasses of `org.apache.flink.cdc.common.types.DataType`), we define CDC Internal types for internal serialization & deserialization and Java classes (for type merging, casting, and UDF evaluations).

## Internal and External Types

Primitive types may have the same representation for Internal type and Java classes (`DataTypes.INT()` only uses `java.lang.Integer`).
Other types use different representations, like `DataTypes.TIMESTAMP` uses `org.apache.flink.cdc.common.data.TimestampData` for internal representation and `java.time.LocalDateTime` for external operations.

If you're writing a pipeline source / sink connector, `DataChangeEvent` carries internal type `RecordData`, and all its fields are internal type instances.

If you're writing a UDF, its arguments and return value types should be defined as its external Java type.

## Full Types List

| CDC Data Type | CDC Internal Type | External Java Class |
|--------------------------------|------------------------------------------------------------|-----------------------------------------------------|
| BOOLEAN | `java.lang.Boolean` | `java.lang.Boolean` |
| TINYINT | `java.lang.Byte` | `java.lang.Byte` |
| SMALLINT | `java.lang.Short` | `java.lang.Short` |
| INTEGER | `java.lang.Integer` | `java.lang.Integer` |
| BIGINT | `java.lang.Long` | `java.lang.Long` |
| FLOAT | `java.lang.Float` | `java.lang.Float` |
| DOUBLE | `java.lang.Double` | `java.lang.Double` |
| DECIMAL | `org.apache.flink.cdc.common.data.DecimalData` | `java.math.BigDecimal` |
| DATE | `org.apache.flink.cdc.common.data.DateData` | `java.time.LocalDate` |
| TIME | `org.apache.flink.cdc.common.data.TimeData` | `java.time.LocalTime` |
| TIMESTAMP | `org.apache.flink.cdc.common.data.TimestampData` | `java.time.LocalDateTime` |
| TIMESTAMP_TZ | `org.apache.flink.cdc.common.data.ZonedTimestampData` | `java.time.ZonedDateTime` |
| TIMESTAMP_LTZ | `org.apache.flink.cdc.common.data.LocalZonedTimestampData` | `java.time.Instant` |
| CHAR<br/>VARCHAR<br/>STRING | `org.apache.flink.cdc.common.data.StringData` | `java.lang.String` |
| BINARY<br/>VARBINARY<br/>BYTES | `byte[]` | `byte[]` |
| ARRAY | `org.apache.flink.cdc.common.data.ArrayData` | `java.util.List<T>` |
| MAP | `org.apache.flink.cdc.common.data.MapData` | `java.util.Map<K, V>` |
| ROW | `org.apache.flink.cdc.common.data.RecordData` | `java.util.List<Object>` |
| VARIANT | `org.apache.flink.cdc.common.types.variant.Variant` | `org.apache.flink.cdc.common.types.variant.Variant` |
Loading