[3.0][cdc-common & runtime] Remove useless flink imports

flink-cdc-common/src/main/java/com/ververica/cdc/common/data/binary/BinaryStringData.java

@@ -138,10 +138,8 @@ public final class BinaryStringData extends LazyBinaryFormat<String> implements
     @Override
     public String toString() {
         if (javaObject == null) {
-            byte[] bytes =
+            byte[] bytes = BinarySegmentUtils.allocateReuseBytes(binarySection.sizeInBytes);
-                    org.apache.flink.table.data.binary.BinarySegmentUtils.allocateReuseBytes(
+            BinarySegmentUtils.copyToBytes(
-                            binarySection.sizeInBytes);
-            org.apache.flink.table.data.binary.BinarySegmentUtils.copyToBytes(
                     binarySection.segments,
                     binarySection.offset,
                     bytes,
@@ -359,7 +357,7 @@ public final class BinaryStringData extends LazyBinaryFormat<String> implements
     public BinaryStringData copy() {
         ensureMaterialized();
         byte[] copy =
-                org.apache.flink.table.data.binary.BinarySegmentUtils.copyToBytes(
+                BinarySegmentUtils.copyToBytes(
                         binarySection.segments, binarySection.offset, binarySection.sizeInBytes);
         return new BinaryStringData(
                 new MemorySegment[] {MemorySegmentFactory.wrap(copy)},
@@ -442,7 +440,7 @@ public final class BinaryStringData extends LazyBinaryFormat<String> implements
 
         if (i > j) {
             return fromBytes(
-                    org.apache.flink.table.data.binary.BinarySegmentUtils.copyToBytes(
+                    BinarySegmentUtils.copyToBytes(
                             binarySection.segments, binarySection.offset + j, i - j));
         } else {
             return EMPTY_UTF8;
@@ -463,7 +461,7 @@ public final class BinaryStringData extends LazyBinaryFormat<String> implements
             return true;
         }
         int find =
-                org.apache.flink.table.data.binary.BinarySegmentUtils.find(
+                BinarySegmentUtils.find(
                         binarySection.segments,
                         binarySection.offset,
                         binarySection.sizeInBytes,
@@ -586,7 +584,7 @@ public final class BinaryStringData extends LazyBinaryFormat<String> implements
                 if (byteIdx + str.binarySection.sizeInBytes > binarySection.sizeInBytes) {
                     return -1;
                 }
-                if (org.apache.flink.table.data.binary.BinarySegmentUtils.equals(
+                if (BinarySegmentUtils.equals(
                         binarySection.segments,
                         binarySection.offset + byteIdx,
                         str.binarySection.segments,
@@ -621,7 +619,7 @@ public final class BinaryStringData extends LazyBinaryFormat<String> implements
             if (byteIdx + str.binarySection.sizeInBytes > binarySection.sizeInBytes) {
                 return -1;
             }
-            if (org.apache.flink.table.data.binary.BinarySegmentUtils.equals(
+            if (BinarySegmentUtils.equals(
                     binarySection.segments,
                     binarySection.offset + byteIdx,
                     str.binarySection.segments,
@@ -743,7 +741,7 @@ public final class BinaryStringData extends LazyBinaryFormat<String> implements
     private boolean matchAtVarSeg(final BinaryStringData s, int pos) {
         return s.binarySection.sizeInBytes + pos <= binarySection.sizeInBytes
                 && pos >= 0
-                && org.apache.flink.table.data.binary.BinarySegmentUtils.equals(
+                && BinarySegmentUtils.equals(
                         binarySection.segments,
                         binarySection.offset + pos,
                         s.binarySection.segments,

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/com/ververica/cdc/debezium/event/DebeziumSchemaDataTypeInference.java

@@ -21,8 +21,6 @@ import com.ververica.cdc.common.types.DataField;
 import com.ververica.cdc.common.types.DataType;
 import com.ververica.cdc.common.types.DataTypes;
 import io.debezium.data.VariableScaleDecimal;
-import io.debezium.data.geometry.Geometry;
-import io.debezium.data.geometry.Point;
 import io.debezium.time.MicroTime;
 import io.debezium.time.MicroTimestamp;
 import io.debezium.time.NanoTime;
@@ -175,10 +173,6 @@ public class DebeziumSchemaDataTypeInference implements SchemaDataTypeInference,
 
     protected DataType inferStruct(Object value, Schema schema) {
         Struct struct = (Struct) value;
-        if (Geometry.LOGICAL_NAME.equals(schema.name())
-                || Point.LOGICAL_NAME.equals(schema.name())) {
-            return DataTypes.STRING();
-        }
         return DataTypes.ROW(
                 schema.fields().stream()
                         .map(

flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/serializer/data/StringDataSerializer.java

@@ -20,9 +20,9 @@ import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot;
 import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
-import org.apache.flink.table.data.binary.BinarySegmentUtils;
 
 import com.ververica.cdc.common.data.StringData;
+import com.ververica.cdc.common.data.binary.BinarySegmentUtils;
 import com.ververica.cdc.common.data.binary.BinaryStringData;
 import com.ververica.cdc.runtime.serializer.TypeSerializerSingleton;
 

flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/serializer/data/writer/BinaryRecordDataWriter.java

@@ -17,10 +17,10 @@
 package com.ververica.cdc.runtime.serializer.data.writer;
 
 import org.apache.flink.core.memory.MemorySegmentFactory;
-import org.apache.flink.table.data.binary.BinarySegmentUtils;
 
 import com.ververica.cdc.common.annotation.Internal;
 import com.ververica.cdc.common.data.binary.BinaryRecordData;
+import com.ververica.cdc.common.data.binary.BinarySegmentUtils;
 
 /** Writer for {@link BinaryRecordData}. */
 @Internal