loserwang1024 commented on code in PR #3150:
URL: https://github.com/apache/flink-cdc/pull/3150#discussion_r1549128925
##########
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:
Thank you very much for your advice, I didn't take that into consideration
##########
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:
Just done it.
--
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
