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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
