loserwang1024 commented on code in PR #3150:
URL: https://github.com/apache/flink-cdc/pull/3150#discussion_r1554803332
##########
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,50 @@ 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(22),
primaryKeyColumn.getColumnSize().orElse(null));
+ assertTrue(!primaryKeyColumn.isNullable());
+
+ // Map to VARCHAR of StarRocks if column is DECIMAL type and primary
key
+ // DECIMAL(20,0) is common in cdc pipeline, for example, the upstream
cdc source is unsigned
+ // BIGINT.
+ StarRocksColumn.Builder unsignedBigIntKeyBuilder =
+ new
StarRocksColumn.Builder().setColumnName("primary_key").setOrdinalPosition(1);
+ new DecimalType(20, 0)
+ .notNull()
+ .accept(new StarRocksUtils.CdcDataTypeTransformer(true,
unsignedBigIntKeyBuilder));
+ StarRocksColumn unsignedBigIntColumn =
unsignedBigIntKeyBuilder.build();
+ assertEquals("primary_key", unsignedBigIntColumn.getColumnName());
+ assertEquals(1, unsignedBigIntColumn.getOrdinalPosition());
+ assertEquals(StarRocksUtils.VARCHAR,
unsignedBigIntColumn.getDataType());
+ assertEquals(Integer.valueOf(22),
unsignedBigIntColumn.getColumnSize().orElse(null));
Review Comment:
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
