public static LogicalType convertLogicType(Column column, Struct struct) {
        String typeName = column.typeName();
        // Oracle Number Type default precision is 38
        Integer p = column.length() == 0 ? 38 : column.length();
        Integer s = column.scale().isPresent() ? column.scale().orElse(0) : struct.getInt32("scale");
        if (NUMBER_TYPE.contains(typeName)) {
            if (p == 1) {
                return new BooleanType();
            }
            if (s <= 0 && p - s < 3) {
                return new TinyIntType();
            }
            if (s <= 0 && p - s < 5) {
                return new SmallIntType();
            }
            if (s <= 0 && p - s < 10) {
                return new IntType();
            }
            if (s <= 0 && p - s < 19) {
                return new BigIntType();
            }
            if (s <= 0 && p - s >= 10 && p - s <= 38) {
                return new DecimalType(p - s, 0);
            }
            if (s > 0) {
                return new DecimalType(p, s);
            }
            if (s <= 0 && p - s > 38) {
                return new VarCharType(Integer.MAX_VALUE);
            }
        }
        if (FLOAT_TYPE.contains(typeName)) {
            return new FloatType();
        }
        if (DOUBLE_TYPE.contains(typeName)) {
            return new DoubleType();
        }
        if (TIMESTAMP_TYPE.contains(typeName)) {
            return new TimestampType(p);
        }
        if (VARCHAR_TYPE.contains(typeName)) {
            return new VarCharType(Integer.MAX_VALUE);
        }
        if (BINARY_TYPE.contains(typeName)) {
            return new BinaryType();
        }
        if (BIGINT_TYPE.contains(typeName)) {
            return new BigIntType();
        }
        return null;
    }
According column's type, get it's logicalType
Refer: https://ververica.github.io/flink-cdc-connectors/master/content/connectors/oracle-cdc.html?from_wecom=1
@param column
@param struct
@return
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/util/RecordUtils.java