You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2026/01/15 06:31:27 UTC
(flink-cdc) branch master updated: [FLINK-38877][common] Add variant type. (#4217)
This is an automated email from the ASF dual-hosted git repository.
kunni pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new f5cec12ca [FLINK-38877][common] Add variant type. (#4217)
f5cec12ca is described below
commit f5cec12ca4ebd27897a6d01fef16699c2ef063ca
Author: Kunni <de...@gmail.com>
AuthorDate: Thu Jan 15 14:31:22 2026 +0800
[FLINK-38877][common] Add variant type. (#4217)
---
.../apache/flink/cdc/common/data/RecordData.java | 7 +
.../cdc/common/data/binary/BinaryRecordData.java | 7 +
.../cdc/common/data/binary/BinarySegmentUtils.java | 19 +
.../flink/cdc/common/types/DataTypeRoot.java | 4 +-
.../flink/cdc/common/types/DataTypeVisitor.java | 4 +
.../apache/flink/cdc/common/types/DataTypes.java | 4 +
.../apache/flink/cdc/common/types/VariantType.java | 65 ++
.../cdc/common/types/variant/BinaryVariant.java | 500 ++++++++++++++++
.../common/types/variant/BinaryVariantBuilder.java | 206 +++++++
.../variant/BinaryVariantInternalBuilder.java | 657 +++++++++++++++++++++
.../common/types/variant/BinaryVariantUtil.java | 629 ++++++++++++++++++++
.../flink/cdc/common/types/variant/Variant.java | 227 +++++++
.../cdc/common/types/variant/VariantBuilder.java | 109 ++++
.../common/types/variant/VariantTypeException.java | 36 ++
.../variant/BinaryVariantInternalBuilderTest.java | 119 ++++
.../common/types/variant/BinaryVariantTest.java | 265 +++++++++
.../runtime/serializer/InternalSerializers.java | 2 +
.../cdc/runtime/serializer/VariantSerializer.java | 125 ++++
.../data/writer/AbstractBinaryWriter.java | 16 +
.../serializer/data/writer/BinaryArrayWriter.java | 1 +
.../serializer/data/writer/BinaryWriter.java | 6 +
.../serializer/schema/DataTypeSerializer.java | 9 +-
.../runtime/serializer/VariantSerializerTest.java | 65 ++
.../typeutils/BinaryRecordDataGeneratorTest.java | 20 +-
24 files changed, 3098 insertions(+), 4 deletions(-)
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/RecordData.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/RecordData.java
index 4bfd20e2e..c0289ab6c 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/RecordData.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/RecordData.java
@@ -19,6 +19,7 @@ package org.apache.flink.cdc.common.data;
import org.apache.flink.cdc.common.annotation.PublicEvolving;
import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.variant.Variant;
import javax.annotation.Nullable;
@@ -169,6 +170,9 @@ public interface RecordData {
/** Returns the Time data at the given position. */
TimeData getTime(int pos);
+ /** Returns the variant value at the given position. */
+ Variant getVariant(int pos);
+
/**
* Creates an accessor for getting elements in an internal RecordData structure at the given
* position.
@@ -242,6 +246,9 @@ public interface RecordData {
final int rowFieldCount = getFieldCount(fieldType);
fieldGetter = row -> row.getRow(fieldPos, rowFieldCount);
break;
+ case VARIANT:
+ fieldGetter = row -> row.getVariant(fieldPos);
+ break;
default:
throw new IllegalArgumentException();
}
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinaryRecordData.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinaryRecordData.java
index 5fe100f22..bc645eb7d 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinaryRecordData.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinaryRecordData.java
@@ -28,6 +28,7 @@ import org.apache.flink.cdc.common.data.StringData;
import org.apache.flink.cdc.common.data.TimeData;
import org.apache.flink.cdc.common.data.TimestampData;
import org.apache.flink.cdc.common.data.ZonedTimestampData;
+import org.apache.flink.cdc.common.types.variant.BinaryVariant;
import org.apache.flink.cdc.common.utils.Preconditions;
import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.core.memory.MemorySegmentFactory;
@@ -230,6 +231,12 @@ public final class BinaryRecordData extends BinarySection implements RecordData,
return TimeData.fromMillisOfDay(getInt(pos));
}
+ @Override
+ public BinaryVariant getVariant(int pos) {
+ assertIndexIsValid(pos);
+ return BinarySegmentUtils.readVariant(segments, offset, getLong(pos));
+ }
+
/** The bit is 1 when the field is null. Default is 0. */
@Override
public boolean anyNull() {
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinarySegmentUtils.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinarySegmentUtils.java
index f8cf849bd..9c463106d 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinarySegmentUtils.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinarySegmentUtils.java
@@ -25,10 +25,12 @@ import org.apache.flink.cdc.common.data.MapData;
import org.apache.flink.cdc.common.data.RecordData;
import org.apache.flink.cdc.common.data.StringData;
import org.apache.flink.cdc.common.data.TimestampData;
+import org.apache.flink.cdc.common.types.variant.BinaryVariant;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.core.memory.MemorySegment;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import static org.apache.flink.cdc.common.data.binary.BinaryFormat.HIGHEST_FIRST_BIT;
@@ -1106,6 +1108,23 @@ public final class BinarySegmentUtils {
return recordData;
}
+ public static BinaryVariant readVariant(
+ MemorySegment[] segments, int baseOffset, long offsetAndSize) {
+ final int size = ((int) offsetAndSize);
+ int offset = (int) (offsetAndSize >> 32);
+ byte[] bytes = copyToBytes(segments, offset + baseOffset, size);
+ ByteBuffer buffer = ByteBuffer.wrap(bytes);
+ int metaLen = buffer.getInt();
+ int valueLen = bytes.length - 4 - metaLen;
+
+ byte[] meta = new byte[metaLen];
+ byte[] value = new byte[valueLen];
+ buffer.get(meta, 0, metaLen);
+ buffer.get(value, 0, valueLen);
+
+ return new BinaryVariant(value, meta);
+ }
+
/**
* Find equal segments2 in segments1.
*
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DataTypeRoot.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DataTypeRoot.java
index 357fea64f..5a1f2b2fc 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DataTypeRoot.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DataTypeRoot.java
@@ -104,7 +104,9 @@ public enum DataTypeRoot {
MAP(DataTypeFamily.CONSTRUCTED, DataTypeFamily.EXTENSION),
- ROW(DataTypeFamily.CONSTRUCTED);
+ ROW(DataTypeFamily.CONSTRUCTED),
+
+ VARIANT(DataTypeFamily.EXTENSION);
private final Set<DataTypeFamily> families;
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DataTypeVisitor.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DataTypeVisitor.java
index fe38fa8ee..94ee43e2d 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DataTypeVisitor.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DataTypeVisitor.java
@@ -67,4 +67,8 @@ public interface DataTypeVisitor<R> {
R visit(MapType mapType);
R visit(RowType rowType);
+
+ default R visit(VariantType variantType) {
+ throw new UnsupportedOperationException("VariantType was not supported.");
+ }
}
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DataTypes.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DataTypes.java
index 4c5b6cb4d..c16c11e82 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DataTypes.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DataTypes.java
@@ -419,6 +419,10 @@ public class DataTypes {
return RowType.builder().fields(fieldTypes).build();
}
+ public static VariantType VARIANT() {
+ return new VariantType();
+ }
+
public static OptionalInt getPrecision(DataType dataType) {
return dataType.accept(PRECISION_EXTRACTOR);
}
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/VariantType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/VariantType.java
new file mode 100644
index 000000000..d07907c44
--- /dev/null
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/VariantType.java
@@ -0,0 +1,65 @@
+/*
+ * 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.flink.cdc.common.types;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Data type of semi-structured data.
+ *
+ * <p>The type supports storing any semi-structured data, including ARRAY, MAP, and scalar types.
+ * VARIANT can only store MAP types with keys of type STRING. The data type of the fields are stored
+ * in the data structure, which is close to the semantics of JSON. Compared to ROW and STRUCTURED
+ * type, VARIANT type has the flexibility that supports highly nested and evolving schema.
+ *
+ * <p>The serializable string representation of this type is {@code VARIANT}.
+ */
+@PublicEvolving
+public final class VariantType extends DataType {
+
+ public VariantType(boolean isNullable) {
+ super(isNullable, DataTypeRoot.VARIANT);
+ }
+
+ public VariantType() {
+ this(true);
+ }
+
+ @Override
+ public DataType copy(boolean isNullable) {
+ return new VariantType(isNullable);
+ }
+
+ @Override
+ public String asSerializableString() {
+ return withNullability("VARIANT");
+ }
+
+ @Override
+ public List<DataType> getChildren() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public <R> R accept(DataTypeVisitor<R> visitor) {
+ return visitor.visit(this);
+ }
+}
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/BinaryVariant.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/BinaryVariant.java
new file mode 100644
index 000000000..5aab53697
--- /dev/null
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/BinaryVariant.java
@@ -0,0 +1,500 @@
+/*
+ * 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.flink.cdc.common.types.variant;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonFactory;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+
+import java.io.CharArrayWriter;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.time.temporal.ChronoUnit;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.Objects;
+
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.BINARY_SEARCH_THRESHOLD;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.SIZE_LIMIT;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.TIMESTAMP_FORMATTER;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.TIMESTAMP_LTZ_FORMATTER;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.VERSION;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.VERSION_MASK;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.checkIndex;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.getMetadataKey;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.handleArray;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.handleObject;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.malformedVariant;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.readUnsigned;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.unexpectedType;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.valueSize;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.variantConstructorSizeLimit;
+
+/**
+ * Copy from <a
+ * href="https://github.com/apache/flink/blob/release-2.2.0/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java">BinaryVariant.java</a>.
+ *
+ * <p>A data structure that represents a semi-structured value. It consists of two binary values:
+ * value and metadata. The value encodes types and values, but not field names. The metadata
+ * currently contains a version flag and a list of field names. We can extend/modify the detailed
+ * binary format given the version flag.
+ *
+ * @see <a href="https://github.com/apache/parquet-format/blob/master/VariantEncoding.md">Variant
+ * Binary Encoding</a> for the detail layout of the data structure.
+ */
+@Internal
+public final class BinaryVariant implements Variant {
+
+ private final byte[] value;
+ private final byte[] metadata;
+ // The variant value doesn't use the whole `value` binary, but starts from its `pos` index and
+ // spans a size of `valueSize(value, pos)`. This design avoids frequent copies of the value
+ // binary when reading a sub-variant in the array/object element.
+ private final int pos;
+
+ public BinaryVariant(byte[] value, byte[] metadata) {
+ this(value, metadata, 0);
+ }
+
+ private BinaryVariant(byte[] value, byte[] metadata, int pos) {
+ this.value = value;
+ this.metadata = metadata;
+ this.pos = pos;
+ // There is currently only one allowed version.
+ if (metadata.length < 1 || (metadata[0] & VERSION_MASK) != VERSION) {
+ throw malformedVariant();
+ }
+ // Don't attempt to use a Variant larger than 16 MiB. We'll never produce one, and it risks
+ // memory instability.
+ if (metadata.length > SIZE_LIMIT || value.length > SIZE_LIMIT) {
+ throw variantConstructorSizeLimit();
+ }
+ }
+
+ @Override
+ public boolean isPrimitive() {
+ return !isArray() && !isObject();
+ }
+
+ @Override
+ public boolean isArray() {
+ return getType() == Type.ARRAY;
+ }
+
+ @Override
+ public boolean isObject() {
+ return getType() == Type.OBJECT;
+ }
+
+ @Override
+ public boolean isNull() {
+ return getType() == Type.NULL;
+ }
+
+ @Override
+ public Type getType() {
+ return BinaryVariantUtil.getType(value, pos);
+ }
+
+ @Override
+ public boolean getBoolean() throws VariantTypeException {
+ checkType(Type.BOOLEAN, getType());
+ return BinaryVariantUtil.getBoolean(value, pos);
+ }
+
+ @Override
+ public byte getByte() throws VariantTypeException {
+ checkType(Type.TINYINT, getType());
+ return (byte) BinaryVariantUtil.getLong(value, pos);
+ }
+
+ @Override
+ public short getShort() throws VariantTypeException {
+ checkType(Type.SMALLINT, getType());
+ return (short) BinaryVariantUtil.getLong(value, pos);
+ }
+
+ @Override
+ public int getInt() throws VariantTypeException {
+ checkType(Type.INT, getType());
+ return (int) BinaryVariantUtil.getLong(value, pos);
+ }
+
+ @Override
+ public long getLong() throws VariantTypeException {
+ checkType(Type.BIGINT, getType());
+ return BinaryVariantUtil.getLong(value, pos);
+ }
+
+ @Override
+ public float getFloat() throws VariantTypeException {
+ checkType(Type.FLOAT, getType());
+ return BinaryVariantUtil.getFloat(value, pos);
+ }
+
+ @Override
+ public BigDecimal getDecimal() throws VariantTypeException {
+ checkType(Type.DECIMAL, getType());
+ return BinaryVariantUtil.getDecimal(value, pos);
+ }
+
+ @Override
+ public double getDouble() throws VariantTypeException {
+ checkType(Type.DOUBLE, getType());
+ return BinaryVariantUtil.getDouble(value, pos);
+ }
+
+ @Override
+ public String getString() throws VariantTypeException {
+ checkType(Type.STRING, getType());
+ return BinaryVariantUtil.getString(value, pos);
+ }
+
+ @Override
+ public LocalDate getDate() throws VariantTypeException {
+ checkType(Type.DATE, getType());
+ return LocalDate.ofEpochDay(BinaryVariantUtil.getLong(value, pos));
+ }
+
+ @Override
+ public LocalDateTime getDateTime() throws VariantTypeException {
+ checkType(Type.TIMESTAMP, getType());
+ return microsToInstant(BinaryVariantUtil.getLong(value, pos))
+ .atZone(ZoneOffset.UTC)
+ .toLocalDateTime();
+ }
+
+ @Override
+ public Instant getInstant() throws VariantTypeException {
+ checkType(Type.TIMESTAMP_LTZ, getType());
+ return microsToInstant(BinaryVariantUtil.getLong(value, pos));
+ }
+
+ @Override
+ public byte[] getBytes() throws VariantTypeException {
+ checkType(Type.BYTES, getType());
+ return BinaryVariantUtil.getBinary(value, pos);
+ }
+
+ @Override
+ public Object get() throws VariantTypeException {
+ switch (getType()) {
+ case NULL:
+ return null;
+ case BOOLEAN:
+ return getBoolean();
+ case TINYINT:
+ return getByte();
+ case SMALLINT:
+ return getShort();
+ case INT:
+ return getInt();
+ case BIGINT:
+ return getLong();
+ case FLOAT:
+ return getFloat();
+ case DOUBLE:
+ return getDouble();
+ case DECIMAL:
+ return getDecimal();
+ case STRING:
+ return getString();
+ case DATE:
+ return getDate();
+ case TIMESTAMP:
+ return getDateTime();
+ case TIMESTAMP_LTZ:
+ return getInstant();
+ case BYTES:
+ return getBytes();
+ default:
+ throw new VariantTypeException(
+ String.format("Expecting a primitive variant but got %s", getType()));
+ }
+ }
+
+ @Override
+ public <T> T getAs() throws VariantTypeException {
+ return (T) get();
+ }
+
+ @Override
+ public Variant getElement(int index) throws VariantTypeException {
+ return getElementAtIndex(index);
+ }
+
+ @Override
+ public Variant getField(String fieldName) throws VariantTypeException {
+ return getFieldByKey(fieldName);
+ }
+
+ @Override
+ public String toJson() {
+ StringBuilder sb = new StringBuilder();
+ toJsonImpl(value, metadata, pos, sb, ZoneOffset.UTC);
+ return sb.toString();
+ }
+
+ public byte[] getValue() {
+ if (pos == 0) {
+ return value;
+ }
+ int size = valueSize(value, pos);
+ checkIndex(pos + size - 1, value.length);
+ return Arrays.copyOfRange(value, pos, pos + size);
+ }
+
+ public byte[] getMetadata() {
+ return metadata;
+ }
+
+ public int getPos() {
+ return pos;
+ }
+
+ private static void toJsonImpl(
+ byte[] value, byte[] metadata, int pos, StringBuilder sb, ZoneId zoneId) {
+ switch (BinaryVariantUtil.getType(value, pos)) {
+ case OBJECT:
+ handleObject(
+ value,
+ pos,
+ (size, idSize, offsetSize, idStart, offsetStart, dataStart) -> {
+ sb.append('{');
+ for (int i = 0; i < size; ++i) {
+ int id = readUnsigned(value, idStart + idSize * i, idSize);
+ int offset =
+ readUnsigned(
+ value, offsetStart + offsetSize * i, offsetSize);
+ int elementPos = dataStart + offset;
+ if (i != 0) {
+ sb.append(',');
+ }
+ sb.append(escapeJson(getMetadataKey(metadata, id)));
+ sb.append(':');
+ toJsonImpl(value, metadata, elementPos, sb, zoneId);
+ }
+ sb.append('}');
+ return null;
+ });
+ break;
+ case ARRAY:
+ handleArray(
+ value,
+ pos,
+ (size, offsetSize, offsetStart, dataStart) -> {
+ sb.append('[');
+ for (int i = 0; i < size; ++i) {
+ int offset =
+ readUnsigned(
+ value, offsetStart + offsetSize * i, offsetSize);
+ int elementPos = dataStart + offset;
+ if (i != 0) {
+ sb.append(',');
+ }
+ toJsonImpl(value, metadata, elementPos, sb, zoneId);
+ }
+ sb.append(']');
+ return null;
+ });
+ break;
+ case NULL:
+ sb.append("null");
+ break;
+ case BOOLEAN:
+ sb.append(BinaryVariantUtil.getBoolean(value, pos));
+ break;
+ case TINYINT:
+ case SMALLINT:
+ case INT:
+ case BIGINT:
+ sb.append(BinaryVariantUtil.getLong(value, pos));
+ break;
+ case STRING:
+ sb.append(escapeJson(BinaryVariantUtil.getString(value, pos)));
+ break;
+ case DOUBLE:
+ sb.append(BinaryVariantUtil.getDouble(value, pos));
+ break;
+ case DECIMAL:
+ sb.append(BinaryVariantUtil.getDecimal(value, pos).toPlainString());
+ break;
+ case DATE:
+ appendQuoted(
+ sb,
+ LocalDate.ofEpochDay((int) BinaryVariantUtil.getLong(value, pos))
+ .toString());
+ break;
+ case TIMESTAMP_LTZ:
+ appendQuoted(
+ sb,
+ TIMESTAMP_LTZ_FORMATTER.format(
+ microsToInstant(BinaryVariantUtil.getLong(value, pos))
+ .atZone(zoneId)));
+ break;
+ case TIMESTAMP:
+ appendQuoted(
+ sb,
+ TIMESTAMP_FORMATTER.format(
+ microsToInstant(BinaryVariantUtil.getLong(value, pos))
+ .atZone(ZoneOffset.UTC)));
+ break;
+ case FLOAT:
+ sb.append(BinaryVariantUtil.getFloat(value, pos));
+ break;
+ case BYTES:
+ appendQuoted(
+ sb,
+ Base64.getEncoder()
+ .encodeToString(BinaryVariantUtil.getBinary(value, pos)));
+ break;
+ default:
+ throw unexpectedType(BinaryVariantUtil.getType(value, pos));
+ }
+ }
+
+ private static Instant microsToInstant(long timestamp) {
+ return Instant.EPOCH.plus(timestamp, ChronoUnit.MICROS);
+ }
+
+ private void checkType(Type expected, Type actual) {
+ if (expected != actual) {
+ throw new VariantTypeException(
+ String.format("Expected type %s but got %s", expected, actual));
+ }
+ }
+
+ // Find the field value whose key is equal to `key`. Return null if the key is not found.
+ // It is only legal to call it when `getType()` is `Type.OBJECT`.
+ private BinaryVariant getFieldByKey(String key) {
+ return handleObject(
+ value,
+ pos,
+ (size, idSize, offsetSize, idStart, offsetStart, dataStart) -> {
+ // Use linear search for a short list. Switch to binary search when the length
+ // reaches `BINARY_SEARCH_THRESHOLD`.
+ if (size < BINARY_SEARCH_THRESHOLD) {
+ for (int i = 0; i < size; ++i) {
+ int id = readUnsigned(value, idStart + idSize * i, idSize);
+ if (key.equals(getMetadataKey(metadata, id))) {
+ int offset =
+ readUnsigned(
+ value, offsetStart + offsetSize * i, offsetSize);
+ return new BinaryVariant(value, metadata, dataStart + offset);
+ }
+ }
+ } else {
+ int low = 0;
+ int high = size - 1;
+ while (low <= high) {
+ // Use unsigned right shift to compute the middle of `low` and `high`.
+ // This is not only a performance optimization, because it can properly
+ // handle the case where `low + high` overflows int.
+ int mid = (low + high) >>> 1;
+ int id = readUnsigned(value, idStart + idSize * mid, idSize);
+ int cmp = getMetadataKey(metadata, id).compareTo(key);
+ if (cmp < 0) {
+ low = mid + 1;
+ } else if (cmp > 0) {
+ high = mid - 1;
+ } else {
+ int offset =
+ readUnsigned(
+ value, offsetStart + offsetSize * mid, offsetSize);
+ return new BinaryVariant(value, metadata, dataStart + offset);
+ }
+ }
+ }
+ return null;
+ });
+ }
+
+ // Get the number of array elements in the variant.
+ // It is only legal to call it when `getType()` is `Type.ARRAY`.
+ @Override
+ public int arraySize() {
+ return handleArray(value, pos, (size, offsetSize, offsetStart, dataStart) -> size);
+ }
+
+ // Get the array element at the `index` slot. Return null if `index` is out of the bound of
+ // `[0, arraySize())`.
+ // It is only legal to call it when `getType()` is `Type.ARRAY`.
+ private BinaryVariant getElementAtIndex(int index) {
+ return handleArray(
+ value,
+ pos,
+ (size, offsetSize, offsetStart, dataStart) -> {
+ if (index < 0 || index >= size) {
+ return null;
+ }
+ int offset = readUnsigned(value, offsetStart + offsetSize * index, offsetSize);
+ return new BinaryVariant(value, metadata, dataStart + offset);
+ });
+ }
+
+ // Escape a string so that it can be pasted into JSON structure.
+ // For example, if `str` only contains a new-line character, then the result content is "\n"
+ // (4 characters).
+ private static String escapeJson(String str) {
+ try (CharArrayWriter writer = new CharArrayWriter();
+ JsonGenerator gen = new JsonFactory().createGenerator(writer)) {
+ gen.writeString(str);
+ gen.flush();
+ return writer.toString();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static void appendQuoted(StringBuilder sb, String str) {
+ sb.append('"');
+ sb.append(str);
+ sb.append('"');
+ }
+
+ @Override
+ public String toString() {
+ return toJson();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof BinaryVariant)) {
+ return false;
+ }
+ BinaryVariant variant = (BinaryVariant) o;
+ return getPos() == variant.getPos()
+ && Objects.deepEquals(getValue(), variant.getValue())
+ && Objects.deepEquals(getMetadata(), variant.getMetadata());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(Arrays.hashCode(value), Arrays.hashCode(metadata), pos);
+ }
+}
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/BinaryVariantBuilder.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/BinaryVariantBuilder.java
new file mode 100644
index 000000000..b50cf581a
--- /dev/null
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/BinaryVariantBuilder.java
@@ -0,0 +1,206 @@
+/*
+ * 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.flink.cdc.common.types.variant;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.annotation.PublicEvolving;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
+
+/**
+ * Copy from <a
+ * href="https://github.com/apache/flink/blob/release-2.2.0/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantBuilder.java">BinaryVariantBuilder.java</a>
+ *
+ * <p>Builder for binary encoded variant.
+ */
+@Internal
+public class BinaryVariantBuilder implements VariantBuilder {
+
+ @Override
+ public Variant of(byte b) {
+ BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
+ builder.appendByte(b);
+ return builder.build();
+ }
+
+ @Override
+ public Variant of(short s) {
+ BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
+ builder.appendShort(s);
+ return builder.build();
+ }
+
+ @Override
+ public Variant of(int i) {
+ BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
+ builder.appendInt(i);
+ return builder.build();
+ }
+
+ @Override
+ public Variant of(long l) {
+ BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
+ builder.appendLong(l);
+ return builder.build();
+ }
+
+ @Override
+ public Variant of(String s) {
+ BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
+ builder.appendString(s);
+ return builder.build();
+ }
+
+ @Override
+ public Variant of(double d) {
+ BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
+ builder.appendDouble(d);
+ return builder.build();
+ }
+
+ @Override
+ public Variant of(float f) {
+ BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
+ builder.appendFloat(f);
+ return builder.build();
+ }
+
+ @Override
+ public Variant of(byte[] bytes) {
+ BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
+ builder.appendBinary(bytes);
+ return builder.build();
+ }
+
+ @Override
+ public Variant of(boolean b) {
+ BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
+ builder.appendBoolean(b);
+ return builder.build();
+ }
+
+ @Override
+ public Variant of(BigDecimal bigDecimal) {
+ BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
+ builder.appendDecimal(bigDecimal);
+ return builder.build();
+ }
+
+ @Override
+ public Variant of(Instant instant) {
+ BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
+ builder.appendTimestampLtz(ChronoUnit.MICROS.between(Instant.EPOCH, instant));
+ return builder.build();
+ }
+
+ @Override
+ public Variant of(LocalDate localDate) {
+ BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
+ builder.appendDate((int) localDate.toEpochDay());
+ return builder.build();
+ }
+
+ @Override
+ public Variant of(LocalDateTime localDateTime) {
+ BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
+ builder.appendTimestamp(
+ ChronoUnit.MICROS.between(Instant.EPOCH, localDateTime.toInstant(ZoneOffset.UTC)));
+ return builder.build();
+ }
+
+ @Override
+ public Variant ofNull() {
+ BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
+ builder.appendNull();
+ return builder.build();
+ }
+
+ @Override
+ public VariantObjectBuilder object() {
+ return object(false);
+ }
+
+ @Override
+ public VariantObjectBuilder object(boolean allowDuplicateKeys) {
+ return new VariantObjectBuilder(allowDuplicateKeys);
+ }
+
+ /** Get the builder for a variant array. */
+ public VariantArrayBuilder array() {
+ return new VariantArrayBuilder();
+ }
+
+ /** Builder for a variant object. */
+ @PublicEvolving
+ public static class VariantObjectBuilder implements VariantBuilder.VariantObjectBuilder {
+ private final BinaryVariantInternalBuilder builder;
+ private final ArrayList<BinaryVariantInternalBuilder.FieldEntry> entries =
+ new ArrayList<>();
+
+ public VariantObjectBuilder(boolean allowDuplicateKeys) {
+ builder = new BinaryVariantInternalBuilder(allowDuplicateKeys);
+ }
+
+ @Override
+ public VariantObjectBuilder add(String key, Variant value) {
+ int id = builder.addKey(key);
+ entries.add(
+ new BinaryVariantInternalBuilder.FieldEntry(key, id, builder.getWritePos()));
+ builder.appendVariant((BinaryVariant) value);
+ return this;
+ }
+
+ @Override
+ public Variant build() {
+ builder.finishWritingObject(0, entries);
+ return builder.build();
+ }
+ }
+
+ /** Builder for a variant array. */
+ @PublicEvolving
+ public static class VariantArrayBuilder implements VariantBuilder.VariantArrayBuilder {
+
+ private final BinaryVariantInternalBuilder builder;
+ private final ArrayList<Integer> offsets = new ArrayList<>();
+
+ public VariantArrayBuilder() {
+ builder = new BinaryVariantInternalBuilder(false);
+ }
+
+ @Override
+ public VariantArrayBuilder add(Variant value) {
+ offsets.add(builder.getWritePos());
+ builder.appendVariant((BinaryVariant) value);
+ return this;
+ }
+
+ @Override
+ public Variant build() {
+ builder.finishWritingArray(0, offsets);
+ return builder.build();
+ }
+ }
+}
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/BinaryVariantInternalBuilder.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/BinaryVariantInternalBuilder.java
new file mode 100644
index 000000000..b53c94bbe
--- /dev/null
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/BinaryVariantInternalBuilder.java
@@ -0,0 +1,657 @@
+/*
+ * 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.flink.cdc.common.types.variant;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonFactory;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParseException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonToken;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.exc.InputCoercionException;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.ARRAY;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.BASIC_TYPE_MASK;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.BINARY;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.DATE;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.DECIMAL16;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.DECIMAL4;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.DECIMAL8;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.DOUBLE;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.FALSE;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.FLOAT;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.INT1;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.INT2;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.INT4;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.INT8;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.LONG_STR;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.MAX_DECIMAL16_PRECISION;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.MAX_DECIMAL4_PRECISION;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.MAX_DECIMAL8_PRECISION;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.MAX_SHORT_STR_SIZE;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.NULL;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.OBJECT;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.SIZE_LIMIT;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.TIMESTAMP;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.TIMESTAMP_LTZ;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.TRUE;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.U16_MAX;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.U24_MAX;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.U24_SIZE;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.U32_SIZE;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.U8_MAX;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.VERSION;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.arrayHeader;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.checkIndex;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.getMetadataKey;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.handleArray;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.handleObject;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.objectHeader;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.primitiveHeader;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.readUnsigned;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.shortStrHeader;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.valueSize;
+import static org.apache.flink.cdc.common.types.variant.BinaryVariantUtil.writeLong;
+
+/**
+ * Copy from <a
+ * href="https://github.com/apache/flink/blob/release-2.2.0/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantBuilder.java">BinaryVariantBuilder.java</a>
+ *
+ * <p>The internal builder for {@link BinaryVariant}.
+ */
+@Internal
+public class BinaryVariantInternalBuilder {
+
+ public static final VariantTypeException VARIANT_SIZE_LIMIT_EXCEPTION =
+ new VariantTypeException("VARIANT_SIZE_LIMIT");
+ public static final VariantTypeException VARIANT_DUPLICATE_KEY_EXCEPTION =
+ new VariantTypeException("VARIANT_DUPLICATE_KEY");
+
+ public BinaryVariantInternalBuilder(boolean allowDuplicateKeys) {
+ this.allowDuplicateKeys = allowDuplicateKeys;
+ }
+
+ /**
+ * Parse a JSON string as a Variant value.
+ *
+ * @throws IOException if any JSON parsing error happens.
+ */
+ public static BinaryVariant parseJson(String json, boolean allowDuplicateKeys)
+ throws IOException {
+ try (JsonParser parser = new JsonFactory().createParser(json)) {
+ parser.nextToken();
+ return parseJson(parser, allowDuplicateKeys);
+ }
+ }
+
+ /**
+ * Similar {@link #parseJson(String, boolean)}, but takes a JSON parser instead of string input.
+ */
+ private static BinaryVariant parseJson(JsonParser parser, boolean allowDuplicateKeys)
+ throws IOException {
+ BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(allowDuplicateKeys);
+ builder.buildJson(parser);
+ return builder.build();
+ }
+
+ // Build the variant metadata from `dictionaryKeys` and return the variant result.
+ public BinaryVariant build() {
+ int numKeys = dictionaryKeys.size();
+ // Use long to avoid overflow in accumulating lengths.
+ long dictionaryStringSize = 0;
+ for (byte[] key : dictionaryKeys) {
+ dictionaryStringSize += key.length;
+ }
+ // Determine the number of bytes required per offset entry.
+ // The largest offset is the one-past-the-end value, which is total string size. It's very
+ // unlikely that the number of keys could be larger, but incorporate that into the
+ // calculation
+ // in case of pathological data.
+ long maxSize = Math.max(dictionaryStringSize, numKeys);
+ if (maxSize > SIZE_LIMIT) {
+ throw VARIANT_SIZE_LIMIT_EXCEPTION;
+ }
+ int offsetSize = getIntegerSize((int) maxSize);
+
+ int offsetStart = 1 + offsetSize;
+ int stringStart = offsetStart + (numKeys + 1) * offsetSize;
+ long metadataSize = stringStart + dictionaryStringSize;
+
+ if (metadataSize > SIZE_LIMIT) {
+ throw VARIANT_SIZE_LIMIT_EXCEPTION;
+ }
+ byte[] metadata = new byte[(int) metadataSize];
+ int headerByte = VERSION | ((offsetSize - 1) << 6);
+ writeLong(metadata, 0, headerByte, 1);
+ writeLong(metadata, 1, numKeys, offsetSize);
+ int currentOffset = 0;
+ for (int i = 0; i < numKeys; ++i) {
+ writeLong(metadata, offsetStart + i * offsetSize, currentOffset, offsetSize);
+ byte[] key = dictionaryKeys.get(i);
+ System.arraycopy(key, 0, metadata, stringStart + currentOffset, key.length);
+ currentOffset += key.length;
+ }
+ writeLong(metadata, offsetStart + numKeys * offsetSize, currentOffset, offsetSize);
+ return new BinaryVariant(Arrays.copyOfRange(writeBuffer, 0, writePos), metadata);
+ }
+
+ public void appendString(String str) {
+ byte[] text = str.getBytes(StandardCharsets.UTF_8);
+ boolean longStr = text.length > MAX_SHORT_STR_SIZE;
+ checkCapacity((longStr ? 1 + U32_SIZE : 1) + text.length);
+ if (longStr) {
+ writeBuffer[writePos++] = primitiveHeader(LONG_STR);
+ writeLong(writeBuffer, writePos, text.length, U32_SIZE);
+ writePos += U32_SIZE;
+ } else {
+ writeBuffer[writePos++] = shortStrHeader(text.length);
+ }
+ System.arraycopy(text, 0, writeBuffer, writePos, text.length);
+ writePos += text.length;
+ }
+
+ public void appendNull() {
+ checkCapacity(1);
+ writeBuffer[writePos++] = primitiveHeader(NULL);
+ }
+
+ public void appendBoolean(boolean b) {
+ checkCapacity(1);
+ writeBuffer[writePos++] = primitiveHeader(b ? TRUE : FALSE);
+ }
+
+ public void appendByte(byte b) {
+ checkCapacity(1 + 1);
+ writeBuffer[writePos++] = primitiveHeader(INT1);
+ writeLong(writeBuffer, writePos, b, 1);
+ writePos += 1;
+ }
+
+ public void appendShort(short s) {
+ checkCapacity(1 + 2);
+ writeBuffer[writePos++] = primitiveHeader(INT2);
+ writeLong(writeBuffer, writePos, s, 2);
+ writePos += 2;
+ }
+
+ public void appendInt(int i) {
+ checkCapacity(1 + 4);
+ writeBuffer[writePos++] = primitiveHeader(INT4);
+ writeLong(writeBuffer, writePos, i, 4);
+ writePos += 4;
+ }
+
+ public void appendLong(long l) {
+ checkCapacity(1 + 8);
+ writeBuffer[writePos++] = primitiveHeader(INT8);
+ writeLong(writeBuffer, writePos, l, 8);
+ writePos += 8;
+ }
+
+ public void appendNumeric(long l) {
+ if (l == (byte) l) {
+ appendByte((byte) l);
+ } else if (l == (short) l) {
+ appendShort((short) l);
+ } else if (l == (int) l) {
+ appendInt((int) l);
+ } else {
+ appendLong(l);
+ }
+ }
+
+ public void appendDouble(double d) {
+ checkCapacity(1 + 8);
+ writeBuffer[writePos++] = primitiveHeader(DOUBLE);
+ writeLong(writeBuffer, writePos, Double.doubleToLongBits(d), 8);
+ writePos += 8;
+ }
+
+ // Append a decimal value to the variant builder. The caller should guarantee that its precision
+ // and scale fit into `MAX_DECIMAL16_PRECISION`.
+ public void appendDecimal(BigDecimal d) {
+ checkCapacity(2 + 16);
+ BigInteger unscaled = d.unscaledValue();
+ if (d.scale() <= MAX_DECIMAL4_PRECISION && d.precision() <= MAX_DECIMAL4_PRECISION) {
+ writeBuffer[writePos++] = primitiveHeader(DECIMAL4);
+ writeBuffer[writePos++] = (byte) d.scale();
+ writeLong(writeBuffer, writePos, unscaled.intValueExact(), 4);
+ writePos += 4;
+ } else if (d.scale() <= MAX_DECIMAL8_PRECISION && d.precision() <= MAX_DECIMAL8_PRECISION) {
+ writeBuffer[writePos++] = primitiveHeader(DECIMAL8);
+ writeBuffer[writePos++] = (byte) d.scale();
+ writeLong(writeBuffer, writePos, unscaled.longValueExact(), 8);
+ writePos += 8;
+ } else {
+ assert d.scale() <= MAX_DECIMAL16_PRECISION && d.precision() <= MAX_DECIMAL16_PRECISION;
+ writeBuffer[writePos++] = primitiveHeader(DECIMAL16);
+ writeBuffer[writePos++] = (byte) d.scale();
+ // `toByteArray` returns a big-endian representation. We need to copy it reversely and
+ // sign
+ // extend it to 16 bytes.
+ byte[] bytes = unscaled.toByteArray();
+ for (int i = 0; i < bytes.length; ++i) {
+ writeBuffer[writePos + i] = bytes[bytes.length - 1 - i];
+ }
+ byte sign = (byte) (bytes[0] < 0 ? -1 : 0);
+ for (int i = bytes.length; i < 16; ++i) {
+ writeBuffer[writePos + i] = sign;
+ }
+ writePos += 16;
+ }
+ }
+
+ public void appendDate(int daysSinceEpoch) {
+ checkCapacity(1 + 4);
+ writeBuffer[writePos++] = primitiveHeader(DATE);
+ writeLong(writeBuffer, writePos, daysSinceEpoch, 4);
+ writePos += 4;
+ }
+
+ public void appendTimestampLtz(long microsSinceEpoch) {
+ checkCapacity(1 + 8);
+ writeBuffer[writePos++] = primitiveHeader(TIMESTAMP_LTZ);
+ writeLong(writeBuffer, writePos, microsSinceEpoch, 8);
+ writePos += 8;
+ }
+
+ public void appendTimestamp(long microsSinceEpoch) {
+ checkCapacity(1 + 8);
+ writeBuffer[writePos++] = primitiveHeader(TIMESTAMP);
+ writeLong(writeBuffer, writePos, microsSinceEpoch, 8);
+ writePos += 8;
+ }
+
+ public void appendFloat(float f) {
+ checkCapacity(1 + 4);
+ writeBuffer[writePos++] = primitiveHeader(FLOAT);
+ writeLong(writeBuffer, writePos, Float.floatToIntBits(f), 8);
+ writePos += 4;
+ }
+
+ public void appendBinary(byte[] binary) {
+ checkCapacity(1 + U32_SIZE + binary.length);
+ writeBuffer[writePos++] = primitiveHeader(BINARY);
+ writeLong(writeBuffer, writePos, binary.length, U32_SIZE);
+ writePos += U32_SIZE;
+ System.arraycopy(binary, 0, writeBuffer, writePos, binary.length);
+ writePos += binary.length;
+ }
+
+ // Add a key to the variant dictionary. If the key already exists, the dictionary is not
+ // modified.
+ // In either case, return the id of the key.
+ public int addKey(String key) {
+ int id;
+ if (dictionary.containsKey(key)) {
+ id = dictionary.get(key);
+ } else {
+ id = dictionaryKeys.size();
+ dictionary.put(key, id);
+ dictionaryKeys.add(key.getBytes(StandardCharsets.UTF_8));
+ }
+ return id;
+ }
+
+ // Return the current write position of the variant builder. It is used together with
+ // `finishWritingObject` or `finishWritingArray`.
+ public int getWritePos() {
+ return writePos;
+ }
+
+ // Finish writing a variant object after all of its fields have already been written. The
+ // process
+ // is as follows:
+ // 1. The caller calls `getWritePos` before writing any fields to obtain the `start` parameter.
+ // 2. The caller appends all the object fields to the builder. In the meantime, it should
+ // maintain
+ // the `fields` parameter. Before appending each field, it should append an entry to `fields` to
+ // record the offset of the field. The offset is computed as `getWritePos() - start`.
+ // 3. The caller calls `finishWritingObject` to finish writing a variant object.
+ //
+ // This function is responsible to sort the fields by key. If there are duplicate field keys:
+ // - when `allowDuplicateKeys` is true, the field with the greatest offset value (the last
+ // appended one) is kept.
+ // - otherwise, throw an exception.
+ public void finishWritingObject(int start, ArrayList<FieldEntry> fields) {
+ int size = fields.size();
+ Collections.sort(fields);
+ int maxId = size == 0 ? 0 : fields.get(0).id;
+ if (allowDuplicateKeys) {
+ int distinctPos = 0;
+ // Maintain a list of distinct keys in-place.
+ for (int i = 1; i < size; ++i) {
+ maxId = Math.max(maxId, fields.get(i).id);
+ if (fields.get(i).id == fields.get(i - 1).id) {
+ // Found a duplicate key. Keep the field with a greater offset.
+ if (fields.get(distinctPos).offset < fields.get(i).offset) {
+ fields.set(
+ distinctPos,
+ fields.get(distinctPos).withNewOffset(fields.get(i).offset));
+ }
+ } else {
+ // Found a distinct key. Add the field to the list.
+ ++distinctPos;
+ fields.set(distinctPos, fields.get(i));
+ }
+ }
+ if (distinctPos + 1 < fields.size()) {
+ size = distinctPos + 1;
+ // Resize `fields` to `size`.
+ fields.subList(size, fields.size()).clear();
+ // Sort the fields by offsets so that we can move the value data of each field to
+ // the new
+ // offset without overwriting the fields after it.
+ fields.sort(Comparator.comparingInt(f -> f.offset));
+ int currentOffset = 0;
+ for (int i = 0; i < size; ++i) {
+ int oldOffset = fields.get(i).offset;
+ int fieldSize = valueSize(writeBuffer, start + oldOffset);
+ System.arraycopy(
+ writeBuffer,
+ start + oldOffset,
+ writeBuffer,
+ start + currentOffset,
+ fieldSize);
+ fields.set(i, fields.get(i).withNewOffset(currentOffset));
+ currentOffset += fieldSize;
+ }
+ writePos = start + currentOffset;
+ // Change back to the sort order by field keys to meet the variant spec.
+ Collections.sort(fields);
+ }
+ } else {
+ for (int i = 1; i < size; ++i) {
+ maxId = Math.max(maxId, fields.get(i).id);
+ String key = fields.get(i).key;
+ if (key.equals(fields.get(i - 1).key)) {
+ throw VARIANT_DUPLICATE_KEY_EXCEPTION;
+ }
+ }
+ }
+ int dataSize = writePos - start;
+ boolean largeSize = size > U8_MAX;
+ int sizeBytes = largeSize ? U32_SIZE : 1;
+ int idSize = getIntegerSize(maxId);
+ int offsetSize = getIntegerSize(dataSize);
+ // The space for header byte, object size, id list, and offset list.
+ int headerSize = 1 + sizeBytes + size * idSize + (size + 1) * offsetSize;
+ checkCapacity(headerSize);
+ // Shift the just-written field data to make room for the object header section.
+ System.arraycopy(writeBuffer, start, writeBuffer, start + headerSize, dataSize);
+ writePos += headerSize;
+ writeBuffer[start] = objectHeader(largeSize, idSize, offsetSize);
+ writeLong(writeBuffer, start + 1, size, sizeBytes);
+ int idStart = start + 1 + sizeBytes;
+ int offsetStart = idStart + size * idSize;
+ for (int i = 0; i < size; ++i) {
+ writeLong(writeBuffer, idStart + i * idSize, fields.get(i).id, idSize);
+ writeLong(writeBuffer, offsetStart + i * offsetSize, fields.get(i).offset, offsetSize);
+ }
+ writeLong(writeBuffer, offsetStart + size * offsetSize, dataSize, offsetSize);
+ }
+
+ // Finish writing a variant array after all of its elements have already been written. The
+ // process
+ // is similar to that of `finishWritingObject`.
+ public void finishWritingArray(int start, ArrayList<Integer> offsets) {
+ int dataSize = writePos - start;
+ int size = offsets.size();
+ boolean largeSize = size > U8_MAX;
+ int sizeBytes = largeSize ? U32_SIZE : 1;
+ int offsetSize = getIntegerSize(dataSize);
+ // The space for header byte, object size, and offset list.
+ int headerSize = 1 + sizeBytes + (size + 1) * offsetSize;
+ checkCapacity(headerSize);
+ // Shift the just-written field data to make room for the header section.
+ System.arraycopy(writeBuffer, start, writeBuffer, start + headerSize, dataSize);
+ writePos += headerSize;
+ writeBuffer[start] = arrayHeader(largeSize, offsetSize);
+ writeLong(writeBuffer, start + 1, size, sizeBytes);
+ int offsetStart = start + 1 + sizeBytes;
+ for (int i = 0; i < size; ++i) {
+ writeLong(writeBuffer, offsetStart + i * offsetSize, offsets.get(i), offsetSize);
+ }
+ writeLong(writeBuffer, offsetStart + size * offsetSize, dataSize, offsetSize);
+ }
+
+ // Append a variant value to the variant builder. We need to insert the keys in the input
+ // variant
+ // into the current variant dictionary and rebuild it with new field ids. For scalar values in
+ // the
+ // input variant, we can directly copy the binary slice.
+ public void appendVariant(BinaryVariant v) {
+ appendVariantImpl(v.getValue(), v.getMetadata(), v.getPos());
+ }
+
+ private void appendVariantImpl(byte[] value, byte[] metadata, int pos) {
+ checkIndex(pos, value.length);
+ int basicType = value[pos] & BASIC_TYPE_MASK;
+ switch (basicType) {
+ case OBJECT:
+ handleObject(
+ value,
+ pos,
+ (size, idSize, offsetSize, idStart, offsetStart, dataStart) -> {
+ ArrayList<FieldEntry> fields = new ArrayList<>(size);
+ int start = writePos;
+ for (int i = 0; i < size; ++i) {
+ int id = readUnsigned(value, idStart + idSize * i, idSize);
+ int offset =
+ readUnsigned(
+ value, offsetStart + offsetSize * i, offsetSize);
+ int elementPos = dataStart + offset;
+ String key = getMetadataKey(metadata, id);
+ int newId = addKey(key);
+ fields.add(new FieldEntry(key, newId, writePos - start));
+ appendVariantImpl(value, metadata, elementPos);
+ }
+ finishWritingObject(start, fields);
+ return null;
+ });
+ break;
+ case ARRAY:
+ handleArray(
+ value,
+ pos,
+ (size, offsetSize, offsetStart, dataStart) -> {
+ ArrayList<Integer> offsets = new ArrayList<>(size);
+ int start = writePos;
+ for (int i = 0; i < size; ++i) {
+ int offset =
+ readUnsigned(
+ value, offsetStart + offsetSize * i, offsetSize);
+ int elementPos = dataStart + offset;
+ offsets.add(writePos - start);
+ appendVariantImpl(value, metadata, elementPos);
+ }
+ finishWritingArray(start, offsets);
+ return null;
+ });
+ break;
+ default:
+ shallowAppendVariantImpl(value, pos);
+ break;
+ }
+ }
+
+ private void shallowAppendVariantImpl(byte[] value, int pos) {
+ int size = valueSize(value, pos);
+ checkIndex(pos + size - 1, value.length);
+ checkCapacity(size);
+ System.arraycopy(value, pos, writeBuffer, writePos, size);
+ writePos += size;
+ }
+
+ private void checkCapacity(int additional) {
+ int required = writePos + additional;
+ if (required > writeBuffer.length) {
+ // Allocate a new buffer with a capacity of the next power of 2 of `required`.
+ int newCapacity = Integer.highestOneBit(required);
+ newCapacity = newCapacity < required ? newCapacity * 2 : newCapacity;
+ if (newCapacity > SIZE_LIMIT) {
+ throw VARIANT_SIZE_LIMIT_EXCEPTION;
+ }
+ byte[] newValue = new byte[newCapacity];
+ System.arraycopy(writeBuffer, 0, newValue, 0, writePos);
+ writeBuffer = newValue;
+ }
+ }
+
+ /**
+ * Temporarily store the information of a field. We need to collect all fields in an JSON
+ * object, sort them by their keys, and build the variant object in sorted order.
+ */
+ public static final class FieldEntry implements Comparable<FieldEntry> {
+ final String key;
+ final int id;
+ final int offset;
+
+ public FieldEntry(String key, int id, int offset) {
+ this.key = key;
+ this.id = id;
+ this.offset = offset;
+ }
+
+ FieldEntry withNewOffset(int newOffset) {
+ return new FieldEntry(key, id, newOffset);
+ }
+
+ @Override
+ public int compareTo(FieldEntry other) {
+ return key.compareTo(other.key);
+ }
+ }
+
+ private void buildJson(JsonParser parser) throws IOException {
+ JsonToken token = parser.currentToken();
+ if (token == null) {
+ throw new JsonParseException(parser, "Unexpected null token");
+ }
+ switch (token) {
+ case START_OBJECT:
+ {
+ ArrayList<FieldEntry> fields = new ArrayList<>();
+ int start = writePos;
+ while (parser.nextToken() != JsonToken.END_OBJECT) {
+ String key = parser.currentName();
+ parser.nextToken();
+ int id = addKey(key);
+ fields.add(new FieldEntry(key, id, writePos - start));
+ buildJson(parser);
+ }
+ finishWritingObject(start, fields);
+ break;
+ }
+ case START_ARRAY:
+ {
+ ArrayList<Integer> offsets = new ArrayList<>();
+ int start = writePos;
+ while (parser.nextToken() != JsonToken.END_ARRAY) {
+ offsets.add(writePos - start);
+ buildJson(parser);
+ }
+ finishWritingArray(start, offsets);
+ break;
+ }
+ case VALUE_STRING:
+ appendString(parser.getText());
+ break;
+ case VALUE_NUMBER_INT:
+ try {
+ appendNumeric(parser.getLongValue());
+ } catch (InputCoercionException ignored) {
+ // If the value doesn't fit any integer type, parse it as decimal or floating
+ // instead.
+ parseFloatingPoint(parser);
+ }
+ break;
+ case VALUE_NUMBER_FLOAT:
+ parseFloatingPoint(parser);
+ break;
+ case VALUE_TRUE:
+ appendBoolean(true);
+ break;
+ case VALUE_FALSE:
+ appendBoolean(false);
+ break;
+ case VALUE_NULL:
+ appendNull();
+ break;
+ default:
+ throw new JsonParseException(parser, "Unexpected token " + token);
+ }
+ }
+
+ // Choose the smallest unsigned integer type that can store `value`. It must be within
+ // `[0, U24_MAX]`.
+ private int getIntegerSize(int value) {
+ assert value >= 0 && value <= U24_MAX;
+ if (value <= U8_MAX) {
+ return 1;
+ }
+ if (value <= U16_MAX) {
+ return 2;
+ }
+ return U24_SIZE;
+ }
+
+ private void parseFloatingPoint(JsonParser parser) throws IOException {
+ if (!tryParseDecimal(parser.getText())) {
+ appendDouble(parser.getDoubleValue());
+ }
+ }
+
+ // Try to parse a JSON number as a decimal. Return whether the parsing succeeds. The input must
+ // only use the decimal format (an integer value with an optional '.' in it) and must not use
+ // scientific notation. It also must fit into the precision limitation of decimal types.
+ private boolean tryParseDecimal(String input) {
+ for (int i = 0; i < input.length(); ++i) {
+ char ch = input.charAt(i);
+ if (ch != '-' && ch != '.' && !(ch >= '0' && ch <= '9')) {
+ return false;
+ }
+ }
+ BigDecimal d = new BigDecimal(input);
+ if (d.scale() <= MAX_DECIMAL16_PRECISION && d.precision() <= MAX_DECIMAL16_PRECISION) {
+ appendDecimal(d);
+ return true;
+ }
+ return false;
+ }
+
+ // The write buffer in building the variant value. Its first `writePos` bytes has been written.
+ private byte[] writeBuffer = new byte[128];
+ private int writePos = 0;
+ // Map keys to a monotonically increasing id.
+ private final HashMap<String, Integer> dictionary = new HashMap<>();
+ // Store all keys in `dictionary` in the order of id.
+ private final ArrayList<byte[]> dictionaryKeys = new ArrayList<>();
+ private final boolean allowDuplicateKeys;
+}
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/BinaryVariantUtil.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/BinaryVariantUtil.java
new file mode 100644
index 000000000..591ca3ded
--- /dev/null
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/BinaryVariantUtil.java
@@ -0,0 +1,629 @@
+/*
+ * 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.flink.cdc.common.types.variant;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.types.variant.Variant.Type;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeFormatterBuilder;
+import java.util.Arrays;
+import java.util.Locale;
+
+/**
+ * Copy from <a
+ * href="https://github.com/apache/flink/blob/release-2.2.0/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantUtil.java">BinaryVariantUtil.java</a>
+ *
+ * <p>This class defines constants related to the variant format and provides functions for
+ * manipulating variant binaries.
+ *
+ * <p>A variant is made up of 2 binaries: value and metadata. A variant value consists of a one-byte
+ * header and a number of content bytes (can be zero). The header byte is divided into upper 6 bits
+ * (called "type info") and lower 2 bits (called "basic type"). The content format is explained in
+ * the below constants for all possible basic type and type info values.
+ *
+ * <p>The variant metadata includes a version id and a dictionary of distinct strings
+ * (case-sensitive). Its binary format is: - Version: 1-byte unsigned integer. The only acceptable
+ * value is 1 currently. - Dictionary size: 4-byte little-endian unsigned integer. The number of
+ * keys in the dictionary. - Offsets: (size + 1) * 4-byte little-endian unsigned integers.
+ * `offsets[i]` represents the starting position of string i, counting starting from the address of
+ * `offsets[0]`. Strings must be stored contiguously, so we don’t need to store the string size,
+ * instead, we compute it with `offset[i + 1] - offset[i]`. - UTF-8 string data.
+ */
+@Internal
+public class BinaryVariantUtil {
+ public static final int BASIC_TYPE_BITS = 2;
+ public static final int BASIC_TYPE_MASK = 0x3;
+ public static final int TYPE_INFO_MASK = 0x3F;
+ // The inclusive maximum value of the type info value. It is the size limit of `SHORT_STR`.
+ public static final int MAX_SHORT_STR_SIZE = 0x3F;
+
+ // Below is all possible basic type values.
+ // Primitive value. The type info value must be one of the values in the below section.
+ public static final int PRIMITIVE = 0;
+ // Short string value. The type info value is the string size, which must be in `[0,
+ // kMaxShortStrSize]`.
+ // The string content bytes directly follow the header byte.
+ public static final int SHORT_STR = 1;
+ // Object value. The content contains a size, a list of field ids, a list of field offsets, and
+ // the actual field data. The length of the id list is `size`, while the length of the offset
+ // list is `size + 1`, where the last offset represent the total size of the field data. The
+ // fields in an object must be sorted by the field name in alphabetical order. Duplicate field
+ // names in one object are not allowed.
+ // We use 5 bits in the type info to specify the integer type of the object header: it should
+ // be 0_b4_b3b2_b1b0 (MSB is 0), where:
+ // - b4 specifies the type of size. When it is 0/1, `size` is a little-endian 1/4-byte
+ // unsigned integer.
+ // - b3b2/b1b0 specifies the integer type of id and offset. When the 2 bits are 0/1/2, the
+ // list contains 1/2/3-byte little-endian unsigned integers.
+ public static final int OBJECT = 2;
+ // Array value. The content contains a size, a list of field offsets, and the actual element
+ // data. It is similar to an object without the id list. The length of the offset list
+ // is `size + 1`, where the last offset represent the total size of the element data.
+ // Its type info should be: 000_b2_b1b0:
+ // - b2 specifies the type of size.
+ // - b1b0 specifies the integer type of offset.
+ public static final int ARRAY = 3;
+
+ // Below is all possible type info values for `PRIMITIVE`.
+ // JSON Null value. Empty content.
+ public static final int NULL = 0;
+ // True value. Empty content.
+ public static final int TRUE = 1;
+ // False value. Empty content.
+ public static final int FALSE = 2;
+ // 1-byte little-endian signed integer.
+ public static final int INT1 = 3;
+ // 2-byte little-endian signed integer.
+ public static final int INT2 = 4;
+ // 4-byte little-endian signed integer.
+ public static final int INT4 = 5;
+ // 4-byte little-endian signed integer.
+ public static final int INT8 = 6;
+ // 8-byte IEEE double.
+ public static final int DOUBLE = 7;
+ // 4-byte decimal. Content is 1-byte scale + 4-byte little-endian signed integer.
+ public static final int DECIMAL4 = 8;
+ // 8-byte decimal. Content is 1-byte scale + 8-byte little-endian signed integer.
+ public static final int DECIMAL8 = 9;
+ // 16-byte decimal. Content is 1-byte scale + 16-byte little-endian signed integer.
+ public static final int DECIMAL16 = 10;
+ // Date value. Content is 4-byte little-endian signed integer that represents the number of days
+ // from the Unix epoch.
+ public static final int DATE = 11;
+ // TimestampLTZ value. Content is 8-byte little-endian signed integer that represents the number
+ // of microseconds elapsed since the Unix epoch, 1970-01-01 00:00:00 UTC. It is displayed to
+ // users in their local time zones and may be displayed differently depending on the execution
+ // environment.
+ public static final int TIMESTAMP_LTZ = 12;
+ // Timestamp value. It has the same content as `TIMESTAMP` but should always be interpreted
+ // as if the local time zone is UTC.
+ public static final int TIMESTAMP = 13;
+ // 4-byte IEEE float.
+ public static final int FLOAT = 14;
+ // Binary value. The content is (4-byte little-endian unsigned integer representing the binary
+ // size) + (size bytes of binary content).
+ public static final int BINARY = 15;
+ // Long string value. The content is (4-byte little-endian unsigned integer representing the
+ // string size) + (size bytes of string content).
+ public static final int LONG_STR = 16;
+
+ public static final byte VERSION = 1;
+ // The lower 4 bits of the first metadata byte contain the version.
+ public static final byte VERSION_MASK = 0x0F;
+
+ public static final int U8_MAX = 0xFF;
+ public static final int U16_MAX = 0xFFFF;
+ public static final int U24_MAX = 0xFFFFFF;
+ public static final int U24_SIZE = 3;
+ public static final int U32_SIZE = 4;
+
+ // Both variant value and variant metadata need to be no longer than 16MiB.
+ public static final int SIZE_LIMIT = U24_MAX + 1;
+
+ public static final int MAX_DECIMAL4_PRECISION = 9;
+ public static final int MAX_DECIMAL8_PRECISION = 18;
+ public static final int MAX_DECIMAL16_PRECISION = 38;
+
+ public static final int BINARY_SEARCH_THRESHOLD = 32;
+
+ public static final DateTimeFormatter TIMESTAMP_FORMATTER =
+ new DateTimeFormatterBuilder()
+ .append(DateTimeFormatter.ISO_LOCAL_DATE)
+ .appendLiteral('T')
+ .append(DateTimeFormatter.ISO_LOCAL_TIME)
+ .toFormatter(Locale.US);
+
+ public static final DateTimeFormatter TIMESTAMP_LTZ_FORMATTER =
+ new DateTimeFormatterBuilder()
+ .append(TIMESTAMP_FORMATTER)
+ .appendOffset("+HH:MM", "+00:00")
+ .toFormatter(Locale.US);
+
+ // Write the least significant `numBytes` bytes in `value` into `bytes[pos, pos + numBytes)` in
+ // little endian.
+ public static void writeLong(byte[] bytes, int pos, long value, int numBytes) {
+ for (int i = 0; i < numBytes; ++i) {
+ bytes[pos + i] = (byte) ((value >>> (8 * i)) & 0xFF);
+ }
+ }
+
+ public static byte primitiveHeader(int type) {
+ return (byte) (type << 2 | PRIMITIVE);
+ }
+
+ public static byte shortStrHeader(int size) {
+ return (byte) (size << 2 | SHORT_STR);
+ }
+
+ public static byte objectHeader(boolean largeSize, int idSize, int offsetSize) {
+ return (byte)
+ (((largeSize ? 1 : 0) << (BASIC_TYPE_BITS + 4))
+ | ((idSize - 1) << (BASIC_TYPE_BITS + 2))
+ | ((offsetSize - 1) << BASIC_TYPE_BITS)
+ | OBJECT);
+ }
+
+ public static byte arrayHeader(boolean largeSize, int offsetSize) {
+ return (byte)
+ (((largeSize ? 1 : 0) << (BASIC_TYPE_BITS + 2))
+ | ((offsetSize - 1) << BASIC_TYPE_BITS)
+ | ARRAY);
+ }
+
+ // An exception indicating that the variant value or metadata doesn't
+ static VariantTypeException malformedVariant() {
+ return new VariantTypeException("MALFORMED_VARIANT");
+ }
+
+ static VariantTypeException unknownPrimitiveTypeInVariant(int id) {
+ return new VariantTypeException("UNKNOWN_PRIMITIVE_TYPE_IN_VARIANT, id: " + id);
+ }
+
+ // An exception indicating that an external caller tried to call the Variant constructor with
+ // value or metadata exceeding the 16MiB size limit. We will never construct a Variant this
+ // large,
+ // so it should only be possible to encounter this exception when reading a Variant produced by
+ // another tool.
+ static VariantTypeException variantConstructorSizeLimit() {
+ return new VariantTypeException("VARIANT_CONSTRUCTOR_SIZE_LIMIT");
+ }
+
+ // Check the validity of an array index `pos`. Throw `MALFORMED_VARIANT` if it is out of bound,
+ // meaning that the variant is malformed.
+ static void checkIndex(int pos, int length) {
+ if (pos < 0 || pos >= length) {
+ throw malformedVariant();
+ }
+ }
+
+ // Read a little-endian signed long value from `bytes[pos, pos + numBytes)`.
+ static long readLong(byte[] bytes, int pos, int numBytes) {
+ checkIndex(pos, bytes.length);
+ checkIndex(pos + numBytes - 1, bytes.length);
+ long result = 0;
+ // All bytes except the most significant byte should be unsign-extended and shifted (so we
+ // need `& 0xFF`). The most significant byte should be sign-extended and is handled after
+ // the loop.
+ for (int i = 0; i < numBytes - 1; ++i) {
+ long unsignedByteValue = bytes[pos + i] & 0xFF;
+ result |= unsignedByteValue << (8 * i);
+ }
+ long signedByteValue = bytes[pos + numBytes - 1];
+ result |= signedByteValue << (8 * (numBytes - 1));
+ return result;
+ }
+
+ // Read a little-endian unsigned int value from `bytes[pos, pos + numBytes)`. The value must fit
+ // into a non-negative int (`[0, Integer.MAX_VALUE]`).
+ static int readUnsigned(byte[] bytes, int pos, int numBytes) {
+ checkIndex(pos, bytes.length);
+ checkIndex(pos + numBytes - 1, bytes.length);
+ int result = 0;
+ // Similar to the `readLong` loop, but all bytes should be unsign-extended.
+ for (int i = 0; i < numBytes; ++i) {
+ int unsignedByteValue = bytes[pos + i] & 0xFF;
+ result |= unsignedByteValue << (8 * i);
+ }
+ if (result < 0) {
+ throw malformedVariant();
+ }
+ return result;
+ }
+
+ public static int getTypeInfo(byte[] value, int pos) {
+ checkIndex(pos, value.length);
+ return (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ }
+
+ // Get the value type of variant value `value[pos...]`. It is only legal to call `get*` if
+ // `getType` returns this type (for example, it is only legal to call `getLong` if `getType`
+ // returns `Type.Long`).
+ // Throw `MALFORMED_VARIANT` if the variant is malformed.
+ public static Type getType(byte[] value, int pos) {
+ checkIndex(pos, value.length);
+ int basicType = value[pos] & BASIC_TYPE_MASK;
+ int typeInfo = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ switch (basicType) {
+ case SHORT_STR:
+ return Type.STRING;
+ case OBJECT:
+ return Type.OBJECT;
+ case ARRAY:
+ return Type.ARRAY;
+ default:
+ switch (typeInfo) {
+ case NULL:
+ return Type.NULL;
+ case TRUE:
+ case FALSE:
+ return Type.BOOLEAN;
+ case INT1:
+ return Type.TINYINT;
+ case INT2:
+ return Type.SMALLINT;
+ case INT4:
+ return Type.INT;
+ case INT8:
+ return Type.BIGINT;
+ case DOUBLE:
+ return Type.DOUBLE;
+ case DECIMAL4:
+ case DECIMAL8:
+ case DECIMAL16:
+ return Type.DECIMAL;
+ case DATE:
+ return Type.DATE;
+ case TIMESTAMP_LTZ:
+ return Type.TIMESTAMP_LTZ;
+ case TIMESTAMP:
+ return Type.TIMESTAMP;
+ case FLOAT:
+ return Type.FLOAT;
+ case BINARY:
+ return Type.BYTES;
+ case LONG_STR:
+ return Type.STRING;
+ default:
+ throw unknownPrimitiveTypeInVariant(typeInfo);
+ }
+ }
+ }
+
+ // Compute the size in bytes of the variant value `value[pos...]`. `value.length - pos` is an
+ // upper bound of the size, but the actual size can be smaller.
+ // Throw `MALFORMED_VARIANT` if the variant is malformed.
+ public static int valueSize(byte[] value, int pos) {
+ checkIndex(pos, value.length);
+ int basicType = value[pos] & BASIC_TYPE_MASK;
+ int typeInfo = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ switch (basicType) {
+ case SHORT_STR:
+ return 1 + typeInfo;
+ case OBJECT:
+ return handleObject(
+ value,
+ pos,
+ (size, idSize, offsetSize, idStart, offsetStart, dataStart) ->
+ dataStart
+ - pos
+ + readUnsigned(
+ value,
+ offsetStart + size * offsetSize,
+ offsetSize));
+ case ARRAY:
+ return handleArray(
+ value,
+ pos,
+ (size, offsetSize, offsetStart, dataStart) ->
+ dataStart
+ - pos
+ + readUnsigned(
+ value,
+ offsetStart + size * offsetSize,
+ offsetSize));
+ default:
+ switch (typeInfo) {
+ case NULL:
+ case TRUE:
+ case FALSE:
+ return 1;
+ case INT1:
+ return 2;
+ case INT2:
+ return 3;
+ case INT4:
+ case DATE:
+ case FLOAT:
+ return 5;
+ case INT8:
+ case DOUBLE:
+ case TIMESTAMP_LTZ:
+ case TIMESTAMP:
+ return 9;
+ case DECIMAL4:
+ return 6;
+ case DECIMAL8:
+ return 10;
+ case DECIMAL16:
+ return 18;
+ case BINARY:
+ case LONG_STR:
+ return 1 + U32_SIZE + readUnsigned(value, pos + 1, U32_SIZE);
+ default:
+ throw unknownPrimitiveTypeInVariant(typeInfo);
+ }
+ }
+ }
+
+ static VariantTypeException unexpectedType(Type type) {
+ return new VariantTypeException("Expect type to be " + type);
+ }
+
+ // Get a boolean value from variant value `value[pos...]`.
+ // Throw `MALFORMED_VARIANT` if the variant is malformed.
+ public static boolean getBoolean(byte[] value, int pos) {
+ checkIndex(pos, value.length);
+ int basicType = value[pos] & BASIC_TYPE_MASK;
+ int typeInfo = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ if (basicType != PRIMITIVE || (typeInfo != TRUE && typeInfo != FALSE)) {
+ throw unexpectedType(Type.BOOLEAN);
+ }
+ return typeInfo == TRUE;
+ }
+
+ // Get a long value from variant value `value[pos...]`.
+ // It is only legal to call it if `getType` returns one of `Type.LONG/DATE/TIMESTAMP/
+ // TIMESTAMP_LTZ`. If the type is `DATE`, the return value is guaranteed to fit into an int and
+ // represents the number of days from the Unix epoch.
+ // If the type is `TIMESTAMP/TIMESTAMP_LTZ`, the return value represents the number of
+ // microseconds from the Unix epoch.
+ public static long getLong(byte[] value, int pos) {
+ checkIndex(pos, value.length);
+ int basicType = value[pos] & BASIC_TYPE_MASK;
+ int typeInfo = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ String exceptionMessage = "Expect type to be LONG/DATE/TIMESTAMP/TIMESTAMP_LTZ";
+ if (basicType != PRIMITIVE) {
+ throw new IllegalStateException(exceptionMessage);
+ }
+ switch (typeInfo) {
+ case INT1:
+ return readLong(value, pos + 1, 1);
+ case INT2:
+ return readLong(value, pos + 1, 2);
+ case INT4:
+ case DATE:
+ return readLong(value, pos + 1, 4);
+ case INT8:
+ case TIMESTAMP_LTZ:
+ case TIMESTAMP:
+ return readLong(value, pos + 1, 8);
+ default:
+ throw new IllegalStateException(exceptionMessage);
+ }
+ }
+
+ // Get a double value from variant value `value[pos...]`.
+ // Throw `MALFORMED_VARIANT` if the variant is malformed.
+ public static double getDouble(byte[] value, int pos) {
+ checkIndex(pos, value.length);
+ int basicType = value[pos] & BASIC_TYPE_MASK;
+ int typeInfo = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ if (basicType != PRIMITIVE || typeInfo != DOUBLE) {
+ throw unexpectedType(Type.DOUBLE);
+ }
+ return Double.longBitsToDouble(readLong(value, pos + 1, 8));
+ }
+
+ // Check whether the precision and scale of the decimal are within the limit.
+ private static void checkDecimal(BigDecimal d, int maxPrecision) {
+ if (d.precision() > maxPrecision || d.scale() > maxPrecision) {
+ throw malformedVariant();
+ }
+ }
+
+ // Get a decimal value from variant value `value[pos...]`.
+ // Throw `MALFORMED_VARIANT` if the variant is malformed.
+ public static BigDecimal getDecimalWithOriginalScale(byte[] value, int pos) {
+ checkIndex(pos, value.length);
+ int basicType = value[pos] & BASIC_TYPE_MASK;
+ int typeInfo = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ if (basicType != PRIMITIVE) {
+ throw unexpectedType(Type.DECIMAL);
+ }
+ // Interpret the scale byte as unsigned. If it is a negative byte, the unsigned value must
+ // be greater than `MAX_DECIMAL16_PRECISION` and will trigger an error in `checkDecimal`.
+ int scale = value[pos + 1] & 0xFF;
+ BigDecimal result;
+ switch (typeInfo) {
+ case DECIMAL4:
+ result = BigDecimal.valueOf(readLong(value, pos + 2, 4), scale);
+ checkDecimal(result, MAX_DECIMAL4_PRECISION);
+ break;
+ case DECIMAL8:
+ result = BigDecimal.valueOf(readLong(value, pos + 2, 8), scale);
+ checkDecimal(result, MAX_DECIMAL8_PRECISION);
+ break;
+ case DECIMAL16:
+ checkIndex(pos + 17, value.length);
+ byte[] bytes = new byte[16];
+ // Copy the bytes reversely because the `BigInteger` constructor expects a
+ // big-endian representation.
+ for (int i = 0; i < 16; ++i) {
+ bytes[i] = value[pos + 17 - i];
+ }
+ result = new BigDecimal(new BigInteger(bytes), scale);
+ checkDecimal(result, MAX_DECIMAL16_PRECISION);
+ break;
+ default:
+ throw unexpectedType(Type.DECIMAL);
+ }
+ return result;
+ }
+
+ public static BigDecimal getDecimal(byte[] value, int pos) {
+ return getDecimalWithOriginalScale(value, pos).stripTrailingZeros();
+ }
+
+ // Get a float value from variant value `value[pos...]`.
+ // Throw `MALFORMED_VARIANT` if the variant is malformed.
+ public static float getFloat(byte[] value, int pos) {
+ checkIndex(pos, value.length);
+ int basicType = value[pos] & BASIC_TYPE_MASK;
+ int typeInfo = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ if (basicType != PRIMITIVE || typeInfo != FLOAT) {
+ throw unexpectedType(Type.FLOAT);
+ }
+ return Float.intBitsToFloat((int) readLong(value, pos + 1, 4));
+ }
+
+ // Get a binary value from variant value `value[pos...]`.
+ // Throw `MALFORMED_VARIANT` if the variant is malformed.
+ public static byte[] getBinary(byte[] value, int pos) {
+ checkIndex(pos, value.length);
+ int basicType = value[pos] & BASIC_TYPE_MASK;
+ int typeInfo = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ if (basicType != PRIMITIVE || typeInfo != BINARY) {
+ throw unexpectedType(Type.BYTES);
+ }
+ int start = pos + 1 + U32_SIZE;
+ int length = readUnsigned(value, pos + 1, U32_SIZE);
+ checkIndex(start + length - 1, value.length);
+ return Arrays.copyOfRange(value, start, start + length);
+ }
+
+ // Get a string value from variant value `value[pos...]`.
+ // Throw `MALFORMED_VARIANT` if the variant is malformed.
+ public static String getString(byte[] value, int pos) {
+ checkIndex(pos, value.length);
+ int basicType = value[pos] & BASIC_TYPE_MASK;
+ int typeInfo = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ if (basicType == SHORT_STR || (basicType == PRIMITIVE && typeInfo == LONG_STR)) {
+ int start;
+ int length;
+ if (basicType == SHORT_STR) {
+ start = pos + 1;
+ length = typeInfo;
+ } else {
+ start = pos + 1 + U32_SIZE;
+ length = readUnsigned(value, pos + 1, U32_SIZE);
+ }
+ checkIndex(start + length - 1, value.length);
+ return new String(value, start, length);
+ }
+ throw unexpectedType(Type.STRING);
+ }
+
+ /** 1. */
+ public interface ObjectHandler<T> {
+ /**
+ * @param size Number of object fields.
+ * @param idSize The integer size of the field id list.
+ * @param offsetSize The integer size of the offset list.
+ * @param idStart The starting index of the field id list in the variant value array.
+ * @param offsetStart The starting index of the offset list in the variant value array.
+ * @param dataStart The starting index of field data in the variant value array.
+ */
+ T apply(int size, int idSize, int offsetSize, int idStart, int offsetStart, int dataStart);
+ }
+
+ // A helper function to access a variant object. It provides `handler` with its required
+ // parameters and returns what it returns.
+ public static <T> T handleObject(byte[] value, int pos, ObjectHandler<T> handler) {
+ checkIndex(pos, value.length);
+ int basicType = value[pos] & BASIC_TYPE_MASK;
+ int typeInfo = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ if (basicType != OBJECT) {
+ throw unexpectedType(Type.OBJECT);
+ }
+ // Refer to the comment of the `OBJECT` constant for the details of the object header
+ // encoding. Suppose `typeInfo` has a bit representation of 0_b4_b3b2_b1b0, the following
+ // line extracts b4 to determine whether the object uses a 1/4-byte size.
+ boolean largeSize = ((typeInfo >> 4) & 0x1) != 0;
+ int sizeBytes = (largeSize ? U32_SIZE : 1);
+ int size = readUnsigned(value, pos + 1, sizeBytes);
+ // Extracts b3b2 to determine the integer size of the field id list.
+ int idSize = ((typeInfo >> 2) & 0x3) + 1;
+ // Extracts b1b0 to determine the integer size of the offset list.
+ int offsetSize = (typeInfo & 0x3) + 1;
+ int idStart = pos + 1 + sizeBytes;
+ int offsetStart = idStart + size * idSize;
+ int dataStart = offsetStart + (size + 1) * offsetSize;
+ return handler.apply(size, idSize, offsetSize, idStart, offsetStart, dataStart);
+ }
+
+ /** 1. */
+ public interface ArrayHandler<T> {
+ /**
+ * @param size Number of array elements.
+ * @param offsetSize The integer size of the offset list.
+ * @param offsetStart The starting index of the offset list in the variant value array.
+ * @param dataStart The starting index of element data in the variant value array.
+ */
+ T apply(int size, int offsetSize, int offsetStart, int dataStart);
+ }
+
+ // A helper function to access a variant array.
+ public static <T> T handleArray(byte[] value, int pos, ArrayHandler<T> handler) {
+ checkIndex(pos, value.length);
+ int basicType = value[pos] & BASIC_TYPE_MASK;
+ int typeInfo = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ if (basicType != ARRAY) {
+ throw unexpectedType(Type.ARRAY);
+ }
+ // Refer to the comment of the `ARRAY` constant for the details of the object header
+ // encoding.
+ // Suppose `typeInfo` has a bit representation of 000_b2_b1b0, the following line extracts
+ // b2 to determine whether the object uses a 1/4-byte size.
+ boolean largeSize = ((typeInfo >> 2) & 0x1) != 0;
+ int sizeBytes = (largeSize ? U32_SIZE : 1);
+ int size = readUnsigned(value, pos + 1, sizeBytes);
+ // Extracts b1b0 to determine the integer size of the offset list.
+ int offsetSize = (typeInfo & 0x3) + 1;
+ int offsetStart = pos + 1 + sizeBytes;
+ int dataStart = offsetStart + (size + 1) * offsetSize;
+ return handler.apply(size, offsetSize, offsetStart, dataStart);
+ }
+
+ // Get a key at `id` in the variant metadata.
+ // Throw `MALFORMED_VARIANT` if the variant is malformed. An out-of-bound `id` is also
+ // considered a malformed variant because it is read from the corresponding variant value.
+ public static String getMetadataKey(byte[] metadata, int id) {
+ checkIndex(0, metadata.length);
+ // Extracts the highest 2 bits in the metadata header to determine the integer size of the
+ // offset list.
+ int offsetSize = ((metadata[0] >> 6) & 0x3) + 1;
+ int dictSize = readUnsigned(metadata, 1, offsetSize);
+ if (id >= dictSize) {
+ throw malformedVariant();
+ }
+ // There are a header byte, a `dictSize` with `offsetSize` bytes, and `(dictSize + 1)`
+ // offsets before the string data.
+ int stringStart = 1 + (dictSize + 2) * offsetSize;
+ int offset = readUnsigned(metadata, 1 + (id + 1) * offsetSize, offsetSize);
+ int nextOffset = readUnsigned(metadata, 1 + (id + 2) * offsetSize, offsetSize);
+ if (offset > nextOffset) {
+ throw malformedVariant();
+ }
+ checkIndex(stringStart + nextOffset - 1, metadata.length);
+ return new String(metadata, stringStart + offset, nextOffset - offset);
+ }
+}
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/Variant.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/Variant.java
new file mode 100644
index 000000000..4f6ab8c6e
--- /dev/null
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/Variant.java
@@ -0,0 +1,227 @@
+/*
+ * 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.flink.cdc.common.types.variant;
+
+import org.apache.flink.cdc.common.annotation.PublicEvolving;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+
+/**
+ * Copy from <a
+ * href="https://github.com/apache/flink/blob/release-2.2.0/flink-core/src/main/java/org/apache/flink/types/variant/Variant.java">Variant.java</a>
+ *
+ * <p>Variant represent a semi-structured data.
+ */
+@PublicEvolving
+public interface Variant {
+
+ /** Returns true if the variant is a primitive typed value, such as INT, DOUBLE, STRING, etc. */
+ boolean isPrimitive();
+
+ /** Returns true if this variant is an Array, false otherwise. */
+ boolean isArray();
+
+ /** Returns true if this variant is an Object, false otherwise. */
+ boolean isObject();
+
+ /** Check If this variant is null. */
+ boolean isNull();
+
+ /** Get the type of variant. */
+ Type getType();
+
+ /**
+ * Get the scalar value of variant as boolean, if the variant type is {@link Type#BOOLEAN}.
+ *
+ * @throws VariantTypeException If this variant is not a scalar value or is not {@link
+ * Type#BOOLEAN}.
+ */
+ boolean getBoolean() throws VariantTypeException;
+
+ /**
+ * Get the scalar value of variant as byte, if the variant type is {@link Type#TINYINT}.
+ *
+ * @throws VariantTypeException If this variant is not a scalar value or is not {@link
+ * Type#TINYINT}.
+ */
+ byte getByte() throws VariantTypeException;
+
+ /**
+ * Get the scalar value of variant as short, if the variant type is {@link Type#SMALLINT}.
+ *
+ * @throws VariantTypeException If this variant is not a scalar value or is not {@link
+ * Type#SMALLINT}.
+ */
+ short getShort() throws VariantTypeException;
+
+ /**
+ * Get the scalar value of variant as int, if the variant type is {@link Type#INT}.
+ *
+ * @throws VariantTypeException if this variant is not a scalar value or is not {@link
+ * Type#INT}.
+ */
+ int getInt() throws VariantTypeException;
+
+ /**
+ * Get the scalar value of variant as long, if the variant type is {@link Type#BIGINT}.
+ *
+ * @throws VariantTypeException If this variant is not a scalar value or is not {@link
+ * Type#BIGINT}.
+ */
+ long getLong() throws VariantTypeException;
+
+ /**
+ * Get the scalar value of variant as float, if the variant type is {@link Type#FLOAT}.
+ *
+ * @throws VariantTypeException If this variant is not a scalar value or is not {@link
+ * Type#FLOAT}.
+ */
+ float getFloat() throws VariantTypeException;
+
+ /**
+ * Get the scalar value of variant as BigDecimal, if the variant type is {@link Type#DECIMAL}.
+ *
+ * @throws VariantTypeException If this variant is not a scalar value or is not {@link
+ * Type#DECIMAL}.
+ */
+ BigDecimal getDecimal() throws VariantTypeException;
+
+ /**
+ * Get the scalar value of variant as double, if the variant type is {@link Type#DOUBLE}.
+ *
+ * @throws VariantTypeException If this variant is not a scalar value or is not {@link
+ * Type#DOUBLE}.
+ */
+ double getDouble() throws VariantTypeException;
+
+ /**
+ * Get the scalar value of variant as string, if the variant type is {@link Type#STRING}.
+ *
+ * @throws VariantTypeException If this variant is not a scalar value or is not {@link
+ * Type#STRING}.
+ */
+ String getString() throws VariantTypeException;
+
+ /**
+ * Get the scalar value of variant as LocalDate, if the variant type is {@link Type#DATE}.
+ *
+ * @throws VariantTypeException If this variant is not a scalar value or is not {@link
+ * Type#DATE}.
+ */
+ LocalDate getDate() throws VariantTypeException;
+
+ /**
+ * Get the scalar value of variant as LocalDateTime, if the variant type is {@link
+ * Type#TIMESTAMP}.
+ *
+ * @throws VariantTypeException If this variant is not a scalar value or is not {@link
+ * Type#TIMESTAMP}.
+ */
+ LocalDateTime getDateTime() throws VariantTypeException;
+
+ /**
+ * Get the scalar value of variant as Instant, if the variant type is {@link Type#TIMESTAMP}.
+ *
+ * @throws VariantTypeException If this variant is not a scalar value or is not {@link
+ * Type#TIMESTAMP}.
+ */
+ Instant getInstant() throws VariantTypeException;
+
+ /**
+ * Get the scalar value of variant as byte array, if the variant type is {@link Type#BYTES}.
+ *
+ * @throws VariantTypeException If this variant is not a scalar value or is not {@link
+ * Type#BYTES}.
+ */
+ byte[] getBytes() throws VariantTypeException;
+
+ /**
+ * Get the scalar value of variant.
+ *
+ * @throws VariantTypeException If this variant is not a scalar value.
+ */
+ Object get() throws VariantTypeException;
+
+ /**
+ * Get the scalar value of variant.
+ *
+ * @throws VariantTypeException If this variant is not a scalar value.
+ */
+ <T> T getAs() throws VariantTypeException;
+
+ /**
+ * Get the array size of variant.
+ *
+ * @throws VariantTypeException If this variant is not an array.
+ */
+ int arraySize() throws VariantTypeException;
+
+ /**
+ * Access value of the specified element of an array variant. If index is out of range, null is
+ * returned.
+ *
+ * <p>NOTE: if the element value has been explicitly set as <code>null</code> (which is
+ * different from removal!), a variant that @{@link Variant#isNull()} returns true will be
+ * returned, not null.
+ *
+ * @throws VariantTypeException If this variant is not an array.
+ */
+ Variant getElement(int index) throws VariantTypeException;
+
+ /**
+ * Access value of the specified field of an object variant. If there is no field with the
+ * specified name, null is returned.
+ *
+ * <p>NOTE: if the property value has been explicitly set as <code>null</code>, a variant
+ * that @{@link Variant#isNull()} returns true will be returned, not null.
+ *
+ * @throws VariantTypeException If this variant is not an object.
+ */
+ Variant getField(String fieldName) throws VariantTypeException;
+
+ /** Parses the variant to json. */
+ String toJson();
+
+ /** The type of variant. */
+ @PublicEvolving
+ enum Type {
+ OBJECT,
+ ARRAY,
+ NULL,
+ BOOLEAN,
+ TINYINT,
+ SMALLINT,
+ INT,
+ BIGINT,
+ FLOAT,
+ DOUBLE,
+ DECIMAL,
+ STRING,
+ DATE,
+ TIMESTAMP,
+ TIMESTAMP_LTZ,
+ BYTES
+ }
+
+ static VariantBuilder newBuilder() {
+ return new BinaryVariantBuilder();
+ }
+}
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/VariantBuilder.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/VariantBuilder.java
new file mode 100644
index 000000000..8b3a950fd
--- /dev/null
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/VariantBuilder.java
@@ -0,0 +1,109 @@
+/*
+ * 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.flink.cdc.common.types.variant;
+
+import org.apache.flink.cdc.common.annotation.PublicEvolving;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+
+/**
+ * Copy from <a
+ * href="https://github.com/apache/flink/blob/release-2.2.0/flink-core/src/main/java/org/apache/flink/types/variant/VariantBuilder.java">VariantBuilder.java</a>
+ *
+ * <p>Builder for variants.
+ */
+@PublicEvolving
+public interface VariantBuilder {
+
+ /** Create a variant from a byte. */
+ Variant of(byte b);
+
+ /** Create a variant from a short. */
+ Variant of(short s);
+
+ /** Create a variant from a int. */
+ Variant of(int i);
+
+ /** Create a variant from a long. */
+ Variant of(long l);
+
+ /** Create a variant from a string. */
+ Variant of(String s);
+
+ /** Create a variant from a double. */
+ Variant of(double d);
+
+ /** Create a variant from a float. */
+ Variant of(float f);
+
+ /** Create a variant from a byte array. */
+ Variant of(byte[] bytes);
+
+ /** Create a variant from a boolean. */
+ Variant of(boolean b);
+
+ /** Create a variant from a BigDecimal. */
+ Variant of(BigDecimal bigDecimal);
+
+ /** Create a variant from an Instant. */
+ Variant of(Instant instant);
+
+ /** Create a variant from a LocalDate. */
+ Variant of(LocalDate localDate);
+
+ /** Create a variant from a LocalDateTime. */
+ Variant of(LocalDateTime localDateTime);
+
+ /** Create a variant of null. */
+ Variant ofNull();
+
+ /** Get the builder of a variant object. */
+ VariantObjectBuilder object();
+
+ /** Get the builder of a variant object. */
+ VariantObjectBuilder object(boolean allowDuplicateKeys);
+
+ /** Get the builder for a variant array. */
+ VariantArrayBuilder array();
+
+ /** Builder for a variant object. */
+ @PublicEvolving
+ interface VariantObjectBuilder {
+
+ /** Add a field to the object. */
+ VariantObjectBuilder add(String key, Variant value);
+
+ /** Build the variant object. */
+ Variant build();
+ }
+
+ /** Builder for a variant array. */
+ @PublicEvolving
+ interface VariantArrayBuilder {
+
+ /** Add a value to the array. */
+ VariantArrayBuilder add(Variant value);
+
+ /** Build the variant array. */
+ Variant build();
+ }
+}
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/VariantTypeException.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/VariantTypeException.java
new file mode 100644
index 000000000..30414f45c
--- /dev/null
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/variant/VariantTypeException.java
@@ -0,0 +1,36 @@
+/*
+ * 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.flink.cdc.common.types.variant;
+
+import org.apache.flink.cdc.common.annotation.PublicEvolving;
+
+/**
+ * Copy from <a
+ * href="https://github.com/apache/flink/blob/release-2.2.0/flink-core/src/main/java/org/apache/flink/types/variant/VariantTypeException.java">VariantTypeException.java</a>
+ *
+ * <p>Exception thrown when a variant value is not of the expected type.
+ */
+@PublicEvolving
+public class VariantTypeException extends RuntimeException {
+ private static final long serialVersionUID = 1L;
+
+ public VariantTypeException(String message) {
+ super(message);
+ }
+}
diff --git a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/types/variant/BinaryVariantInternalBuilderTest.java b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/types/variant/BinaryVariantInternalBuilderTest.java
new file mode 100644
index 000000000..b3667d0e5
--- /dev/null
+++ b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/types/variant/BinaryVariantInternalBuilderTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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.flink.cdc.common.types.variant;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class BinaryVariantInternalBuilderTest {
+
+ @Test
+ void testParseScalarJson() throws IOException {
+ assertThat(BinaryVariantInternalBuilder.parseJson("1", false).getByte())
+ .isEqualTo((byte) 1);
+ short s = (short) (Byte.MAX_VALUE + 1L);
+ assertThat(BinaryVariantInternalBuilder.parseJson(String.valueOf(s), false).getShort())
+ .isEqualTo(s);
+ int i = (int) (Short.MAX_VALUE + 1L);
+ assertThat(BinaryVariantInternalBuilder.parseJson(String.valueOf(i), false).getInt())
+ .isEqualTo(i);
+ long l = Integer.MAX_VALUE + 1L;
+ assertThat(BinaryVariantInternalBuilder.parseJson(String.valueOf(l), false).getLong())
+ .isEqualTo(l);
+
+ BigDecimal bigDecimal = BigDecimal.valueOf(Long.MAX_VALUE).add(BigDecimal.ONE);
+ assertThat(
+ BinaryVariantInternalBuilder.parseJson(bigDecimal.toPlainString(), false)
+ .getDecimal())
+ .isEqualTo(bigDecimal);
+
+ assertThat(BinaryVariantInternalBuilder.parseJson("1.123", false).getDecimal())
+ .isEqualTo(BigDecimal.valueOf(1.123));
+ assertThat(
+ BinaryVariantInternalBuilder.parseJson(
+ String.valueOf(Double.MAX_VALUE), false)
+ .getDouble())
+ .isEqualTo(Double.MAX_VALUE);
+
+ assertThat(BinaryVariantInternalBuilder.parseJson("\"hello\"", false).getString())
+ .isEqualTo("hello");
+
+ assertThat(BinaryVariantInternalBuilder.parseJson("true", false).getBoolean()).isTrue();
+
+ assertThat(BinaryVariantInternalBuilder.parseJson("false", false).getBoolean()).isFalse();
+
+ assertThat(BinaryVariantInternalBuilder.parseJson("null", false).isNull()).isTrue();
+ }
+
+ @Test
+ void testParseJsonArray() throws IOException {
+ BinaryVariant variant = BinaryVariantInternalBuilder.parseJson("[]", false);
+ assertThat(variant.getElement(0)).isNull();
+
+ variant = BinaryVariantInternalBuilder.parseJson("[1,\"hello\",3.1, null]", false);
+ assertThat(variant.getElement(0).getByte()).isEqualTo((byte) 1);
+ assertThat(variant.getElement(1).getString()).isEqualTo("hello");
+ assertThat(variant.getElement(2).getDecimal()).isEqualTo(BigDecimal.valueOf(3.1));
+ assertThat(variant.getElement(3).isNull()).isTrue();
+
+ variant = BinaryVariantInternalBuilder.parseJson("[1,[\"hello\",[3.1]]]", false);
+ assertThat(variant.getElement(0).getByte()).isEqualTo((byte) 1);
+ assertThat(variant.getElement(1).getElement(0).getString()).isEqualTo("hello");
+ assertThat(variant.getElement(1).getElement(1).getElement(0).getDecimal())
+ .isEqualTo(BigDecimal.valueOf(3.1));
+ }
+
+ @Test
+ void testParseJsonObject() throws IOException {
+ BinaryVariant variant = BinaryVariantInternalBuilder.parseJson("{}", false);
+ assertThat(variant.getField("a")).isNull();
+
+ variant =
+ BinaryVariantInternalBuilder.parseJson(
+ "{\"a\":1,\"b\":\"hello\",\"c\":3.1}", false);
+
+ assertThat(variant.getField("a").getByte()).isEqualTo((byte) 1);
+ assertThat(variant.getField("b").getString()).isEqualTo("hello");
+ assertThat(variant.getField("c").getDecimal()).isEqualTo(BigDecimal.valueOf(3.1));
+
+ variant =
+ BinaryVariantInternalBuilder.parseJson(
+ "{\"a\":1,\"b\":{\"c\":\"hello\",\"d\":[3.1]}}", false);
+ assertThat(variant.getField("a").getByte()).isEqualTo((byte) 1);
+ assertThat(variant.getField("b").getField("c").getString()).isEqualTo("hello");
+ assertThat(variant.getField("b").getField("d").getElement(0).getDecimal())
+ .isEqualTo(BigDecimal.valueOf(3.1));
+
+ assertThatThrownBy(
+ () ->
+ BinaryVariantInternalBuilder.parseJson(
+ "{\"k1\":1,\"k1\":2,\"k2\":1.5}", false))
+ .isInstanceOf(VariantTypeException.class)
+ .hasMessage("VARIANT_DUPLICATE_KEY");
+
+ variant = BinaryVariantInternalBuilder.parseJson("{\"k1\":1,\"k1\":2,\"k2\":1.5}", true);
+ assertThat(variant.getField("k1").getByte()).isEqualTo((byte) 2);
+ assertThat(variant.getField("k2").getDecimal()).isEqualTo(BigDecimal.valueOf(1.5));
+ }
+}
diff --git a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/types/variant/BinaryVariantTest.java b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/types/variant/BinaryVariantTest.java
new file mode 100644
index 000000000..41726d14a
--- /dev/null
+++ b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/types/variant/BinaryVariantTest.java
@@ -0,0 +1,265 @@
+/*
+ * 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.flink.cdc.common.types.variant;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.temporal.ChronoUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class BinaryVariantTest {
+
+ private BinaryVariantBuilder builder;
+
+ @BeforeEach
+ void setUp() {
+ builder = new BinaryVariantBuilder();
+ }
+
+ @Test
+ void testScalarVariant() {
+
+ assertThat(builder.of((byte) 10).isPrimitive()).isTrue();
+ assertThat(builder.of((byte) 10).isNull()).isFalse();
+ assertThat(builder.of((byte) 10).isArray()).isFalse();
+ assertThat(builder.of((byte) 10).isObject()).isFalse();
+ assertThat(builder.of((byte) 10).getType()).isEqualTo(Variant.Type.TINYINT);
+
+ assertThat(builder.of((byte) 10).getByte()).isEqualTo((byte) 10);
+ assertThat(builder.of((byte) 10).get()).isEqualTo((byte) 10);
+ assertThat((byte) builder.of((byte) 10).getAs()).isEqualTo((byte) 10);
+
+ assertThat(builder.of((short) 10).getShort()).isEqualTo((short) 10);
+ assertThat(builder.of((short) 10).get()).isEqualTo((short) 10);
+
+ assertThat(builder.of(10).getInt()).isEqualTo(10);
+ assertThat(builder.of(10).get()).isEqualTo(10);
+
+ assertThat(builder.of(10L).getLong()).isEqualTo(10L);
+ assertThat(builder.of(10L).get()).isEqualTo(10L);
+
+ assertThat(builder.of(10.0).getDouble()).isEqualTo(10.0d);
+ assertThat(builder.of(10.0).get()).isEqualTo(10.0d);
+
+ assertThat(builder.of(10.0f).getFloat()).isEqualTo(10.0f);
+ assertThat(builder.of(10.0f).get()).isEqualTo(10.0f);
+
+ assertThat(builder.of("hello").getString()).isEqualTo("hello");
+ assertThat(builder.of("hello").get()).isEqualTo("hello");
+
+ assertThat(builder.of("hello".getBytes()).getBytes()).isEqualTo("hello".getBytes());
+ assertThat(builder.of("hello".getBytes()).get()).isEqualTo("hello".getBytes());
+
+ assertThat(builder.of(true).getBoolean()).isTrue();
+ assertThat(builder.of(true).get()).isEqualTo(true);
+
+ assertThat(builder.of(BigDecimal.valueOf(100)).getDecimal())
+ .isEqualByComparingTo(BigDecimal.valueOf(100));
+ assertThat((BigDecimal) builder.of(BigDecimal.valueOf(100)).get())
+ .isEqualByComparingTo(BigDecimal.valueOf(100));
+
+ Instant instant = Instant.now().truncatedTo(ChronoUnit.MICROS);
+ assertThat(builder.of(instant).getInstant()).isEqualTo(instant);
+ assertThat(builder.of(instant).get()).isEqualTo(instant);
+
+ LocalDateTime localDateTime = LocalDateTime.now().truncatedTo(ChronoUnit.MICROS);
+ assertThat(builder.of(localDateTime).getDateTime()).isEqualTo(localDateTime);
+ assertThat(builder.of(localDateTime).get()).isEqualTo(localDateTime);
+
+ LocalDate localDate = LocalDate.now();
+ assertThat(builder.of(localDate).getDate()).isEqualTo(localDate);
+ assertThat(builder.of(localDate).get()).isEqualTo(localDate);
+
+ assertThat(builder.ofNull().get()).isEqualTo(null);
+ assertThat(builder.ofNull().isNull()).isTrue();
+ }
+
+ @Test
+ void testArrayVariant() {
+ Instant now = Instant.now().truncatedTo(ChronoUnit.MICROS);
+ Variant variant =
+ builder.array()
+ .add(builder.of(1))
+ .add(builder.of("hello"))
+ .add(builder.of(now))
+ .add(builder.array().add(builder.of("hello2")).add(builder.of(10f)).build())
+ .add(builder.ofNull())
+ .build();
+
+ assertThat(variant.isArray()).isTrue();
+ assertThat(variant.isPrimitive()).isFalse();
+ assertThat(variant.isObject()).isFalse();
+ assertThat(variant.getType()).isEqualTo(Variant.Type.ARRAY);
+
+ assertThat(variant.getElement(-1)).isNull();
+ assertThat(variant.getElement(0).getInt()).isEqualTo(1);
+ assertThat(variant.getElement(1).getString()).isEqualTo("hello");
+ assertThat(variant.getElement(2).getInstant()).isEqualTo(now);
+ assertThat(variant.getElement(3).getElement(0).getString()).isEqualTo("hello2");
+ assertThat(variant.getElement(3).getElement(1).getFloat()).isEqualTo(10f);
+ assertThat(variant.getElement(4).isNull()).isTrue();
+ assertThat(variant.getElement(5)).isNull();
+ }
+
+ @Test
+ void testObjectVariant() {
+ Variant variant =
+ builder.object()
+ .add(
+ "list",
+ builder.array().add(builder.of("hello")).add(builder.of(1)).build())
+ .add(
+ "object",
+ builder.object()
+ .add("ss", builder.of((short) 1))
+ .add("ff", builder.of(10.0f))
+ .build())
+ .add("bb", builder.of((byte) 10))
+ .build();
+
+ assertThat(variant.isArray()).isFalse();
+ assertThat(variant.isPrimitive()).isFalse();
+ assertThat(variant.isObject()).isTrue();
+ assertThat(variant.getType()).isEqualTo(Variant.Type.OBJECT);
+
+ assertThat(variant.getField("list").isArray()).isTrue();
+ assertThat(variant.getField("list").getElement(0).getString()).isEqualTo("hello");
+ assertThat(variant.getField("list").getElement(1).getInt()).isEqualTo(1);
+
+ assertThat(variant.getField("object").isObject()).isTrue();
+ assertThat(variant.getField("object").getField("ss").getShort()).isEqualTo((short) 1);
+ assertThat(variant.getField("object").getField("ff").getFloat()).isEqualTo((10.0f));
+
+ assertThat(variant.getField("bb").getByte()).isEqualTo((byte) 10);
+ assertThat(variant.getField("non_exist")).isNull();
+
+ BinaryVariantBuilder.VariantObjectBuilder objectBuilder = builder.object();
+
+ for (int i = 0; i < 100; i++) {
+ objectBuilder.add(String.valueOf(i), builder.of(i));
+ }
+ variant = objectBuilder.build();
+ for (int i = 0; i < 100; i++) {
+ assertThat(variant.getField(String.valueOf(i)).getInt()).isEqualTo(i);
+ }
+ }
+
+ @Test
+ void testDuplicatedKeyObjectVariant() {
+ assertThatThrownBy(
+ () ->
+ builder.object(false)
+ .add("k", builder.of((byte) 10))
+ .add("k", builder.of("hello"))
+ .build())
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("VARIANT_DUPLICATE_KEY");
+
+ Variant variant =
+ builder.object(true)
+ .add("k", builder.of((byte) 10))
+ .add("k", builder.of("hello"))
+ .add("k1", builder.of(10))
+ .build();
+
+ assertThat(variant.getField("k").getString()).isEqualTo("hello");
+ assertThat(variant.getField("k1").getInt()).isEqualTo(10);
+ }
+
+ @Test
+ void testToJsonScalar() {
+ Instant instant = Instant.EPOCH;
+ LocalDateTime localDateTime = LocalDateTime.of(2000, 1, 1, 0, 0);
+ LocalDate localDate = LocalDate.of(2000, 1, 1);
+
+ assertThat(builder.of((byte) 1).toJson()).isEqualTo("1");
+ assertThat(builder.of((short) 1).toJson()).isEqualTo("1");
+ assertThat(builder.of(1L).toJson()).isEqualTo("1");
+ assertThat(builder.of(1).toJson()).isEqualTo("1");
+ assertThat(builder.of("hello").toJson()).isEqualTo("\"hello\"");
+ assertThat(builder.of(true).toJson()).isEqualTo("true");
+ assertThat(builder.of(10.0f).toJson()).isEqualTo("10.0");
+ assertThat(builder.of(10.0d).toJson()).isEqualTo("10.0");
+ assertThat(builder.of(BigDecimal.valueOf(100)).toJson()).isEqualTo("100");
+ assertThat(builder.of(instant).toJson()).isEqualTo("\"1970-01-01T00:00:00+00:00\"");
+ assertThat(builder.of(localDateTime).toJson()).isEqualTo("\"2000-01-01T00:00:00\"");
+ assertThat(builder.of(localDate).toJson()).isEqualTo("\"2000-01-01\"");
+ assertThat(builder.of("hello".getBytes()).toJson()).isEqualTo("\"aGVsbG8=\"");
+ assertThat(builder.ofNull().toJson()).isEqualTo("null");
+ }
+
+ @Test
+ void testToJsonNested() {
+ Variant variant =
+ builder.object()
+ .add(
+ "list",
+ builder.array().add(builder.of("hello")).add(builder.of(1)).build())
+ .add(
+ "object",
+ builder.object()
+ .add("ss", builder.of((short) 1))
+ .add("ff", builder.of(10.0f))
+ .build())
+ .build();
+
+ String json = variant.toJson();
+ assertThat(json)
+ .isEqualTo("{" + "\"list\":[\"hello\",1]," + "\"object\":{\"ff\":10.0,\"ss\":1}}");
+ }
+
+ @Test
+ void testVariantException() {
+ assertThatThrownBy(() -> new BinaryVariant(new byte[0], new byte[0]))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("MALFORMED_VARIANT");
+
+ byte[] meta = new byte[1];
+ meta[0] = (byte) 0x02;
+ assertThatThrownBy(() -> new BinaryVariant(new byte[1], meta))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("MALFORMED_VARIANT");
+
+ byte[] oversize = new byte[0xFFFFFF + 2];
+ meta[0] = (byte) 0x01;
+ oversize[0] = (byte) 0x01;
+ assertThatThrownBy(() -> new BinaryVariant(oversize, meta))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("VARIANT_CONSTRUCTOR_SIZE_LIMIT");
+
+ assertThatThrownBy(() -> new BinaryVariant(new byte[1], oversize))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("VARIANT_CONSTRUCTOR_SIZE_LIMIT");
+ }
+
+ @Test
+ void testGetThrowException() {
+ Variant variant = builder.of(10f);
+ assertThatThrownBy(variant::getDouble)
+ .isInstanceOf(VariantTypeException.class)
+ .hasMessage("Expected type DOUBLE but got FLOAT");
+ }
+}
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/InternalSerializers.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/InternalSerializers.java
index c0c9f3ec4..8685f8333 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/InternalSerializers.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/InternalSerializers.java
@@ -87,6 +87,8 @@ public class InternalSerializers {
case MAP:
MapType mapType = (MapType) type;
return new MapDataSerializer(mapType.getKeyType(), mapType.getValueType());
+ case VARIANT:
+ return VariantSerializer.INSTANCE;
default:
throw new UnsupportedOperationException(
"Unsupported type '" + type + "' to get internal serializer");
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/VariantSerializer.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/VariantSerializer.java
new file mode 100644
index 000000000..689f900b3
--- /dev/null
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/VariantSerializer.java
@@ -0,0 +1,125 @@
+/*
+ * 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.flink.cdc.runtime.serializer;
+
+import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.types.variant.BinaryVariant;
+import org.apache.flink.cdc.common.types.variant.Variant;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+/** Serializer for {@link Variant}. */
+@Internal
+public class VariantSerializer extends TypeSerializerSingleton<Variant> {
+
+ public static final VariantSerializer INSTANCE = new VariantSerializer();
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public Variant createInstance() {
+ return Variant.newBuilder().ofNull();
+ }
+
+ @Override
+ public Variant copy(Variant from) {
+ BinaryVariant binaryVariant = toBinaryVariant(from);
+ return new BinaryVariant(
+ Arrays.copyOf(binaryVariant.getValue(), binaryVariant.getValue().length),
+ Arrays.copyOf(binaryVariant.getMetadata(), binaryVariant.getMetadata().length));
+ }
+
+ @Override
+ public Variant copy(Variant from, Variant reuse) {
+ return copy(from);
+ }
+
+ @Override
+ public int getLength() {
+ return -1;
+ }
+
+ @Override
+ public void serialize(Variant record, DataOutputView target) throws IOException {
+
+ BinaryVariant binaryVariant = toBinaryVariant(record);
+
+ target.writeInt(binaryVariant.getValue().length);
+ target.writeInt(binaryVariant.getMetadata().length);
+ target.write(binaryVariant.getValue());
+ target.write(binaryVariant.getMetadata());
+ }
+
+ @Override
+ public Variant deserialize(DataInputView source) throws IOException {
+ int valueLength = source.readInt();
+ int metadataLength = source.readInt();
+ byte[] value = new byte[valueLength];
+ byte[] metaData = new byte[metadataLength];
+
+ source.read(value);
+ source.read(metaData);
+ return new BinaryVariant(value, metaData);
+ }
+
+ @Override
+ public Variant deserialize(Variant reuse, DataInputView source) throws IOException {
+ return this.deserialize(source);
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws IOException {
+ int valueLength = source.readInt();
+ int metadataLength = source.readInt();
+ target.writeInt(valueLength);
+ target.writeInt(metadataLength);
+ target.write(source, valueLength + metadataLength);
+ }
+
+ @Override
+ public TypeSerializerSnapshot<Variant> snapshotConfiguration() {
+ return new VariantSerializerSnapshot();
+ }
+
+ private BinaryVariant toBinaryVariant(Variant variant) {
+ if (variant instanceof BinaryVariant) {
+ return (BinaryVariant) variant;
+ }
+
+ throw new UnsupportedOperationException("Unsupported variant type: " + variant.getClass());
+ }
+
+ /** Snapshot class for the {@link VariantSerializer}. */
+ @Internal
+ public static final class VariantSerializerSnapshot
+ extends SimpleTypeSerializerSnapshot<Variant> {
+ /** Constructor to create snapshot from serializer (writing the snapshot). */
+ public VariantSerializerSnapshot() {
+ super(() -> INSTANCE);
+ }
+ }
+}
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/AbstractBinaryWriter.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/AbstractBinaryWriter.java
index 8663f2229..5dc05bcdd 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/AbstractBinaryWriter.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/AbstractBinaryWriter.java
@@ -35,6 +35,8 @@ import org.apache.flink.cdc.common.data.binary.BinaryMapData;
import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
import org.apache.flink.cdc.common.data.binary.BinarySegmentUtils;
import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.types.variant.BinaryVariant;
+import org.apache.flink.cdc.common.types.variant.Variant;
import org.apache.flink.cdc.runtime.serializer.data.ArrayDataSerializer;
import org.apache.flink.cdc.runtime.serializer.data.MapDataSerializer;
import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
@@ -43,6 +45,7 @@ import org.apache.flink.core.memory.MemorySegmentFactory;
import java.io.IOException;
import java.io.OutputStream;
+import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
@@ -236,6 +239,19 @@ abstract class AbstractBinaryWriter implements BinaryWriter {
writeInt(pos, value.toMillisOfDay());
}
+ @Override
+ public void writeVariant(int pos, Variant variant) {
+ byte[] metadata = ((BinaryVariant) variant).getMetadata();
+ byte[] value = ((BinaryVariant) variant).getValue();
+ int metadataLen = metadata.length;
+
+ int length = metadata.length + value.length + 4;
+ ByteBuffer buffer = ByteBuffer.allocate(length);
+ buffer.putInt(metadataLen).put(metadata).put(value);
+
+ writeBytesToVarLenPart(pos, buffer.array(), length);
+ }
+
private void zeroBytes(int offset, int size) {
for (int i = offset; i < offset + size; i++) {
segment.put(i, (byte) 0);
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/BinaryArrayWriter.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/BinaryArrayWriter.java
index a2e79d52f..1177bcdee 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/BinaryArrayWriter.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/BinaryArrayWriter.java
@@ -132,6 +132,7 @@ public final class BinaryArrayWriter extends AbstractBinaryWriter {
case BIGINT:
case TIMESTAMP_WITHOUT_TIME_ZONE:
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ case VARIANT:
setNullLong(pos);
break;
case FLOAT:
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/BinaryWriter.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/BinaryWriter.java
index bf003448b..f6afad756 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/BinaryWriter.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/BinaryWriter.java
@@ -35,6 +35,7 @@ import org.apache.flink.cdc.common.types.LocalZonedTimestampType;
import org.apache.flink.cdc.common.types.TimeType;
import org.apache.flink.cdc.common.types.TimestampType;
import org.apache.flink.cdc.common.types.ZonedTimestampType;
+import org.apache.flink.cdc.common.types.variant.Variant;
import org.apache.flink.cdc.runtime.serializer.NullableSerializerWrapper;
import org.apache.flink.cdc.runtime.serializer.data.ArrayDataSerializer;
import org.apache.flink.cdc.runtime.serializer.data.MapDataSerializer;
@@ -89,6 +90,8 @@ public interface BinaryWriter {
void writeTime(int pos, TimeData value, int precision);
+ void writeVariant(int pos, Variant variant);
+
/** Finally, complete write to set real size to binary. */
void complete();
@@ -174,6 +177,9 @@ public interface BinaryWriter {
case VARBINARY:
writer.writeBinary(pos, (byte[]) o);
break;
+ case VARIANT:
+ writer.writeVariant(pos, (Variant) o);
+ break;
default:
throw new UnsupportedOperationException("Not support type: " + type);
}
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/schema/DataTypeSerializer.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/schema/DataTypeSerializer.java
index 214824132..517a0db49 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/schema/DataTypeSerializer.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/schema/DataTypeSerializer.java
@@ -40,6 +40,7 @@ import org.apache.flink.cdc.common.types.TimestampType;
import org.apache.flink.cdc.common.types.TinyIntType;
import org.apache.flink.cdc.common.types.VarBinaryType;
import org.apache.flink.cdc.common.types.VarCharType;
+import org.apache.flink.cdc.common.types.VariantType;
import org.apache.flink.cdc.common.types.ZonedTimestampType;
import org.apache.flink.cdc.runtime.serializer.EnumSerializer;
import org.apache.flink.core.memory.DataInputView;
@@ -176,6 +177,9 @@ public class DataTypeSerializer extends TypeSerializer<DataType> {
} else if (record instanceof BigIntType) {
enumSerializer.serialize(DataTypeClass.BIGINT, target);
target.writeBoolean(record.isNullable());
+ } else if (record instanceof VariantType) {
+ enumSerializer.serialize(DataTypeClass.VARIANT, target);
+ target.writeBoolean(record.isNullable());
} else {
throw new IllegalArgumentException("Unknown data type : " + record);
}
@@ -237,6 +241,8 @@ public class DataTypeSerializer extends TypeSerializer<DataType> {
return new DoubleType(isNullable);
case BIGINT:
return new BigIntType(isNullable);
+ case VARIANT:
+ return new VariantType(isNullable);
default:
throw new IllegalArgumentException("Unknown data type : " + dataTypeClass);
}
@@ -305,6 +311,7 @@ public class DataTypeSerializer extends TypeSerializer<DataType> {
DATE,
ZONED_TIMESTAMP,
DOUBLE,
- BIGINT
+ BIGINT,
+ VARIANT
}
}
diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/VariantSerializerTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/VariantSerializerTest.java
new file mode 100644
index 000000000..573a2c10d
--- /dev/null
+++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/serializer/VariantSerializerTest.java
@@ -0,0 +1,65 @@
+/*
+ * 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.flink.cdc.runtime.serializer;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.cdc.common.types.variant.Variant;
+import org.apache.flink.cdc.common.types.variant.VariantBuilder;
+
+/** A test for the {@link VariantSerializer}. */
+public class VariantSerializerTest extends SerializerTestBase<Variant> {
+
+ @Override
+ protected TypeSerializer<Variant> createSerializer() {
+ return VariantSerializer.INSTANCE;
+ }
+
+ @Override
+ protected int getLength() {
+ return -1;
+ }
+
+ @Override
+ protected Class<Variant> getTypeClass() {
+ return Variant.class;
+ }
+
+ @Override
+ protected Variant[] getTestData() {
+ VariantBuilder builder = Variant.newBuilder();
+ return new Variant[] {
+ builder.of(1),
+ builder.object()
+ .add("k", builder.of(1))
+ .add("object", builder.object().add("k", builder.of("hello")).build())
+ .add(
+ "array",
+ builder.array()
+ .add(builder.of(1))
+ .add(builder.of(2))
+ .add(builder.object().add("kk", builder.of(1.123f)).build())
+ .build())
+ .build(),
+ builder.array()
+ .add(builder.object().add("k", builder.of(1)).build())
+ .add(builder.of("hello"))
+ .add(builder.object().add("k", builder.of(2)).build())
+ .build()
+ };
+ }
+}
diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/typeutils/BinaryRecordDataGeneratorTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/typeutils/BinaryRecordDataGeneratorTest.java
index 847d36130..d6c0dd4d6 100644
--- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/typeutils/BinaryRecordDataGeneratorTest.java
+++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/typeutils/BinaryRecordDataGeneratorTest.java
@@ -30,10 +30,14 @@ import org.apache.flink.cdc.common.types.LocalZonedTimestampType;
import org.apache.flink.cdc.common.types.RowType;
import org.apache.flink.cdc.common.types.TimestampType;
import org.apache.flink.cdc.common.types.ZonedTimestampType;
+import org.apache.flink.cdc.common.types.variant.BinaryVariantBuilder;
+import org.apache.flink.cdc.common.types.variant.Variant;
import org.junit.jupiter.api.Test;
import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
import static org.assertj.core.api.Assertions.assertThat;
@@ -71,7 +75,17 @@ class BinaryRecordDataGeneratorTest {
DataTypes.ROW(
DataTypes.FIELD("t1", DataTypes.STRING()),
DataTypes.FIELD("t2", DataTypes.BIGINT())),
- DataTypes.STRING());
+ DataTypes.STRING(),
+ DataTypes.VARIANT());
+ BinaryVariantBuilder builder = new BinaryVariantBuilder();
+ Variant variant =
+ builder.array()
+ .add(builder.of(1))
+ .add(builder.of("hello"))
+ .add(builder.of(Instant.now().truncatedTo(ChronoUnit.MICROS)))
+ .add(builder.array().add(builder.of("hello2")).add(builder.of(10f)).build())
+ .add(builder.ofNull())
+ .build();
Object[] testData =
new Object[] {
@@ -102,7 +116,8 @@ class BinaryRecordDataGeneratorTest {
new BinaryRecordDataGenerator(
RowType.of(DataTypes.STRING(), DataTypes.BIGINT()))
.generate(new Object[] {BinaryStringData.fromString("test"), 24L}),
- null
+ null,
+ variant
};
BinaryRecordData actual = new BinaryRecordDataGenerator(rowType).generate(testData);
@@ -143,5 +158,6 @@ class BinaryRecordDataGeneratorTest {
.isEqualTo(BinaryStringData.fromString("test"));
assertThat(actual.getRow(24, 2).getLong(1)).isEqualTo(24L);
assertThat(actual.isNullAt(25)).isTrue();
+ assertThat(actual.getVariant(26)).isEqualTo(variant);
}
}