snuyanzin commented on code in PR #49:
URL:
https://github.com/apache/flink-connector-jdbc/pull/49#discussion_r1221225513
##########
flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/databases/clickhouse/dialect/ClickHouseRowConvert.java:
##########
@@ -0,0 +1,85 @@
+package org.apache.flink.connector.jdbc.databases.clickhouse.dialect;
+
+import org.apache.flink.connector.jdbc.converter.AbstractJdbcRowConverter;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import com.clickhouse.data.value.UnsignedByte;
+import com.clickhouse.data.value.UnsignedInteger;
+import com.clickhouse.data.value.UnsignedShort;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+
+/**
+ * Runtime converter that responsible to convert between JDBC object and Flink
internal object for
+ * ClickHouse.
+ */
+public class ClickHouseRowConvert extends AbstractJdbcRowConverter {
+ @Override
+ public String converterName() {
+ return "ClickHouse";
+ }
+
+ public ClickHouseRowConvert(RowType rowType) {
+ super(rowType);
+ }
+
+ @Override
+ protected JdbcDeserializationConverter createInternalConverter(LogicalType
type) {
+ switch (type.getTypeRoot()) {
+ case NULL:
+ return null;
+ case BOOLEAN:
+ case FLOAT:
+ case DOUBLE:
+ return val -> val;
+ case TINYINT:
+ return val -> ((Byte) val).byteValue();
+ case SMALLINT:
+ return val ->
+ val instanceof UnsignedByte
+ ? ((UnsignedByte) val).shortValue()
+ : ((Short) val).shortValue();
+ case INTEGER:
+ return val ->
+ val instanceof UnsignedShort
+ ? ((UnsignedShort) val).intValue()
+ : ((Integer) val).intValue();
+ case BIGINT:
+ return jdbcField -> {
+ if (jdbcField instanceof UnsignedInteger) {
+ return ((UnsignedInteger) jdbcField).longValue();
+ } else if (jdbcField instanceof Long) {
+ return ((Long) jdbcField).longValue();
+ }
+ // UINT64 is not supported,the uint64 range exceeds the
long range
Review Comment:
i still don't see the issue why for `UInt64`, `UInt128`, `UInt256` we can
not return `BigInteger` ?
--
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]
