public static ByteBuffer getHashValue(Object o, DataType type, EtlJobConfig.EtlColumn column) {
        // null as int 0
        if (o == null) {
            ByteBuffer buffer = ByteBuffer.allocate(8);
            buffer.order(ByteOrder.LITTLE_ENDIAN);
            buffer.putInt(0);
            buffer.flip();
            return buffer;
        }

        // varchar and char
        if (type.equals(DataTypes.StringType)) {
            try {
                String str = String.valueOf(o);
                return ByteBuffer.wrap(str.getBytes("UTF-8"));
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

        // date
        // fe/fe-core/src/main/java/com/starrocks/analysis/DateLiteral.java#L273
        if (type.equals(DataTypes.DateType)) {
            try {
                Date d = (Date) o;
                String str = ColumnParser.DATE_FORMATTER.format(d.toLocalDate());
                return ByteBuffer.wrap(str.getBytes("UTF-8"));
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

        // datetime
        // fe/fe-core/src/main/java/com/starrocks/analysis/DateLiteral.java#L273
        if (type.equals(DataTypes.TimestampType)) {
            try {
                Timestamp t = (Timestamp) o;
                String str;
                if (t.getNanos() > 1000) {
                    str = ColumnParser.DATE_TIME_WITH_MS_FORMATTER.format(t.toLocalDateTime());
                } else {
                    str = ColumnParser.DATE_TIME_FORMATTER.format(t.toLocalDateTime());
                }
                return ByteBuffer.wrap(str.getBytes("UTF-8"));
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

        // decimal
        // fe/fe-core/src/main/java/com/starrocks/analysis/DecimalLiteral.java#L292
        if (type instanceof DecimalType) {
            BigDecimal d = (BigDecimal) o;
            ByteBuffer buffer = getDecimalByteBuffer(d, column);
            buffer.flip();
            return buffer;
        }

        // bool tinyint smallint int bigint
        ByteBuffer buffer = ByteBuffer.allocate(8);
        buffer.order(ByteOrder.LITTLE_ENDIAN);
        if (type.equals(DataTypes.BooleanType)) {
            byte b = (boolean) o ? (byte) 1 : (byte) 0;
            buffer.put(b);
        } else if (type.equals(DataTypes.ByteType)) {
            buffer.put((byte) o);
        } else if (type.equals(DataTypes.ShortType)) {
            buffer.putShort((Short) o);
        } else if (type.equals(DataTypes.IntegerType)) {
            buffer.putInt((Integer) o);
        } else if (type.equals(DataTypes.LongType)) {
            buffer.putLong((Long) o);
        }
        buffer.flip();
        return buffer;
    }
@Test
    public void testGetHashValue() {
        CRC32 hashValue = new CRC32();
        ByteBuffer bf = null;

        EtlJobConfig.EtlColumn column = new EtlJobConfig.EtlColumn("k1", "INT", true, true, "NONE", "0", 0, 0, 0);
        bf = DppUtils.getHashValue(null, DataTypes.IntegerType, column);
        hashValue.reset();
        hashValue.update(bf.array(), 0, bf.limit());
        Assert.assertEquals(558161692L, hashValue.getValue());

        column = new EtlJobConfig.EtlColumn("k1", "TINYINT", true, true, "NONE", "0", 0, 0, 0);
        bf = DppUtils.getHashValue(new Byte((byte) 1), DataTypes.ByteType, column);
        hashValue.reset();
        hashValue.update(bf.array(), 0, bf.limit());
        Assert.assertEquals(2768625435L, hashValue.getValue());

        column = new EtlJobConfig.EtlColumn("k1", "SMALLINT", true, true, "NONE", "0", 0, 0, 0);
        bf = DppUtils.getHashValue(new Short((short) 1), DataTypes.ShortType, column);
        hashValue.reset();
        hashValue.update(bf.array(), 0, bf.limit());
        Assert.assertEquals(1489118142L, hashValue.getValue());

        column = new EtlJobConfig.EtlColumn("k1", "INT", true, true, "NONE", "0", 0, 0, 0);
        bf = DppUtils.getHashValue(new Integer(1), DataTypes.IntegerType, column);
        hashValue.reset();
        hashValue.update(bf.array(), 0, bf.limit());
        Assert.assertEquals(2583214201L, hashValue.getValue());

        column = new EtlJobConfig.EtlColumn("k1", "BIGINT", true, true, "NONE", "0", 0, 0, 0);
        bf = DppUtils.getHashValue(new Long(1), DataTypes.LongType, column);
        hashValue.reset();
        hashValue.update(bf.array(), 0, bf.limit());
        Assert.assertEquals(2844319735L, hashValue.getValue());

        column = new EtlJobConfig.EtlColumn("k1", "VARCHAR", true, true, "NONE", "0", 100, 0, 0);
        bf = DppUtils.getHashValue("12345abcde", DataTypes.StringType, column);
        hashValue.reset();
        hashValue.update(bf.array(), 0, bf.limit());
        Assert.assertEquals(603981213L, hashValue.getValue());

        column = new EtlJobConfig.EtlColumn("k1", "BOOLEAN", true, true, "NONE", "0", 0, 0, 0);
        bf = DppUtils.getHashValue(new Boolean(true), DataTypes.BooleanType, column);
        hashValue.reset();
        hashValue.update(bf.array(), 0, bf.limit());
        Assert.assertEquals(2768625435L, hashValue.getValue());

        // date
        column = new EtlJobConfig.EtlColumn("k1", "DATE", true, true, "NONE", "0", 0, 0, 0);
        bf = DppUtils.getHashValue(Date.valueOf("2023-07-11"), DataTypes.DateType, column);
        hashValue.reset();
        hashValue.update(bf.array(), 0, bf.limit());
        Assert.assertEquals(791129379L, hashValue.getValue());

        // datetime
        column = new EtlJobConfig.EtlColumn("k1", "DATETIME", true, true, "NONE", "0", 0, 0, 0);
        bf = DppUtils.getHashValue(Timestamp.valueOf("2023-07-11 12:12:12"), DataTypes.TimestampType, column);
        hashValue.reset();
        hashValue.update(bf.array(), 0, bf.limit());
        Assert.assertEquals(541277948L, hashValue.getValue());

        column = new EtlJobConfig.EtlColumn("k1", "DATETIME", true, true, "NONE", "0", 0, 0, 0);
        bf = DppUtils.getHashValue(Timestamp.valueOf("2023-07-11 12:12:12.123456"), DataTypes.TimestampType, column);
        hashValue.reset();
        hashValue.update(bf.array(), 0, bf.limit());
        Assert.assertEquals(2559661226L, hashValue.getValue());

        // decimal
        column = new EtlJobConfig.EtlColumn("k1", "DECIMALV2", true, true, "NONE", "0", 0, 27, 9);
        bf = DppUtils.getHashValue(new BigDecimal("1.234"), DecimalType.apply(27, 9), column);
        hashValue.reset();
        hashValue.update(bf.array(), 0, bf.limit());
        Assert.assertEquals(3199857971L, hashValue.getValue());

        column = new EtlJobConfig.EtlColumn("k1", "DECIMALV2", true, true, "NONE", "0", 0, 20, 3);
        bf = DppUtils.getHashValue(new BigDecimal("1.234"), DecimalType.apply(20, 3), column);
        hashValue.reset();
        hashValue.update(bf.array(), 0, bf.limit());
        Assert.assertEquals(3199857971L, hashValue.getValue());

        column = new EtlJobConfig.EtlColumn("k1", "DECIMAL32", true, true, "NONE", "0", 0, 9, 3);
        bf = DppUtils.getHashValue(new BigDecimal("1.234"), DecimalType.apply(9, 3), column);
        hashValue.reset();
        hashValue.update(bf.array(), 0, bf.limit());
        Assert.assertEquals(2853177042L, hashValue.getValue());

        column = new EtlJobConfig.EtlColumn("k1", "DECIMAL64", true, true, "NONE", "0", 0, 18, 6);
        bf = DppUtils.getHashValue(new BigDecimal("1.234"), DecimalType.apply(18, 6), column);
        hashValue.reset();
        hashValue.update(bf.array(), 0, bf.limit());
        Assert.assertEquals(563521641L, hashValue.getValue());

        column = new EtlJobConfig.EtlColumn("k1", "DECIMAL128", true, true, "NONE", "0", 0, 38, 9);
        bf = DppUtils.getHashValue(new BigDecimal("1.234"), DecimalType.apply(38, 9), column);
        hashValue.reset();
        hashValue.update(bf.array(), 0, bf.limit());
        Assert.assertEquals(4291267310L, hashValue.getValue());

        column = new EtlJobConfig.EtlColumn("k1", "DECIMAL128", true, true, "NONE", "0", 0, 27, 9);
        bf = DppUtils.getHashValue(new BigDecimal("1.234"), DecimalType.apply(27, 9), column);
        hashValue.reset();
        hashValue.update(bf.array(), 0, bf.limit());
        Assert.assertEquals(3199857971L, hashValue.getValue());
    }