[FLINK-37017][cdc-common] Supports map and array types binary record data

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinaryMapData.java

@@ -0,0 +1,118 @@
+/*
+ * 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.data.binary;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.data.MapData;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.flink.cdc.common.utils.Preconditions.checkArgument;
+
+/**
+ * [4 byte(keyArray size in bytes)] + [Key BinaryArray] + [Value BinaryArray].
+ *
+ * <p>{@code BinaryMap} are influenced by Apache Spark UnsafeMapData.
+ */
+@Internal
+public class BinaryMapData extends BinarySection implements MapData {
+    private final BinaryArrayData keys;
+    private final BinaryArrayData values;
+
+    public BinaryMapData() {
+        keys = new BinaryArrayData();
+        values = new BinaryArrayData();
+    }
+
+    public int size() {
+        return keys.size();
+    }
+
+    @Override
+    public void pointTo(MemorySegment[] segments, int offset, int sizeInBytes) {
+        // Read the numBytes of key array from the first 4 bytes.
+        final int keyArrayBytes = BinarySegmentUtils.getInt(segments, offset);
+        assert keyArrayBytes >= 0 : "keyArraySize (" + keyArrayBytes + ") should >= 0";
+        final int valueArrayBytes = sizeInBytes - keyArrayBytes - 4;
+        assert valueArrayBytes >= 0 : "valueArraySize (" + valueArrayBytes + ") should >= 0";
+
+        keys.pointTo(segments, offset + 4, keyArrayBytes);
+        values.pointTo(segments, offset + 4 + keyArrayBytes, valueArrayBytes);
+
+        assert keys.size() == values.size();
+
+        this.segments = segments;
+        this.offset = offset;
+        this.sizeInBytes = sizeInBytes;
+    }
+
+    public BinaryArrayData keyArray() {
+        return keys;
+    }
+
+    public BinaryArrayData valueArray() {
+        return values;
+    }
+
+    public Map<?, ?> toJavaMap(DataType keyType, DataType valueType) {
+        Object[] keyArray = keys.toObjectArray(keyType);
+        Object[] valueArray = values.toObjectArray(valueType);
+
+        Map<Object, Object> map = new HashMap<>();
+        for (int i = 0; i < keyArray.length; i++) {
+            map.put(keyArray[i], valueArray[i]);
+        }
+        return map;
+    }
+
+    public BinaryMapData copy() {
+        return copy(new BinaryMapData());
+    }
+
+    public BinaryMapData copy(BinaryMapData reuse) {
+        byte[] bytes = BinarySegmentUtils.copyToBytes(segments, offset, sizeInBytes);
+        reuse.pointTo(MemorySegmentFactory.wrap(bytes), 0, sizeInBytes);
+        return reuse;
+    }
+
+    @Override
+    public int hashCode() {
+        return BinarySegmentUtils.hashByWords(segments, offset, sizeInBytes);
+    }
+
+    // ------------------------------------------------------------------------------------------
+    // Construction Utilities
+    // ------------------------------------------------------------------------------------------
+
+    public static BinaryMapData valueOf(BinaryArrayData key, BinaryArrayData value) {
+        checkArgument(key.segments.length == 1 && value.getSegments().length == 1);
+        byte[] bytes = new byte[4 + key.sizeInBytes + value.sizeInBytes];
+        MemorySegment segment = MemorySegmentFactory.wrap(bytes);
+        segment.putInt(0, key.sizeInBytes);
+        key.getSegments()[0].copyTo(key.getOffset(), segment, 4, key.sizeInBytes);
+        value.getSegments()[0].copyTo(
+                value.getOffset(), segment, 4 + key.sizeInBytes, value.sizeInBytes);
+        BinaryMapData map = new BinaryMapData();
+        map.pointTo(segment, 0, bytes.length);
+        return map;
+    }
+}

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinaryRecordData.java

@@ -200,12 +200,14 @@ public byte[] getBinary(int pos) {
 
     @Override
     public ArrayData getArray(int pos) {
-        throw new UnsupportedOperationException("Not support ArrayData");
+        assertIndexIsValid(pos);
+        return BinarySegmentUtils.readArrayData(segments, offset, getLong(pos));
     }
 
     @Override
     public MapData getMap(int pos) {
-        throw new UnsupportedOperationException("Not support MapData.");
+        assertIndexIsValid(pos);
+        return BinarySegmentUtils.readMapData(segments, offset, getLong(pos));
     }
 
     @Override

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinarySegmentUtils.java

@@ -18,8 +18,10 @@
 package org.apache.flink.cdc.common.data.binary;
 
 import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.data.ArrayData;
 import org.apache.flink.cdc.common.data.DecimalData;
 import org.apache.flink.cdc.common.data.LocalZonedTimestampData;
+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;
@@ -318,7 +320,7 @@ public static boolean equals(
         }
     }
 
-    static boolean equalsMultiSegments(
+    public static boolean equalsMultiSegments(
             MemorySegment[] segments1,
             int offset1,
             MemorySegment[] segments2,
@@ -1154,4 +1156,24 @@ private static int findInMultiSegments(
         }
         return -1;
     }
+
+    /** Gets an instance of {@link MapData} from underlying {@link MemorySegment}. */
+    public static MapData readMapData(
+            MemorySegment[] segments, int baseOffset, long offsetAndSize) {
+        final int size = ((int) offsetAndSize);
+        int offset = (int) (offsetAndSize >> 32);
+        BinaryMapData map = new BinaryMapData();
+        map.pointTo(segments, offset + baseOffset, size);
+        return map;
+    }
+
+    /** Gets an instance of {@link ArrayData} from underlying {@link MemorySegment}. */
+    public static ArrayData readArrayData(
+            MemorySegment[] segments, int baseOffset, long offsetAndSize) {
+        final int size = ((int) offsetAndSize);
+        int offset = (int) (offsetAndSize >> 32);
+        BinaryArrayData array = new BinaryArrayData();
+        array.pointTo(segments, offset + baseOffset, size);
+        return array;
+    }
 }

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonWriterHelper.java

@@ -17,20 +17,31 @@
 
 package org.apache.flink.cdc.connectors.paimon.sink.v2;
 
+import org.apache.flink.cdc.common.data.ArrayData;
 import org.apache.flink.cdc.common.data.DecimalData;
+import org.apache.flink.cdc.common.data.MapData;
 import org.apache.flink.cdc.common.data.RecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryArrayData;
+import org.apache.flink.cdc.common.data.binary.BinaryMapData;
+import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
 import org.apache.flink.cdc.common.event.DataChangeEvent;
 import org.apache.flink.cdc.common.schema.Column;
 import org.apache.flink.cdc.common.schema.Schema;
 import org.apache.flink.cdc.common.types.DataType;
 import org.apache.flink.cdc.common.types.DataTypeChecks;
+import org.apache.flink.cdc.common.types.DataTypeRoot;
+import org.apache.flink.core.memory.MemorySegment;
 
+import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.data.BinaryString;
 import org.apache.paimon.data.Decimal;
 import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.memory.MemorySegmentUtils;
 import org.apache.paimon.types.RowKind;
 
+import java.nio.ByteBuffer;
 import java.time.ZoneId;
 import java.util.ArrayList;
 import java.util.List;
@@ -118,7 +129,11 @@ private static RecordData.FieldGetter createFieldGetter(
                 break;
             case ROW:
                 final int rowFieldCount = getFieldCount(fieldType);
-                fieldGetter = row -> row.getRow(fieldPos, rowFieldCount);
+                fieldGetter = new BinaryFieldDataGetter(fieldPos, DataTypeRoot.ROW, rowFieldCount);
+                break;
+            case ARRAY:
+            case MAP:
+                fieldGetter = new BinaryFieldDataGetter(fieldPos, fieldType.getTypeRoot());
                 break;
             default:
                 throw new IllegalArgumentException(
@@ -163,4 +178,121 @@ public static GenericRow convertEventToGenericRow(
         }
         return genericRow;
     }
+
+    /** A helper class for {@link PaimonWriter} to create FieldGetter and GenericRow. */
+    public static class BinaryFieldDataGetter implements RecordData.FieldGetter {
+        private final int fieldPos;
+        private final DataTypeRoot dataTypeRoot;
+        private final int rowFieldCount;
+
+        BinaryFieldDataGetter(int fieldPos, DataTypeRoot dataTypeRoot) {
+            this(fieldPos, dataTypeRoot, -1);
+        }
+
+        BinaryFieldDataGetter(int fieldPos, DataTypeRoot dataTypeRoot, int rowFieldCount) {
+            this.fieldPos = fieldPos;
+            this.dataTypeRoot = dataTypeRoot;
+            this.rowFieldCount = rowFieldCount;
+        }
+
+        @Override
+        public Object getFieldOrNull(RecordData row) {
+            switch (dataTypeRoot) {
+                case ARRAY:
+                    return getArrayField(row);
+                case MAP:
+                    return getMapField(row);
+                case ROW:
+                    return getRecordField(row);
+                default:
+                    throw new IllegalArgumentException("Unsupported field type: " + dataTypeRoot);
+            }
+        }
+
+        private Object getArrayField(RecordData row) {
+            ArrayData arrayData = row.getArray(fieldPos);
+            if (!(arrayData instanceof BinaryArrayData)) {
+                throw new IllegalArgumentException(
+                        "Expected BinaryArrayData but was " + arrayData.getClass().getSimpleName());
+            }
+            BinaryArrayData binaryArrayData = (BinaryArrayData) arrayData;
+            return convertSegments(
+                    binaryArrayData.getSegments(),
+                    binaryArrayData.getOffset(),
+                    binaryArrayData.getSizeInBytes(),
+                    MemorySegmentUtils::readArrayData);
+        }
+
+        private Object getMapField(RecordData row) {
+            MapData mapData = row.getMap(fieldPos);
+            if (!(mapData instanceof BinaryMapData)) {
+                throw new IllegalArgumentException(
+                        "Expected BinaryMapData but was " + mapData.getClass().getSimpleName());
+            }
+            BinaryMapData binaryMapData = (BinaryMapData) mapData;
+            return convertSegments(
+                    binaryMapData.getSegments(),
+                    binaryMapData.getOffset(),
+                    binaryMapData.getSizeInBytes(),
+                    MemorySegmentUtils::readMapData);
+        }
+
+        private Object getRecordField(RecordData row) {
+            RecordData recordData = row.getRow(fieldPos, rowFieldCount);
+            if (!(recordData instanceof BinaryRecordData)) {
+                throw new IllegalArgumentException(
+                        "Expected BinaryRecordData but was "
+                                + recordData.getClass().getSimpleName());
+            }
+            BinaryRecordData binaryRecordData = (BinaryRecordData) recordData;
+            return convertSegments(
+                    binaryRecordData.getSegments(),
+                    binaryRecordData.getOffset(),
+                    binaryRecordData.getSizeInBytes(),
+                    (segments, offset, sizeInBytes) ->
+                            MemorySegmentUtils.readRowData(
+                                    segments, rowFieldCount, offset, sizeInBytes));
+        }
+
+        private <T> T convertSegments(
+                MemorySegment[] segments,
+                int offset,
+                int sizeInBytes,
+                SegmentConverter<T> converter) {
+            org.apache.paimon.memory.MemorySegment[] paimonMemorySegments =
+                    new org.apache.paimon.memory.MemorySegment[segments.length];
+            for (int i = 0; i < segments.length; i++) {
+                MemorySegment currMemorySegment = segments[i];
+                ByteBuffer byteBuffer = currMemorySegment.wrap(0, currMemorySegment.size());
+
+                // Allocate a new byte array and copy the data from the ByteBuffer
+                byte[] bytes = new byte[currMemorySegment.size()];
+                byteBuffer.get(bytes);
+
+                paimonMemorySegments[i] = org.apache.paimon.memory.MemorySegment.wrap(bytes);
+            }
+            return converter.convert(paimonMemorySegments, offset, sizeInBytes);
+        }
+
+        private interface SegmentConverter<T> {
+            T convert(
+                    org.apache.paimon.memory.MemorySegment[] segments, int offset, int sizeInBytes);
+        }
+
+        /**
+         * Gets an instance of {@link InternalRow} from underlying {@link
+         * org.apache.paimon.memory.MemorySegment}.
+         */
+        public InternalRow readRowData(
+                org.apache.paimon.memory.MemorySegment[] segments,
+                int numFields,
+                int baseOffset,
+                long offsetAndSize) {
+            final int size = ((int) offsetAndSize);
+            int offset = (int) (offsetAndSize >> 32);
+            BinaryRow row = new BinaryRow(numFields);
+            row.pointTo(segments, offset + baseOffset, size);
+            return row;
+        }
+    }
 }