public boolean validateData(Object srcValue, EtlJobConfig.EtlColumn etlColumn, ColumnParser columnParser, Row row) {

        switch (etlColumn.columnType.toUpperCase()) {
            case "DECIMALV2":
            case "DECIMAL32":
            case "DECIMAL64":
            case "DECIMAL128":
                // TODO(wb):  support decimal round; see be DecimalV2Value::round
                DecimalParser decimalParser = (DecimalParser) columnParser;
                BigDecimal srcBigDecimal = (BigDecimal) srcValue;
                if (srcValue != null && (decimalParser.getMaxValue().compareTo(srcBigDecimal) < 0 ||
                        decimalParser.getMinValue().compareTo(srcBigDecimal) > 0)) {
                    LOG.warn(String.format(
                            "decimal value is not valid for defination, column=%s, value=%s,precision=%s,scale=%s",
                            etlColumn.columnName, srcValue.toString(), srcBigDecimal.precision(),
                            srcBigDecimal.scale()));
                    return false;
                }
                break;
            case "CHAR":
            case "VARCHAR":
                // TODO(wb) padding char type
                int strSize = 0;
                if (srcValue != null &&
                        (strSize = srcValue.toString().getBytes(StandardCharsets.UTF_8).length) >
                                etlColumn.stringLength) {
                    LOG.warn(String.format(
                            "the length of input is too long than schema. column_name:%s," +
                                    "input_str[%s],schema length:%s,actual length:%s",
                            etlColumn.columnName, row.toString(), etlColumn.stringLength, strSize));
                    return false;
                }
                break;
        }
        return true;
    }
@Test
    public void testValidateData() {
        SparkDpp sparkDpp = new SparkDpp();

        // decimal
        EtlJobConfig.EtlColumn etlColumn = new EtlJobConfig.EtlColumn();
        etlColumn.columnType = "DECIMALV2";
        etlColumn.precision = 3;
        etlColumn.scale = 2;

        DecimalParser decimalParser = new DecimalParser(etlColumn);
        // test max/min
        Assert.assertTrue(decimalParser.getMaxValue().toString().equals("9.99"));
        Assert.assertTrue(decimalParser.getMinValue().toString().equals("-9.99"));
        // normal
        BigDecimal bigDecimal = new BigDecimal("1.21");
        Assert.assertTrue(sparkDpp.validateData(bigDecimal, etlColumn, decimalParser, RowFactory.create(bigDecimal)));
        // failed
        BigDecimal bigDecimalFailed = new BigDecimal("10");
        Assert.assertFalse(
                sparkDpp.validateData(bigDecimalFailed, etlColumn, decimalParser, RowFactory.create(bigDecimalFailed)));

        // string
        EtlJobConfig.EtlColumn stringColumn = new EtlJobConfig.EtlColumn();
        stringColumn.stringLength = 3;
        stringColumn.columnType = "VARCHAR";
        StringParser stringParser = new StringParser(stringColumn);
        // normal
        String normalString = "a1";
        Assert.assertTrue(
                sparkDpp.validateData(normalString, stringColumn, stringParser, RowFactory.create(normalString)));
        // cn normal
        String normalStringCN = "中";
        Assert.assertTrue(
                sparkDpp.validateData(normalStringCN, stringColumn, stringParser, RowFactory.create(normalStringCN)));
        // cn failed
        String failedStringCN = "中a";
        Assert.assertFalse(
                sparkDpp.validateData(failedStringCN, stringColumn, stringParser, RowFactory.create(failedStringCN)));
    }