public static LogicalType convertLogicType(BsonValue bsonValue) {
        BsonType bsonType = bsonValue.getBsonType();
        if (INT_TYPE.contains(bsonType)) {
            return new BigIntType();
        }
        if (BOOL_TYPE.contains(bsonType)) {
            return new BooleanType();
        }
        if (DOUBLE_TYPE.contains(bsonType)) {
            return new DoubleType();
        }
        if (DECIMAL_TYPE.contains(bsonType)) {
            return new DecimalType();
        }
        if (TIMESTAMP_TYPE.contains(bsonType)) {
            return new TimestampType();
        }
        if (VARCHAR_TYPE.contains(bsonType)) {
            return new VarCharType();
        }
        if (BINARY_TYPE.contains(bsonType)) {
            return new BinaryType();
        }
        if (ARRAY_TYPE.contains(bsonType)) {
            return new ArrayType(new IntType());
        }
        return null;
    }
Get table id from source record
@param dataRecord
@return
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/debezium/utils/RecordUtils.java