Skip to content

Commit 13e51e0

Browse files
sheyanjie-qqsheyanjie-qq
authored
feat: supports delete meta subscription (#277)
* feat: supports delete meta subscription * chore: improve meta enum implementation
1 parent c4b41ff commit 13e51e0

12 files changed

Lines changed: 476 additions & 118 deletions

File tree

‎deploy-pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55

66
<groupId>com.taosdata.jdbc</groupId>
77
<artifactId>taos-jdbcdriver</artifactId>
8-
<version>3.7.3</version>
8+
<version>3.7.4</version>
99
<packaging>jar</packaging>
1010

1111
<name>JDBCDriver</name>

‎pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33
<modelVersion>4.0.0</modelVersion>
44
<groupId>com.taosdata.jdbc</groupId>
55
<artifactId>taos-jdbcdriver</artifactId>
6-
<version>3.7.3</version>
6+
<version>3.7.4</version>
77

88
<packaging>jar</packaging>
99
<name>JDBCDriver</name>

‎src/main/java/com/taosdata/jdbc/utils/Utils.java‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -268,4 +268,30 @@ public static void retainByteBuf(ByteBuf byteBuf){
268268
byteBuf.refCnt());
269269
}
270270
}
271+
272+
public static String unescapeUnicode(String input) {
273+
StringBuilder builder = new StringBuilder();
274+
int i = 0;
275+
while (i < input.length()) {
276+
if (i < input.length() - 5 &&
277+
input.charAt(i) == '\\' &&
278+
input.charAt(i + 1) == 'u') {
279+
// extract 4 hex code
280+
String hexCode = input.substring(i + 2, i + 6);
281+
try {
282+
int codePoint = Integer.parseInt(hexCode, 16);
283+
builder.append((char) codePoint);
284+
i += 6; // skip past the unicode escape sequence
285+
} catch (NumberFormatException e) {
286+
// invalid hex code, treat as normal characters
287+
builder.append(input.charAt(i));
288+
i++;
289+
}
290+
} else {
291+
builder.append(input.charAt(i));
292+
i++;
293+
}
294+
}
295+
return builder.toString();
296+
}
271297
}

‎src/main/java/com/taosdata/jdbc/ws/tmq/WSConsumer.java‎

Lines changed: 71 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,6 @@
1515
import com.taosdata.jdbc.ws.FutureResponse;
1616
import com.taosdata.jdbc.ws.InFlightRequest;
1717
import com.taosdata.jdbc.ws.Transport;
18-
import com.taosdata.jdbc.ws.entity.Action;
1918
import com.taosdata.jdbc.ws.entity.Code;
2019
import com.taosdata.jdbc.ws.entity.Request;
2120
import com.taosdata.jdbc.ws.entity.Response;
@@ -26,7 +25,6 @@
2625
import java.nio.ByteOrder;
2726
import java.sql.SQLException;
2827
import java.time.Duration;
29-
import java.time.ZoneId;
3028
import java.util.*;
3129
import java.util.stream.Collectors;
3230

@@ -136,46 +134,22 @@ private boolean handleReconnect() throws SQLException {
136134
}
137135
}
138136

139-
@SuppressWarnings("unchecked")
140-
private ConsumerRecords<V> doPoll(Duration timeout, Deserializer<V> deserializer) throws SQLException{
141-
if (param.isAutoCommit() && (0 != messageId)) {
142-
long now = System.currentTimeMillis();
143-
if (now - lastCommitTime > param.getAutoCommitInterval()) {
144-
commitSync();
145-
lastCommitTime = now;
146-
}
137+
private ConsumerRecords<V> getMeta(PollResp pollResp) throws SQLException{
138+
Request fetchJsonMetaReq = factory.generateFetchJsonMeata(pollResp.getMessageId());
139+
FetchJsonMetaResp fetchJsonMetaResp = (FetchJsonMetaResp) transport.send(fetchJsonMetaReq);
140+
if (Code.SUCCESS.getCode() != fetchJsonMetaResp.getCode()) {
141+
throw new SQLException("consumer fetch json meta error, code: (0x" + Integer.toHexString(fetchJsonMetaResp.getCode()) + "), message: " + fetchJsonMetaResp.getMessage());
147142
}
148143

149-
Request request = factory.generatePoll(lastMessageId, timeout.toMillis());
150-
PollResp pollResp = (PollResp) transport.send(request);
151-
152-
if (Code.SUCCESS.getCode() != pollResp.getCode()) {
153-
throw new SQLException("consumer poll error, code: (0x" + Integer.toHexString(pollResp.getCode()) + "), message: " + pollResp.getMessage());
154-
}
155-
if (!pollResp.isHaveMessage()) {
144+
if (fetchJsonMetaResp.getData() == null || fetchJsonMetaResp.getData().getMetas() == null) {
156145
return ConsumerRecords.emptyRecord();
157146
}
158147

159-
messageId = pollResp.getMessageId();
160-
lastMessageId = messageId;
161-
162-
if (pollResp.getMessageType() == TmqMessageType.TMQ_RES_TABLE_META.getCode() || pollResp.getMessageType() == TmqMessageType.TMQ_RES_METADATA.getCode()) {
163-
Request fetchJsonMetaReq = factory.generateFetchJsonMeata(pollResp.getMessageId());
164-
FetchJsonMetaResp fetchJsonMetaResp = (FetchJsonMetaResp) transport.send(fetchJsonMetaReq);
165-
if (Code.SUCCESS.getCode() != fetchJsonMetaResp.getCode()) {
166-
throw new SQLException("consumer fetch json meta error, code: (0x" + Integer.toHexString(fetchJsonMetaResp.getCode()) + "), message: " + fetchJsonMetaResp.getMessage());
167-
}
168-
169-
if (fetchJsonMetaResp.getData() == null || fetchJsonMetaResp.getData().getMetas() == null) {
170-
return ConsumerRecords.emptyRecord();
171-
}
172-
173-
ConsumerRecords<V> records = new ConsumerRecords<>();
174-
175-
for (Meta meta : fetchJsonMetaResp.getData().getMetas()){
176-
TopicPartition tp = new TopicPartition(pollResp.getTopic(), pollResp.getVgroupId());
148+
ConsumerRecords<V> records = new ConsumerRecords<>();
149+
TopicPartition tp = new TopicPartition(pollResp.getTopic(), pollResp.getVgroupId());
177150

178-
ConsumerRecord<V> r = new ConsumerRecord.Builder<V>()
151+
for (Meta meta : fetchJsonMetaResp.getData().getMetas()){
152+
ConsumerRecord<V> r = new ConsumerRecord.Builder<V>()
179153
.topic(pollResp.getTopic())
180154
.dbName(pollResp.getDatabase())
181155
.vGroupId(pollResp.getVgroupId())
@@ -184,27 +158,24 @@ private ConsumerRecords<V> doPoll(Duration timeout, Deserializer<V> deserializer
184158
.meta(meta)
185159
.value(null)
186160
.build();
187-
records.put(tp, r);
188-
}
189-
return records;
190-
}
191-
192-
if (pollResp.getMessageType() != TmqMessageType.TMQ_RES_DATA.getCode()) {
193-
return ConsumerRecords.emptyRecord();
161+
records.put(tp, r);
194162
}
163+
return records;
164+
}
195165

166+
private ConsumerRecords<V> getData(PollResp pollResp, Deserializer<V> deserializer) throws SQLException{
196167
ConsumerRecords<V> records = new ConsumerRecords<>();
197168
try (WSConsumerResultSet rs = new WSConsumerResultSet(transport, factory, pollResp.getMessageId(), pollResp.getDatabase(), param.getConnectionParam().getZoneId())) {
198169
if (deserializer instanceof MapEnhanceDeserializer){
199170
ConsumerRecords<TMQEnhMap> resultRecords = rs.handleSubscribeDB(pollResp);
200171
return (ConsumerRecords<V>) resultRecords;
201172
}
202-
while (rs.next()) {
203-
String topic = pollResp.getTopic();
204-
String dbName = pollResp.getDatabase();
205-
int vGroupId = pollResp.getVgroupId();
206-
TopicPartition tp = new TopicPartition(topic, vGroupId);
173+
String topic = pollResp.getTopic();
174+
String dbName = pollResp.getDatabase();
175+
int vGroupId = pollResp.getVgroupId();
176+
TopicPartition tp = new TopicPartition(topic, vGroupId);
207177

178+
while (rs.next()) {
208179
V v = deserializer.deserialize(rs, topic, dbName);
209180
ConsumerRecord<V> r = new ConsumerRecord.Builder<V>()
210181
.topic(topic)
@@ -221,6 +192,58 @@ private ConsumerRecords<V> doPoll(Duration timeout, Deserializer<V> deserializer
221192
return records;
222193
}
223194

195+
@SuppressWarnings("unchecked")
196+
private ConsumerRecords<V> doPoll(Duration timeout, Deserializer<V> deserializer) throws SQLException{
197+
if (param.isAutoCommit() && (0 != messageId)) {
198+
long now = System.currentTimeMillis();
199+
if (now - lastCommitTime > param.getAutoCommitInterval()) {
200+
commitSync();
201+
lastCommitTime = now;
202+
}
203+
}
204+
205+
Request request = factory.generatePoll(lastMessageId, timeout.toMillis());
206+
PollResp pollResp = (PollResp) transport.send(request);
207+
208+
if (Code.SUCCESS.getCode() != pollResp.getCode()) {
209+
throw new SQLException("consumer poll error, code: (0x" + Integer.toHexString(pollResp.getCode()) + "), message: " + pollResp.getMessage());
210+
}
211+
if (!pollResp.isHaveMessage()) {
212+
return ConsumerRecords.emptyRecord();
213+
}
214+
215+
messageId = pollResp.getMessageId();
216+
lastMessageId = messageId;
217+
218+
if (pollResp.getMessageType() == TmqMessageType.TMQ_RES_TABLE_META.getCode()){
219+
return getMeta(pollResp);
220+
}
221+
222+
if (pollResp.getMessageType() == TmqMessageType.TMQ_RES_DATA.getCode()){
223+
return getData(pollResp, deserializer);
224+
}
225+
226+
if (pollResp.getMessageType() == TmqMessageType.TMQ_RES_METADATA.getCode()) {
227+
ConsumerRecords<V> metaRecords = getMeta(pollResp);
228+
ConsumerRecords<V> dataRecords = getData(pollResp, deserializer);
229+
if (metaRecords.isEmpty()){
230+
return dataRecords;
231+
}
232+
if (dataRecords.isEmpty()){
233+
return metaRecords;
234+
}
235+
236+
TopicPartition tp = new TopicPartition(pollResp.getTopic(), pollResp.getVgroupId());
237+
238+
for (ConsumerRecord<V> r : metaRecords.get(tp)){
239+
dataRecords.put(tp, r);
240+
}
241+
242+
return dataRecords;
243+
}
244+
return ConsumerRecords.emptyRecord();
245+
}
246+
224247
@Override
225248
public ConsumerRecords<V> poll(Duration timeout, Deserializer<V> deserializer) throws SQLException {
226249

‎src/main/java/com/taosdata/jdbc/ws/tmq/meta/MetaDeleteData.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
package com.taosdata.jdbc.ws.tmq.meta;
2+
3+
import java.util.Objects;
4+
5+
public class MetaDeleteData extends Meta {
6+
private String sql;
7+
8+
public String getSql() {
9+
return sql;
10+
}
11+
12+
public void setSql(String sql) {
13+
this.sql = sql;
14+
}
15+
@Override
16+
public boolean equals(Object o) {
17+
if (this == o) return true;
18+
if (o == null || getClass() != o.getClass()) return false;
19+
if (!super.equals(o)) return false;
20+
MetaDeleteData that = (MetaDeleteData) o;
21+
return Objects.equals(sql, that.sql);
22+
}
23+
24+
@Override
25+
public int hashCode() {
26+
return Objects.hash(super.hashCode(), sql);
27+
}
28+
}

‎src/main/java/com/taosdata/jdbc/ws/tmq/meta/MetaDeserializer.java‎

Lines changed: 33 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
import com.fasterxml.jackson.databind.DeserializationContext;
77
import com.fasterxml.jackson.databind.JsonNode;
88
import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
9+
import com.taosdata.jdbc.utils.Utils;
910

1011
import java.io.IOException;
1112

@@ -35,27 +36,42 @@ public Meta deserialize(JsonParser p, DeserializationContext ctxt) throws IOExce
3536
}
3637

3738
private Meta getMetaBasedOnTableType(ObjectCodec mapper, JsonNode node, String type, String tableType) throws JsonProcessingException {
38-
if (MetaType.CREATE.toString().equalsIgnoreCase(type)) {
39-
if (TableType.SUPER.toString().equalsIgnoreCase(tableType)) {
39+
MetaType metaType = MetaType.fromString(type);
40+
TableType tblType = TableType.fromString(tableType);
41+
42+
switch (metaType) {
43+
case CREATE:
44+
return createMeta(mapper, node, tblType);
45+
case DROP:
46+
return dropMeta(mapper, node, tblType);
47+
case ALTER:
48+
return mapper.treeToValue(node, MetaAlterTable.class);
49+
case DELETE:
50+
MetaDeleteData meta = mapper.treeToValue(node, MetaDeleteData.class);
51+
meta.setSql(Utils.unescapeUnicode(meta.getSql()));
52+
return meta;
53+
default:
54+
throw new IllegalArgumentException("Unsupported MetaType: " + metaType);
55+
}
56+
}
57+
58+
private Meta createMeta(ObjectCodec mapper, JsonNode node, TableType tableType) throws JsonProcessingException {
59+
switch (tableType) {
60+
case SUPER:
4061
return mapper.treeToValue(node, MetaCreateSuperTable.class);
41-
} else if (TableType.NORMAL.toString().equalsIgnoreCase(tableType)) {
62+
case NORMAL:
4263
return mapper.treeToValue(node, MetaCreateNormalTable.class);
43-
} else if (TableType.CHILD.toString().equalsIgnoreCase(tableType)) {
64+
case CHILD:
4465
return mapper.treeToValue(node, MetaCreateChildTable.class);
45-
}
66+
default:
67+
throw new IllegalArgumentException("Unsupported TableType for CREATE: " + tableType);
4668
}
47-
48-
if (MetaType.DROP.toString().equalsIgnoreCase(type)) {
49-
if (TableType.SUPER.toString().equalsIgnoreCase(tableType)) {
50-
return mapper.treeToValue(node, MetaDropSuperTable.class);
51-
} else {
52-
return mapper.treeToValue(node, MetaDropTable.class);
53-
}
54-
}
55-
56-
if (MetaType.ALTER.toString().equalsIgnoreCase(type)) {
57-
return mapper.treeToValue(node, MetaAlterTable.class);
69+
}
70+
private Meta dropMeta(ObjectCodec mapper, JsonNode node, TableType tableType) throws JsonProcessingException {
71+
if (tableType == TableType.SUPER) {
72+
return mapper.treeToValue(node, MetaDropSuperTable.class);
73+
} else {
74+
return mapper.treeToValue(node, MetaDropTable.class);
5875
}
59-
throw new IllegalArgumentException("Unsupported combination of 'type' and 'tableType' values: type=" + type + ", tableType=" + tableType);
6076
}
6177
}

‎src/main/java/com/taosdata/jdbc/ws/tmq/meta/MetaType.java‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,36 @@
11
package com.taosdata.jdbc.ws.tmq.meta;
22

3+
import java.util.HashMap;
4+
import java.util.Map;
5+
36
public enum MetaType {
47
// Type: "CREATE" - Create tables, "DROP" - Drop tables, "ALTER" - Alter a table, "DELETE" - Delete data
58
CREATE,
69
DROP,
710
ALTER,
811
DELETE;
12+
13+
private static final Map<String, MetaType> VALUE_MAP;
14+
static {
15+
VALUE_MAP = new HashMap<>(values().length);
16+
for (MetaType type : values()) {
17+
VALUE_MAP.put(type.name().toUpperCase(), type);
18+
}
19+
}
20+
21+
public static MetaType fromString(String value) {
22+
if (value == null) {
23+
throw new IllegalArgumentException("MetaType value cannot be null");
24+
}
25+
26+
MetaType type = VALUE_MAP.get(value.toUpperCase());
27+
if (type == null) {
28+
throw new IllegalArgumentException("Invalid MetaType value: " + value);
29+
}
30+
return type;
31+
}
32+
33+
public boolean matches(String value) {
34+
return this == fromString(value);
35+
}
936
}

‎src/main/java/com/taosdata/jdbc/ws/tmq/meta/TableType.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,35 @@
11
package com.taosdata.jdbc.ws.tmq.meta;
22

3+
import java.util.HashMap;
4+
import java.util.Map;
5+
36
public enum TableType {
47
SUPER,
58
CHILD,
69
NORMAL;
10+
11+
private static final Map<String, TableType> VALUE_MAP;
12+
13+
static {
14+
VALUE_MAP = new HashMap<>(values().length);
15+
for (TableType type : values()) {
16+
VALUE_MAP.put(type.name().toUpperCase(), type);
17+
}
18+
}
19+
20+
public static TableType fromString(String value) {
21+
if (value == null) {
22+
return null;
23+
}
24+
25+
TableType type = VALUE_MAP.get(value.toUpperCase());
26+
if (type == null) {
27+
throw new IllegalArgumentException("Invalid TableType value: " + value);
28+
}
29+
return type;
30+
}
31+
32+
public boolean matches(String value) {
33+
return this == fromString(value);
34+
}
735
}

0 commit comments

Comments
 (0)