yuxiqian commented on code in PR #3150: URL: https://github.com/apache/flink-cdc/pull/3150#discussion_r1548950314
########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/CdcDataTypeTransformerTest.java: ########## @@ -75,6 +76,35 @@ public void testCharTypeForPrimaryKey() { assertTrue(smallLengthColumn.isNullable()); } + @Test + public void testDecimalForPrimaryKey() { + // Map to DECIMAL of StarRocks if column is DECIMAL type and not primary key. + StarRocksColumn.Builder noPrimaryKeyBuilder = + new StarRocksColumn.Builder().setColumnName("no_primary_key").setOrdinalPosition(0); + new DecimalType(20, 1) + .accept(new StarRocksUtils.CdcDataTypeTransformer(false, noPrimaryKeyBuilder)); + StarRocksColumn noPrimaryKeyColumn = noPrimaryKeyBuilder.build(); + assertEquals("no_primary_key", noPrimaryKeyColumn.getColumnName()); + assertEquals(0, noPrimaryKeyColumn.getOrdinalPosition()); + assertEquals(StarRocksUtils.DECIMAL, noPrimaryKeyColumn.getDataType()); + assertEquals(Integer.valueOf(20), noPrimaryKeyColumn.getColumnSize().orElse(null)); + assertEquals(Integer.valueOf(1), noPrimaryKeyColumn.getDecimalDigits().get()); + assertTrue(noPrimaryKeyColumn.isNullable()); + + // Map to VARCHAR of StarRocks if column is DECIMAL type and primary key. + StarRocksColumn.Builder primaryKeyBuilder = + new StarRocksColumn.Builder().setColumnName("primary_key").setOrdinalPosition(1); + new DecimalType(20, 1) + .notNull() + .accept(new StarRocksUtils.CdcDataTypeTransformer(true, primaryKeyBuilder)); + StarRocksColumn primaryKeyColumn = primaryKeyBuilder.build(); + assertEquals("primary_key", primaryKeyColumn.getColumnName()); + assertEquals(1, primaryKeyColumn.getOrdinalPosition()); + assertEquals(StarRocksUtils.VARCHAR, primaryKeyColumn.getDataType()); + assertEquals(Integer.valueOf(20), primaryKeyColumn.getColumnSize().orElse(null)); + assertTrue(!primaryKeyColumn.isNullable()); Review Comment: Should we add test cases covering both zero scaled & non-zero scaled decimal types? ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksUtils.java: ########## @@ -297,10 +297,17 @@ public StarRocksColumn.Builder visit(DoubleType doubleType) { @Override public StarRocksColumn.Builder visit(DecimalType decimalType) { - builder.setDataType(DECIMAL); + // StarRocks does not support Decimal as primary key, so decimal should be cast to + // VARCHAR. + if (!isPrimaryKeys) { + builder.setDataType(DECIMAL); + builder.setColumnSize(decimalType.getPrecision()); + builder.setDecimalDigits(decimalType.getScale()); + } else { + builder.setDataType(VARCHAR); + builder.setColumnSize(Math.min(decimalType.getPrecision(), MAX_VARCHAR_SIZE)); Review Comment: For a `DecimalType` with precision `N`, we may need `N + 1` or `N + 2` characters to store it as a string (one for negative sign, and one for decimal point) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org