fix: database subscription data type conversion bug (#250)

sheyanjie-qq · web-flow · commit 2ead789a9406 · 2025-05-19T11:22:47.000+08:00
* fix: data type conversion in database subscription

* test: added database subscription data type tests
diff --git a/deploy-pom.xml b/deploy-pom.xml
@@ -5,7 +5,7 @@
 
   <groupId>com.taosdata.jdbc</groupId>
   <artifactId>taos-jdbcdriver</artifactId>
-  <version>3.6.2</version>
+  <version>3.6.3</version>
   <packaging>jar</packaging>
 
   <name>JDBCDriver</name>
diff --git a/pom.xml b/pom.xml
@@ -3,7 +3,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.taosdata.jdbc</groupId>
     <artifactId>taos-jdbcdriver</artifactId>
-    <version>3.6.2</version>
+    <version>3.6.3</version>
 
     <packaging>jar</packaging>
     <name>JDBCDriver</name>
diff --git a/src/main/java/com/taosdata/jdbc/ws/BlockResultSet.java b/src/main/java/com/taosdata/jdbc/ws/BlockResultSet.java
@@ -42,15 +42,6 @@ public Object parseValue(int columnIndex) {
         return DataTypeConverUtil.parseValue(type, source);
     }
 
-    public Object parseValueWithZoneId(int columnIndex, ZoneId zoneId) {
-        Object source = result.get(columnIndex - 1).get(rowIndex);
-        if (null == source)
-            return null;
-
-        int type = fields.get(columnIndex - 1).getTaosType();
-        return DataTypeConverUtil.parseValue(type, source);
-    }
-
     @Override
     public String getString(int columnIndex) throws SQLException {
         checkAvailability(columnIndex, fields.size());
@@ -505,15 +496,4 @@ public Statement getStatement() throws SQLException {
     public Timestamp getTimestamp(int columnIndex, Calendar cal) throws SQLException {
         return getTimestamp(columnIndex);
     }
-
-    //    ceil(numOfRows/8.0)
-    private int BitmapLen(int n) {
-        return (n + 0x7) >> 3;
-    }
-
-    private boolean isNull(byte[] c, int n) {
-        int position = n >>> 3;
-        int index = n & 0x7;
-        return (c[position] & (1 << (7 - index))) == (1 << (7 - index));
-    }
 }
diff --git a/src/main/java/com/taosdata/jdbc/ws/Transport.java b/src/main/java/com/taosdata/jdbc/ws/Transport.java
@@ -101,7 +101,7 @@ public void checkServerTrusted(X509Certificate[] certs, String authType) {
         this.connectionParam = param;
         this.wsFunction = function;
 
-        this.timeout = param.getRequestTimeout();
+        setTimeout(param.getRequestTimeout());
     }
 
     public void setTextMessageHandler(Consumer<String> textMessageHandler) {
@@ -117,6 +117,11 @@ public void setBinaryMessageHandler(Consumer<ByteBuffer> binaryMessageHandler) {
     }
 
     public void setTimeout(long timeout) {
+        if (timeout < 0){
+            timeout = DEFAULT_MESSAGE_WAIT_TIMEOUT;
+        } else if (timeout == 0){
+            timeout = Integer.MAX_VALUE;
+        }
         this.timeout = timeout;
     }
 
diff --git a/src/main/java/com/taosdata/jdbc/ws/WSStatement.java b/src/main/java/com/taosdata/jdbc/ws/WSStatement.java
@@ -121,9 +121,6 @@ public void setQueryTimeout(int seconds) throws SQLException {
         if (seconds < 0)
             throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_INVALID_VARIABLE);
 
-        if (seconds == 0){
-            seconds = Integer.MAX_VALUE;
-        }
         this.queryTimeout = seconds;
         transport.setTimeout(seconds * 1000L);
     }
diff --git a/src/main/java/com/taosdata/jdbc/ws/tmq/WSConsumer.java b/src/main/java/com/taosdata/jdbc/ws/tmq/WSConsumer.java
@@ -177,7 +177,7 @@ private ConsumerRecords<V> doPoll(Duration timeout, Deserializer<V> deserializer
         }
 
         ConsumerRecords<V> records = new ConsumerRecords<>();
-        try (WSConsumerResultSet rs = new WSConsumerResultSet(transport, factory, pollResp.getMessageId(), pollResp.getDatabase())) {
+        try (WSConsumerResultSet rs = new WSConsumerResultSet(transport, factory, pollResp.getMessageId(), pollResp.getDatabase(), param.getConnectionParam().getZoneId())) {
             if (deserializer instanceof MapEnhanceDeserializer){
                 ConsumerRecords<TMQEnhMap> resultRecords = rs.handleSubscribeDB(pollResp);
                 return (ConsumerRecords<V>) resultRecords;
diff --git a/src/main/java/com/taosdata/jdbc/ws/tmq/WSConsumerResultSet.java b/src/main/java/com/taosdata/jdbc/ws/tmq/WSConsumerResultSet.java
@@ -51,12 +51,14 @@ public class WSConsumerResultSet extends AbstractResultSet {
 
     protected int numOfRows = 0;
     protected int rowIndex = 0;
+    protected final ZoneId zoneId;
 
-    public WSConsumerResultSet(Transport transport, TMQRequestFactory factory, long messageId, String database) {
+    public WSConsumerResultSet(Transport transport, TMQRequestFactory factory, long messageId, String database, ZoneId zoneId) {
         this.transport = transport;
         this.factory = factory;
         this.messageId = messageId;
         this.database = database;
+        this.zoneId = zoneId;
     }
 
     private boolean forward() {
@@ -114,7 +116,7 @@ public ConsumerRecords<TMQEnhMap> handleSubscribeDB(PollResp pollResp) throws SQ
         if (Code.SUCCESS.getCode() != fetchResp.getCode())
             throw TSDBError.createSQLException(fetchResp.getCode(), fetchResp.getMessage());
 
-        return fetchResp.getEhnMapList(pollResp);
+        return fetchResp.getEhnMapList(pollResp, zoneId);
     }
 
     @Override
@@ -286,7 +288,7 @@ public Timestamp getTimestamp(int columnIndex) throws SQLException {
             return (Timestamp) value;
         if (value instanceof Long) {
             Instant instant = DateTimeUtils.parseTimestampColumnData((long) value, this.timestampPrecision);
-            return DateTimeUtils.getTimestamp(instant, null);
+            return DateTimeUtils.getTimestamp(instant, zoneId);
         }
         Timestamp ret;
         try {
@@ -491,11 +493,8 @@ public Object parseValue(int columnIndex) {
         if (type == TSDB_DATA_TYPE_TIMESTAMP){
             Long o = (Long)DataTypeConverUtil.parseValue(type, source);
             Instant instant = DateTimeUtils.parseTimestampColumnData(o, this.timestampPrecision);
-            return DateTimeUtils.getTimestamp(instant, null);
+            return DateTimeUtils.getTimestamp(instant, zoneId);
         }
         return DataTypeConverUtil.parseValue(type, source);
     }
-
-    //    ceil(numOfRows/8.0)
-
 }
diff --git a/src/main/java/com/taosdata/jdbc/ws/tmq/entity/FetchRawBlockResp.java b/src/main/java/com/taosdata/jdbc/ws/tmq/entity/FetchRawBlockResp.java
@@ -6,6 +6,8 @@
 import com.taosdata.jdbc.rs.RestfulResultSet;
 import com.taosdata.jdbc.rs.RestfulResultSetMetaData;
 import com.taosdata.jdbc.tmq.*;
+import com.taosdata.jdbc.utils.DataTypeConverUtil;
+import com.taosdata.jdbc.utils.DateTimeUtils;
 import com.taosdata.jdbc.utils.DecimalUtil;
 import com.taosdata.jdbc.ws.tmq.ConsumerAction;
 import com.taosdata.jdbc.ws.entity.Response;
@@ -14,6 +16,9 @@
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.ZoneId;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.LinkedList;
@@ -163,7 +168,7 @@ public void parseBlockInfos() throws SQLException {
             rows = resultData.get(0).size();
         }
     }
-    public ConsumerRecords<TMQEnhMap> getEhnMapList(PollResp pollResp) throws SQLException {
+    public ConsumerRecords<TMQEnhMap> getEhnMapList(PollResp pollResp, ZoneId zoneId) throws SQLException {
         skipHead();
         int blockNum = buffer.getInt();
         int cols = 0;
@@ -219,7 +224,15 @@ public ConsumerRecords<TMQEnhMap> getEhnMapList(PollResp pollResp) throws SQLExc
             for (int j = 0; j < lineNum; j++) {
                 HashMap<String, Object> lineDataMap = new HashMap<>();
                 for (int k = 0; k < cols; k++) {
-                    lineDataMap.put(columnNames.get(k), resultData.get(k).get(j));
+                    if (fields.get(k).getTaosType() == TSDB_DATA_TYPE_TIMESTAMP){
+                        Long o = (Long) DataTypeConverUtil.parseValue(TSDB_DATA_TYPE_TIMESTAMP, resultData.get(k).get(j));
+                        Instant instant = DateTimeUtils.parseTimestampColumnData(o, precision);
+                        Timestamp t = DateTimeUtils.getTimestamp(instant, zoneId);
+                        lineDataMap.put(columnNames.get(k), t);
+                        continue;
+                    }
+                    Object o = DataTypeConverUtil.parseValue(fields.get(k).getTaosType(), resultData.get(k).get(j));
+                    lineDataMap.put(columnNames.get(k), o);
                 }
                 TMQEnhMap map = new TMQEnhMap(tableName, lineDataMap);
                 ConsumerRecord<TMQEnhMap> r = new ConsumerRecord.Builder<TMQEnhMap>()
diff --git a/src/test/java/com/taosdata/jdbc/ws/WSConsumerResultSetTest.java b/src/test/java/com/taosdata/jdbc/ws/WSConsumerResultSetTest.java
@@ -16,7 +16,7 @@ public class WSConsumerResultSetTest {
 
     @Before
     public void setUp() {
-        wsConsumerResultSet = new WSConsumerResultSet(null, null, 0, null);
+        wsConsumerResultSet = new WSConsumerResultSet(null, null, 0, null, null);
     }
 
     @Test
diff --git a/src/test/java/com/taosdata/jdbc/ws/WSConsumerSubscribeDBTest.java b/src/test/java/com/taosdata/jdbc/ws/WSConsumerSubscribeDBTest.java
@@ -9,10 +9,9 @@
 import org.junit.Before;
 import org.junit.Test;
 
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.SQLException;
-import java.sql.Statement;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.sql.*;
 import java.time.Duration;
 import java.util.Collections;
 import java.util.Properties;
@@ -24,6 +23,7 @@ public class WSConsumerSubscribeDBTest {
     private static final String dbName = "tmq_ws_enh_test";
     private static final String superTable1 = "st1";
     private static final String superTable2 = "st2";
+    private static final String superTableFullType = "st3";
     private static Connection connection;
     private static Statement statement;
     private static String[] topics = {"topic_ws_map"};
@@ -41,7 +41,7 @@ public void testWSEhnMapDB() throws Exception {
         properties.setProperty(TMQConstants.BOOTSTRAP_SERVERS, "127.0.0.1:6041");
         properties.setProperty(TMQConstants.MSG_WITH_TABLE_NAME, "true");
         properties.setProperty(TMQConstants.ENABLE_AUTO_COMMIT, "true");
-        properties.setProperty(TMQConstants.GROUP_ID, "ws_map");
+        properties.setProperty(TMQConstants.GROUP_ID, "ws_map1");
         properties.setProperty(TMQConstants.VALUE_DESERIALIZER, "com.taosdata.jdbc.tmq.MapEnhanceDeserializer");
         properties.setProperty(TMQConstants.CONNECT_TYPE, "ws");
         properties.setProperty("fetch.max.wait.ms", "5000");
@@ -84,6 +84,90 @@ public void testWSEhnMapDB() throws Exception {
         }
     }
 
+    @Test
+    public void testWSEhnMapDBAllType() throws Exception {
+        AtomicInteger a = new AtomicInteger(1);
+        String topic = topics[0];
+        // create topic
+        statement.executeUpdate("create topic if not exists " + topic + " as database " + dbName);
+
+        Properties properties = new Properties();
+        properties.setProperty(TMQConstants.CONNECT_USER, "root");
+        properties.setProperty(TMQConstants.CONNECT_PASS, "taosdata");
+        properties.setProperty(TMQConstants.BOOTSTRAP_SERVERS, "127.0.0.1:6041");
+        properties.setProperty(TMQConstants.MSG_WITH_TABLE_NAME, "true");
+        properties.setProperty(TMQConstants.ENABLE_AUTO_COMMIT, "true");
+        properties.setProperty(TMQConstants.GROUP_ID, "ws_map2");
+        properties.setProperty(TMQConstants.VALUE_DESERIALIZER, "com.taosdata.jdbc.tmq.MapEnhanceDeserializer");
+        properties.setProperty(TMQConstants.CONNECT_TYPE, "ws");
+        properties.setProperty("fetch.max.wait.ms", "5000");
+        properties.setProperty("min.poll.rows", "1000");
+
+        boolean pass = false;
+        try (TaosConsumer<TMQEnhMap> consumer = new TaosConsumer<>(properties)) {
+            consumer.subscribe(Collections.singletonList(topic));
+            for (int i = 0; i < 10; i++) {
+                ConsumerRecords<TMQEnhMap> consumerRecords = consumer.poll(Duration.ofMillis(100));
+                if (i == 0){
+                    statement.executeUpdate("insert into " + dbName + ".ct0 values(1747474225447, 1, 100, 2.2, 2.3, '1', 12, 2, true, '一', 'POINT(1 1)', '\\x0101', 1.2234, 1747474225448, 255, 65535, 4294967295, 18446744073709551615, -12345678901234567890123.4567890000, 12345678.901234)");
+                }
+                if (consumerRecords.isEmpty()){
+                    continue;
+                }
+                for (ConsumerRecord<TMQEnhMap> r : consumerRecords) {
+                    if (r.value().getTableName().equalsIgnoreCase("ct0")){
+                        Assert.assertEquals(20, r.value().getMap().size());
+
+                        Assert.assertTrue(r.value().getMap().get("ts") instanceof Timestamp);
+                        Assert.assertTrue(r.value().getMap().get("c1") instanceof Integer);
+                        Assert.assertTrue(r.value().getMap().get("c2") instanceof Long);
+                        Assert.assertTrue(r.value().getMap().get("c3") instanceof Float);
+                        Assert.assertTrue(r.value().getMap().get("c4") instanceof Double);
+                        Assert.assertTrue(r.value().getMap().get("c5") instanceof byte[]);
+                        Assert.assertTrue(r.value().getMap().get("c6") instanceof Short);
+                        Assert.assertTrue(r.value().getMap().get("c7") instanceof Byte);
+                        Assert.assertTrue(r.value().getMap().get("c8") instanceof Boolean);
+                        Assert.assertTrue(r.value().getMap().get("c9") instanceof String);
+                        Assert.assertTrue(r.value().getMap().get("c10") instanceof byte[]);
+                        Assert.assertTrue(r.value().getMap().get("c11") instanceof byte[]);
+                        Assert.assertTrue(r.value().getMap().get("c12") instanceof Double);
+                        Assert.assertTrue(r.value().getMap().get("c13") instanceof Timestamp);
+                        Assert.assertTrue(r.value().getMap().get("c14") instanceof Short);
+                        Assert.assertTrue(r.value().getMap().get("c15") instanceof Integer);
+                        Assert.assertTrue(r.value().getMap().get("c16") instanceof Long);
+                        Assert.assertTrue(r.value().getMap().get("c17") instanceof BigInteger);
+                        Assert.assertTrue(r.value().getMap().get("c18") instanceof BigDecimal);
+                        Assert.assertTrue(r.value().getMap().get("c19") instanceof BigDecimal);
+
+                        Assert.assertEquals(new Timestamp(1747474225447L), r.value().getMap().get("ts"));
+                        Assert.assertEquals(r.value().getMap().get("c1"), 1);
+                        Assert.assertEquals(100L, r.value().getMap().get("c2"));
+                        Assert.assertEquals(2.2F, r.value().getMap().get("c3"));
+                        Assert.assertEquals(2.3, r.value().getMap().get("c4"));
+                        Assert.assertArrayEquals("1".getBytes(), (byte[]) r.value().getMap().get("c5"));
+                        Assert.assertEquals((short) 12, r.value().getMap().get("c6"));
+                        Assert.assertEquals((byte) 2, r.value().getMap().get("c7"));
+                        Assert.assertEquals(true, r.value().getMap().get("c8"));
+                        Assert.assertEquals("一", r.value().getMap().get("c9"));
+                        Assert.assertArrayEquals(new byte[]{1, 1}, (byte[]) r.value().getMap().get("c11"));
+                        Assert.assertEquals(1.2234, r.value().getMap().get("c12"));
+                        Assert.assertEquals(new Timestamp(1747474225448L), r.value().getMap().get("c13"));
+                        Assert.assertEquals((short) 255, r.value().getMap().get("c14"));
+                        Assert.assertEquals(65535, r.value().getMap().get("c15"));
+                        Assert.assertEquals(4294967295L, r.value().getMap().get("c16"));
+                        Assert.assertEquals(new BigInteger("18446744073709551615"), r.value().getMap().get("c17"));
+                        Assert.assertEquals(new BigDecimal("-12345678901234567890123.4567890000"), r.value().getMap().get("c18"));
+                        Assert.assertEquals(new BigDecimal("12345678.901234"), r.value().getMap().get("c19"));
+                       pass = true;
+                    }
+                }
+            }
+
+            consumer.unsubscribe();
+            Assert.assertTrue(pass);
+        }
+    }
+
     @Test
     public void testWSEhnMapStable() throws Exception {
         AtomicInteger a = new AtomicInteger(1);
@@ -97,7 +181,7 @@ public void testWSEhnMapStable() throws Exception {
         properties.setProperty(TMQConstants.BOOTSTRAP_SERVERS, "127.0.0.1:6041");
         properties.setProperty(TMQConstants.MSG_WITH_TABLE_NAME, "true");
         properties.setProperty(TMQConstants.ENABLE_AUTO_COMMIT, "true");
-        properties.setProperty(TMQConstants.GROUP_ID, "ws_map");
+        properties.setProperty(TMQConstants.GROUP_ID, "ws_map3");
         properties.setProperty(TMQConstants.VALUE_DESERIALIZER, "com.taosdata.jdbc.tmq.MapEnhanceDeserializer");
         properties.setProperty(TMQConstants.CONNECT_TYPE, "ws");
         properties.setProperty("fetch.max.wait.ms", "5000");
@@ -164,6 +248,12 @@ public void before() throws SQLException {
 
         statement.execute("create stable if not exists " + superTable2
                 + " (ts timestamp, cc1 int) tags(t1 int, t2 int)");
+
+        statement.execute("create stable if not exists " + superTableFullType
+                + " (ts timestamp, c1 int, c2 bigint, c3 float, c4 double, c5 binary(10), c6 SMALLINT, c7 TINYINT, " +
+                "c8 BOOL, c9 nchar(100), c10 GEOMETRY(100), c11 VARBINARY(100), c12 double, c13 timestamp, " +
+                "c14 tinyint unsigned, c15 smallint unsigned, c16 int unsigned, c17 bigint unsigned, c18 decimal(38, 10), c19 decimal(18, 6)) tags(t1 int)");
+        statement.execute("create table if not exists ct0 using " + superTableFullType + " tags(1000)");
     }
 
     @After

Original file line number	Diff line number	Diff line change
`@@ -101,7 +101,7 @@ public void checkServerTrusted(X509Certificate[] certs, String authType) {`
`101`	`101`	`this.connectionParam = param;`
`102`	`102`	`this.wsFunction = function;`
`103`	`103`
`104`		`- this.timeout = param.getRequestTimeout();`
	`104`	`+ setTimeout(param.getRequestTimeout());`
`105`	`105`	`}`
`106`	`106`
`107`	`107`	`public void setTextMessageHandler(Consumer<String> textMessageHandler) {`
`@@ -117,6 +117,11 @@ public void setBinaryMessageHandler(Consumer<ByteBuffer> binaryMessageHandler) {`
`117`	`117`	`}`
`118`	`118`
`119`	`119`	`public void setTimeout(long timeout) {`
	`120`	`+ if (timeout < 0){`
	`121`	`+ timeout = DEFAULT_MESSAGE_WAIT_TIMEOUT;`
	`122`	`+ } else if (timeout == 0){`
	`123`	`+ timeout = Integer.MAX_VALUE;`
	`124`	`+ }`
`120`	`125`	`this.timeout = timeout;`
`121`	`126`	`}`
`122`	`127`
Original file line number	Diff line number	Diff line change
`@@ -121,9 +121,6 @@ public void setQueryTimeout(int seconds) throws SQLException {`
`121`	`121`	`if (seconds < 0)`
`122`	`122`	`throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_INVALID_VARIABLE);`
`123`	`123`
`124`		`- if (seconds == 0){`
`125`		`- seconds = Integer.MAX_VALUE;`
`126`		`- }`
`127`	`124`	`this.queryTimeout = seconds;`
`128`	`125`	`transport.setTimeout(seconds * 1000L);`
`129`	`126`	`}`
Original file line number	Diff line number	Diff line change
`@@ -177,7 +177,7 @@ private ConsumerRecords<V> doPoll(Duration timeout, Deserializer<V> deserializer`
`177`	`177`	`}`
`178`	`178`
`179`	`179`	`ConsumerRecords<V> records = new ConsumerRecords<>();`
`180`		`- try (WSConsumerResultSet rs = new WSConsumerResultSet(transport, factory, pollResp.getMessageId(), pollResp.getDatabase())) {`
	`180`	`+ try (WSConsumerResultSet rs = new WSConsumerResultSet(transport, factory, pollResp.getMessageId(), pollResp.getDatabase(), param.getConnectionParam().getZoneId())) {`
`181`	`181`	`if (deserializer instanceof MapEnhanceDeserializer){`
`182`	`182`	`ConsumerRecords<TMQEnhMap> resultRecords = rs.handleSubscribeDB(pollResp);`
`183`	`183`	`return (ConsumerRecords<V>) resultRecords;`
Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,7 @@ public class WSConsumerResultSetTest {`
`16`	`16`
`17`	`17`	`@Before`
`18`	`18`	`public void setUp() {`
`19`		`- wsConsumerResultSet = new WSConsumerResultSet(null, null, 0, null);`
	`19`	`+ wsConsumerResultSet = new WSConsumerResultSet(null, null, 0, null, null);`
`20`	`20`	`}`
`21`	`21`
`22`	`22`	`@Test`