This is an automated email from the ASF dual-hosted git repository.
shuwenwei pushed a commit to branch flink-iotdb-table-connector
in repository https://gitbox.apache.org/repos/asf/iotdb-extras.git
The following commit(s) were added to refs/heads/flink-iotdb-table-connector by
this push:
new 49d8fcb feat: support CDC and lookup cache
49d8fcb is described below
commit 49d8fcba64f827c1ed53eed1f13c5e15969cc461
Author: shuwenwei <[email protected]>
AuthorDate: Wed Sep 23 11:36:49 2026 +0800
feat: support CDC and lookup cache
- add IoTDB subscription based CDC source (scan.mode=cdc)
- unify scan/CDC deserialization via IoTDBDataIterator adapters
- reorganize source packages into common/scan/cdc
- support Flink built-in partial lookup cache
- bump iotdb.version to 2.0.7-SNAPSHOT for iotdb-subscription
---
.../flink-iotdb-table-connector-base/pom.xml | 5 +
.../iotdb/relational/flink/cfg/IoTDBOptions.java | 152 ++++++++++++++++
.../flink/source/cdc/SubscriptionDataIterator.java | 98 ++++++++++
.../source/cdc/client/IoTDBSubscriptionClient.java | 85 +++++++++
.../IoTDBSubscriptionEnumeratorState.java} | 27 ++-
...IoTDBSubscriptionEnumeratorStateSerializer.java | 82 +++++++++
.../split/IoTDBSubscriptionSplit.java} | 55 +++---
.../split/IoTDBSubscriptionSplitSerializer.java | 61 +++++++
.../flink/source/common/IoTDBDataIterator.java | 58 ++++++
.../IoTDBDeserializationSchema.java | 15 +-
.../common/RowDataDeserializationSchema.java | 139 ++++++++++++++
.../deserializer/RowDataDeserializationSchema.java | 114 ------------
.../flink/source/lookup/IoTDBLookupReader.java | 6 +-
.../flink/source/scan/SessionScanDataIterator.java | 151 ++++++++++++++++
.../enumerator/IoTDBSourceEnumeratorState.java | 4 +-
.../IoTDBSourceEnumeratorStateSerializer.java | 6 +-
.../source/{ => scan}/pushdown/AggregateSpec.java | 2 +-
.../pushdown/IoTDBAggregatePushDownUtils.java | 2 +-
.../pushdown/IoTDBExpressionVisitor.java | 2 +-
.../{ => scan}/pushdown/IoTDBLiteralUtils.java | 2 +-
.../source/{ => scan}/split/IoTDBSourceSplit.java | 2 +-
.../split/IoTDBSourceSplitSerializer.java | 2 +-
.../flink/source/cdc/IoTDBCDCSource.java | 92 ++++++++++
.../source/cdc/IoTDBSubscriptionEnumerator.java | 127 +++++++++++++
.../source/cdc/IoTDBSubscriptionSourceReader.java | 200 +++++++++++++++++++++
.../flink/source/{ => scan}/IoTDBSource.java | 14 +-
.../source/{ => scan}/IoTDBSourceEnumerator.java | 8 +-
.../flink/source/{ => scan}/IoTDBSourceReader.java | 11 +-
.../table/IoTDBRelationalDynamicTableFactory.java | 29 ++-
.../table/IoTDBRelationalDynamicTableSource.java | 61 +++++--
pom.xml | 7 +-
31 files changed, 1409 insertions(+), 210 deletions(-)
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/pom.xml
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/pom.xml
index 8c3db7b..7e79c64 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/pom.xml
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/pom.xml
@@ -59,6 +59,11 @@
<artifactId>service-rpc</artifactId>
<version>${iotdb.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-subscription</artifactId>
+ <version>${iotdb.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.tsfile</groupId>
<artifactId>common</artifactId>
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/cfg/IoTDBOptions.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/cfg/IoTDBOptions.java
index 0bc6cf9..7ab12c6 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/cfg/IoTDBOptions.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/cfg/IoTDBOptions.java
@@ -71,6 +71,35 @@ public class IoTDBOptions implements Serializable {
public static final ConfigOption<Integer> LOOKUP_THREAD_SIZE =
ConfigOptions.key("lookup.thread-size").intType().defaultValue(5);
+ public static final ConfigOption<ScanMode> SCAN_MODE =
+
ConfigOptions.key("scan.mode").enumType(ScanMode.class).defaultValue(ScanMode.SNAPSHOT);
+
+ public static final ConfigOption<String> CDC_TOPIC =
+ ConfigOptions.key("cdc.topic").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> CDC_CONSUMER_GROUP =
+ ConfigOptions.key("cdc.consumer-group").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> CDC_MODE =
+ ConfigOptions.key("cdc.mode").stringType().defaultValue("live");
+
+ public static final ConfigOption<String> CDC_START_TIME =
+ ConfigOptions.key("cdc.start-time").stringType().noDefaultValue();
+
+ public static final ConfigOption<Long> CDC_POLL_TIMEOUT_MS =
+ ConfigOptions.key("cdc.poll-timeout-ms").longType().defaultValue(1000L);
+
+ public static final ConfigOption<Boolean> CDC_AUTO_COMMIT =
+ ConfigOptions.key("cdc.auto-commit").booleanType().defaultValue(true);
+
+ private static final String CDC_TOPIC_PREFIX = "flink_iotdb_table_";
+
+ /** Scan mode of the table source. */
+ public enum ScanMode {
+ SNAPSHOT,
+ CDC
+ }
+
private final List<String> nodeUrls;
private final String username;
private final String password;
@@ -82,6 +111,13 @@ public class IoTDBOptions implements Serializable {
private final List<String> attributeColumns;
private final boolean lookupAsync;
private final int lookupThreadSize;
+ private final ScanMode scanMode;
+ private final String cdcTopic;
+ private final String cdcConsumerGroup;
+ private final String cdcMode;
+ private final String cdcStartTime;
+ private final long cdcPollTimeoutMs;
+ private final boolean cdcAutoCommit;
private IoTDBOptions(Builder builder) {
this.nodeUrls = builder.nodeUrls;
@@ -95,6 +131,13 @@ public class IoTDBOptions implements Serializable {
this.attributeColumns = builder.attributeColumns;
this.lookupAsync = builder.lookupAsync;
this.lookupThreadSize = builder.lookupThreadSize;
+ this.scanMode = builder.scanMode;
+ this.cdcTopic = builder.cdcTopic;
+ this.cdcConsumerGroup = builder.cdcConsumerGroup;
+ this.cdcMode = builder.cdcMode;
+ this.cdcStartTime = builder.cdcStartTime;
+ this.cdcPollTimeoutMs = builder.cdcPollTimeoutMs;
+ this.cdcAutoCommit = builder.cdcAutoCommit;
}
/**
@@ -174,6 +217,73 @@ public class IoTDBOptions implements Serializable {
return lookupThreadSize;
}
+ /**
+ * @return the scan mode of the table source.
+ */
+ public ScanMode getScanMode() {
+ return scanMode;
+ }
+
+ /**
+ * @return whether the CDC (subscription) source is used instead of the
bounded snapshot scan.
+ */
+ public boolean isCdc() {
+ return scanMode == ScanMode.CDC;
+ }
+
+ /**
+ * @return the effective CDC topic name. When unset, it is derived from the
connector marker and
+ * the database/table names so the topic origin is recognizable.
+ */
+ public String getCdcTopic() {
+ if (cdcTopic != null && !cdcTopic.isEmpty()) {
+ return cdcTopic;
+ }
+ return CDC_TOPIC_PREFIX + sanitize(database) + "_" + sanitize(table);
+ }
+
+ /**
+ * @return the effective CDC consumer group id. When unset, it is derived
from the topic name.
+ */
+ public String getCdcConsumerGroup() {
+ if (cdcConsumerGroup != null && !cdcConsumerGroup.isEmpty()) {
+ return cdcConsumerGroup;
+ }
+ return getCdcTopic() + "_group";
+ }
+
+ /**
+ * @return the CDC topic mode, e.g. {@code live} or {@code snapshot}.
+ */
+ public String getCdcMode() {
+ return cdcMode;
+ }
+
+ /**
+ * @return the CDC history start time, or {@code null} when unset.
+ */
+ public String getCdcStartTime() {
+ return cdcStartTime;
+ }
+
+ /**
+ * @return the CDC poll timeout in milliseconds.
+ */
+ public long getCdcPollTimeoutMs() {
+ return cdcPollTimeoutMs;
+ }
+
+ /**
+ * @return whether the CDC consumer commits offsets automatically.
+ */
+ public boolean isCdcAutoCommit() {
+ return cdcAutoCommit;
+ }
+
+ private static String sanitize(String value) {
+ return value == null ? "" : value.replaceAll("[^A-Za-z0-9_]", "_");
+ }
+
/**
* @return a new builder
*/
@@ -195,6 +305,13 @@ public class IoTDBOptions implements Serializable {
private List<String> attributeColumns = Collections.emptyList();
private boolean lookupAsync = false;
private int lookupThreadSize = 5;
+ private ScanMode scanMode = ScanMode.SNAPSHOT;
+ private String cdcTopic;
+ private String cdcConsumerGroup;
+ private String cdcMode = "live";
+ private String cdcStartTime;
+ private long cdcPollTimeoutMs = 1000L;
+ private boolean cdcAutoCommit = true;
public Builder withNodeUrls(List<String> nodeUrls) {
this.nodeUrls = nodeUrls;
@@ -251,6 +368,41 @@ public class IoTDBOptions implements Serializable {
return this;
}
+ public Builder withScanMode(ScanMode scanMode) {
+ this.scanMode = scanMode;
+ return this;
+ }
+
+ public Builder withCdcTopic(String cdcTopic) {
+ this.cdcTopic = cdcTopic;
+ return this;
+ }
+
+ public Builder withCdcConsumerGroup(String cdcConsumerGroup) {
+ this.cdcConsumerGroup = cdcConsumerGroup;
+ return this;
+ }
+
+ public Builder withCdcMode(String cdcMode) {
+ this.cdcMode = cdcMode;
+ return this;
+ }
+
+ public Builder withCdcStartTime(String cdcStartTime) {
+ this.cdcStartTime = cdcStartTime;
+ return this;
+ }
+
+ public Builder withCdcPollTimeoutMs(long cdcPollTimeoutMs) {
+ this.cdcPollTimeoutMs = cdcPollTimeoutMs;
+ return this;
+ }
+
+ public Builder withCdcAutoCommit(boolean cdcAutoCommit) {
+ this.cdcAutoCommit = cdcAutoCommit;
+ return this;
+ }
+
/**
* @return the built options
*/
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/SubscriptionDataIterator.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/SubscriptionDataIterator.java
new file mode 100644
index 0000000..1b1b785
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/SubscriptionDataIterator.java
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.relational.flink.source.cdc;
+
+import org.apache.iotdb.relational.flink.source.common.IoTDBDataIterator;
+import
org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler.SubscriptionResultSet;
+
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.util.List;
+
+/** Adapts a subscription {@link SubscriptionResultSet} to the unified {@link
IoTDBDataIterator}. */
+public class SubscriptionDataIterator implements IoTDBDataIterator {
+
+ private final SubscriptionResultSet resultSet;
+
+ public SubscriptionDataIterator(SubscriptionResultSet resultSet) {
+ this.resultSet = resultSet;
+ }
+
+ @Override
+ public boolean next() throws IOException {
+ return resultSet.next();
+ }
+
+ @Override
+ public List<String> getColumnNames() {
+ return resultSet.getColumnNames();
+ }
+
+ @Override
+ public boolean isNull(int columnIndex) {
+ return resultSet.isNull(columnIndex);
+ }
+
+ @Override
+ public boolean getBoolean(int columnIndex) {
+ return resultSet.getBoolean(columnIndex);
+ }
+
+ @Override
+ public int getInt(int columnIndex) {
+ return resultSet.getInt(columnIndex);
+ }
+
+ @Override
+ public long getLong(int columnIndex) {
+ return resultSet.getLong(columnIndex);
+ }
+
+ @Override
+ public float getFloat(int columnIndex) {
+ return resultSet.getFloat(columnIndex);
+ }
+
+ @Override
+ public double getDouble(int columnIndex) {
+ return resultSet.getDouble(columnIndex);
+ }
+
+ @Override
+ public String getString(int columnIndex) {
+ return resultSet.getString(columnIndex);
+ }
+
+ @Override
+ public byte[] getBinary(int columnIndex) {
+ return resultSet.getBinary(columnIndex);
+ }
+
+ @Override
+ public LocalDate getDate(int columnIndex) {
+ return resultSet.getDate(columnIndex);
+ }
+
+ @Override
+ public Timestamp getTimestamp(int columnIndex) {
+ return new Timestamp(resultSet.getLong(columnIndex));
+ }
+}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/client/IoTDBSubscriptionClient.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/client/IoTDBSubscriptionClient.java
new file mode 100644
index 0000000..8df5242
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/client/IoTDBSubscriptionClient.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.relational.flink.source.cdc.client;
+
+import org.apache.iotdb.relational.flink.cfg.IoTDBOptions;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+import org.apache.iotdb.session.subscription.ISubscriptionTableSession;
+import org.apache.iotdb.session.subscription.SubscriptionTableSessionBuilder;
+import
org.apache.iotdb.session.subscription.consumer.ISubscriptionTablePullConsumer;
+import
org.apache.iotdb.session.subscription.consumer.table.SubscriptionTablePullConsumerBuilder;
+
+import java.util.List;
+import java.util.Properties;
+
+/** Creates the IoTDB subscription topic and the pull consumer used by the CDC
source. */
+public final class IoTDBSubscriptionClient {
+
+ private IoTDBSubscriptionClient() {}
+
+ /** Creates the topic if it does not exist, always using the row-level
record format. */
+ public static void createTopicIfNotExists(IoTDBOptions options) throws
Exception {
+ List<String> nodeUrls = options.getNodeUrls();
+ if (nodeUrls == null || nodeUrls.isEmpty()) {
+ throw new IllegalArgumentException("IoTDB nodeUrls must not be empty for
CDC.");
+ }
+ String[] hostPort = splitNodeUrl(nodeUrls.get(0));
+
+ Properties properties = new Properties();
+ properties.setProperty(TopicConstant.DATABASE_KEY, options.getDatabase());
+ properties.setProperty(TopicConstant.TABLE_KEY, options.getTable());
+ properties.setProperty(TopicConstant.MODE_KEY, options.getCdcMode());
+ properties.setProperty(TopicConstant.FORMAT_KEY,
TopicConstant.FORMAT_RECORD_HANDLER_VALUE);
+ if (options.getCdcStartTime() != null &&
!options.getCdcStartTime().isEmpty()) {
+ properties.setProperty(TopicConstant.START_TIME_KEY,
options.getCdcStartTime());
+ }
+
+ try (ISubscriptionTableSession session =
+ new SubscriptionTableSessionBuilder()
+ .host(hostPort[0])
+ .port(Integer.parseInt(hostPort[1]))
+ .username(options.getUsername())
+ .password(options.getPassword())
+ .build()) {
+ session.open();
+ session.createTopicIfNotExists(options.getCdcTopic(), properties);
+ }
+ }
+
+ /** Builds a pull consumer bound to the configured consumer group. */
+ public static ISubscriptionTablePullConsumer createPullConsumer(IoTDBOptions
options) {
+ return new SubscriptionTablePullConsumerBuilder()
+ .nodeUrls(options.getNodeUrls())
+ .username(options.getUsername())
+ .password(options.getPassword())
+ .consumerGroupId(options.getCdcConsumerGroup())
+ .autoCommit(options.isCdcAutoCommit())
+ .build();
+ }
+
+ private static String[] splitNodeUrl(String nodeUrl) {
+ String[] parts = nodeUrl.split(":");
+ if (parts.length != 2) {
+ throw new IllegalArgumentException(
+ "IoTDB node url must be in the format host:port, but was: " +
nodeUrl);
+ }
+ return parts;
+ }
+}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/enumerator/IoTDBSourceEnumeratorState.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/enumerator/IoTDBSubscriptionEnumeratorState.java
similarity index 53%
copy from
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/enumerator/IoTDBSourceEnumeratorState.java
copy to
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/enumerator/IoTDBSubscriptionEnumeratorState.java
index 884ebb7..667f829 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/enumerator/IoTDBSourceEnumeratorState.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/enumerator/IoTDBSubscriptionEnumeratorState.java
@@ -17,26 +17,35 @@
* under the License.
*/
-package org.apache.iotdb.relational.flink.source.enumerator;
+package org.apache.iotdb.relational.flink.source.cdc.enumerator;
-import org.apache.iotdb.relational.flink.source.split.IoTDBSourceSplit;
+import
org.apache.iotdb.relational.flink.source.cdc.split.IoTDBSubscriptionSplit;
import java.io.Serializable;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-/** Checkpoint state of the IoTDB source enumerator. */
-public class IoTDBSourceEnumeratorState implements Serializable {
+/** Checkpoint state of the CDC enumerator. */
+public class IoTDBSubscriptionEnumeratorState implements Serializable {
private static final long serialVersionUID = 1L;
- private final List<IoTDBSourceSplit> remainingSplits;
+ private final boolean allSplitsCreated;
+ private final List<IoTDBSubscriptionSplit> remainingSplits;
- public IoTDBSourceEnumeratorState(List<IoTDBSourceSplit> remainingSplits) {
- this.remainingSplits = remainingSplits;
+ public IoTDBSubscriptionEnumeratorState(
+ boolean allSplitsCreated, List<IoTDBSubscriptionSplit> remainingSplits) {
+ this.allSplitsCreated = allSplitsCreated;
+ this.remainingSplits =
+ remainingSplits == null ? Collections.emptyList() : new
ArrayList<>(remainingSplits);
}
- public List<IoTDBSourceSplit> getRemainingSplits() {
- return remainingSplits == null ? Collections.emptyList() : remainingSplits;
+ public boolean isAllSplitsCreated() {
+ return allSplitsCreated;
+ }
+
+ public List<IoTDBSubscriptionSplit> getRemainingSplits() {
+ return remainingSplits;
}
}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/enumerator/IoTDBSubscriptionEnumeratorStateSerializer.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/enumerator/IoTDBSubscriptionEnumeratorStateSerializer.java
new file mode 100644
index 0000000..c6353d5
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/enumerator/IoTDBSubscriptionEnumeratorStateSerializer.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.relational.flink.source.cdc.enumerator;
+
+import
org.apache.iotdb.relational.flink.source.cdc.split.IoTDBSubscriptionSplit;
+import
org.apache.iotdb.relational.flink.source.cdc.split.IoTDBSubscriptionSplitSerializer;
+
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/** Serializer for {@link IoTDBSubscriptionEnumeratorState}. */
+public class IoTDBSubscriptionEnumeratorStateSerializer
+ implements SimpleVersionedSerializer<IoTDBSubscriptionEnumeratorState> {
+
+ private static final int VERSION = 1;
+
+ private final IoTDBSubscriptionSplitSerializer splitSerializer =
+ new IoTDBSubscriptionSplitSerializer();
+
+ @Override
+ public int getVersion() {
+ return VERSION;
+ }
+
+ @Override
+ public byte[] serialize(IoTDBSubscriptionEnumeratorState state) throws
IOException {
+ try (ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(buffer)) {
+ out.writeBoolean(state.isAllSplitsCreated());
+ List<IoTDBSubscriptionSplit> splits = state.getRemainingSplits();
+ out.writeInt(splits.size());
+ for (IoTDBSubscriptionSplit split : splits) {
+ byte[] bytes = splitSerializer.serialize(split);
+ out.writeInt(bytes.length);
+ out.write(bytes);
+ }
+ out.flush();
+ return buffer.toByteArray();
+ }
+ }
+
+ @Override
+ public IoTDBSubscriptionEnumeratorState deserialize(int version, byte[]
serialized)
+ throws IOException {
+ try (ByteArrayInputStream buffer = new ByteArrayInputStream(serialized);
+ DataInputStream in = new DataInputStream(buffer)) {
+ boolean allSplitsCreated = in.readBoolean();
+ int size = in.readInt();
+ List<IoTDBSubscriptionSplit> splits = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ byte[] bytes = new byte[in.readInt()];
+ in.readFully(bytes);
+ splits.add(splitSerializer.deserialize(splitSerializer.getVersion(),
bytes));
+ }
+ return new IoTDBSubscriptionEnumeratorState(allSplitsCreated, splits);
+ }
+ }
+}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/split/IoTDBSourceSplit.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/split/IoTDBSubscriptionSplit.java
similarity index 53%
copy from
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/split/IoTDBSourceSplit.java
copy to
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/split/IoTDBSubscriptionSplit.java
index 23dcf5f..07f8783 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/split/IoTDBSourceSplit.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/split/IoTDBSubscriptionSplit.java
@@ -17,45 +17,36 @@
* under the License.
*/
-package org.apache.iotdb.relational.flink.source.split;
+package org.apache.iotdb.relational.flink.source.cdc.split;
import org.apache.flink.api.connector.source.SourceSplit;
-import java.io.Serializable;
import java.util.Objects;
-/** Read split for the IoTDB relational table source. */
-public class IoTDBSourceSplit implements SourceSplit, Serializable {
+/** A CDC split identifying the IoTDB subscription topic and consumer group to
consume. */
+public class IoTDBSubscriptionSplit implements SourceSplit {
private static final long serialVersionUID = 1L;
- private final String splitId;
- private final String database;
- private final String table;
- private final String sql;
+ private final String topic;
+ private final String consumerGroup;
- public IoTDBSourceSplit(String splitId, String database, String table,
String sql) {
- this.splitId = splitId;
- this.database = database;
- this.table = table;
- this.sql = sql;
+ public IoTDBSubscriptionSplit(String topic, String consumerGroup) {
+ this.topic = topic;
+ this.consumerGroup = consumerGroup;
}
- @Override
- public String splitId() {
- return splitId;
+ public String getTopic() {
+ return topic;
}
- public String getDatabase() {
- return database;
+ public String getConsumerGroup() {
+ return consumerGroup;
}
- public String getTable() {
- return table;
- }
-
- public String getSql() {
- return sql;
+ @Override
+ public String splitId() {
+ return topic + ":" + consumerGroup;
}
@Override
@@ -63,18 +54,20 @@ public class IoTDBSourceSplit implements SourceSplit,
Serializable {
if (this == o) {
return true;
}
- if (!(o instanceof IoTDBSourceSplit)) {
+ if (o == null || getClass() != o.getClass()) {
return false;
}
- IoTDBSourceSplit that = (IoTDBSourceSplit) o;
- return Objects.equals(splitId, that.splitId)
- && Objects.equals(database, that.database)
- && Objects.equals(table, that.table)
- && Objects.equals(sql, that.sql);
+ IoTDBSubscriptionSplit that = (IoTDBSubscriptionSplit) o;
+ return Objects.equals(topic, that.topic) && Objects.equals(consumerGroup,
that.consumerGroup);
}
@Override
public int hashCode() {
- return Objects.hash(splitId, database, table, sql);
+ return Objects.hash(topic, consumerGroup);
+ }
+
+ @Override
+ public String toString() {
+ return "IoTDBSubscriptionSplit{topic='" + topic + "', consumerGroup='" +
consumerGroup + "'}";
}
}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/split/IoTDBSubscriptionSplitSerializer.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/split/IoTDBSubscriptionSplitSerializer.java
new file mode 100644
index 0000000..444c617
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/split/IoTDBSubscriptionSplitSerializer.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.relational.flink.source.cdc.split;
+
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+/** Serializer for {@link IoTDBSubscriptionSplit}. */
+public class IoTDBSubscriptionSplitSerializer
+ implements SimpleVersionedSerializer<IoTDBSubscriptionSplit> {
+
+ private static final int VERSION = 1;
+
+ @Override
+ public int getVersion() {
+ return VERSION;
+ }
+
+ @Override
+ public byte[] serialize(IoTDBSubscriptionSplit split) throws IOException {
+ try (ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(buffer)) {
+ out.writeUTF(split.getTopic());
+ out.writeUTF(split.getConsumerGroup());
+ out.flush();
+ return buffer.toByteArray();
+ }
+ }
+
+ @Override
+ public IoTDBSubscriptionSplit deserialize(int version, byte[] serialized)
throws IOException {
+ try (ByteArrayInputStream buffer = new ByteArrayInputStream(serialized);
+ DataInputStream in = new DataInputStream(buffer)) {
+ String topic = in.readUTF();
+ String consumerGroup = in.readUTF();
+ return new IoTDBSubscriptionSplit(topic, consumerGroup);
+ }
+ }
+}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/IoTDBDataIterator.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/IoTDBDataIterator.java
new file mode 100644
index 0000000..c334dd8
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/IoTDBDataIterator.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.relational.flink.source.common;
+
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.util.List;
+
+/**
+ * Unified iterator-style row cursor shared by the scan and CDC sources.
+ *
+ * <p>The caller advances with {@link #next()} and then reads the current row
by column index. The
+ * index is always 0-based, regardless of the underlying IoTDB API.
+ */
+public interface IoTDBDataIterator {
+
+ boolean next() throws IOException;
+
+ List<String> getColumnNames();
+
+ boolean isNull(int columnIndex) throws IOException;
+
+ boolean getBoolean(int columnIndex) throws IOException;
+
+ int getInt(int columnIndex) throws IOException;
+
+ long getLong(int columnIndex) throws IOException;
+
+ float getFloat(int columnIndex) throws IOException;
+
+ double getDouble(int columnIndex) throws IOException;
+
+ String getString(int columnIndex) throws IOException;
+
+ byte[] getBinary(int columnIndex) throws IOException;
+
+ LocalDate getDate(int columnIndex) throws IOException;
+
+ Timestamp getTimestamp(int columnIndex) throws IOException;
+}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/deserializer/IoTDBDeserializationSchema.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/IoTDBDeserializationSchema.java
similarity index 65%
rename from
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/deserializer/IoTDBDeserializationSchema.java
rename to
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/IoTDBDeserializationSchema.java
index a9f762e..c2a9710 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/deserializer/IoTDBDeserializationSchema.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/IoTDBDeserializationSchema.java
@@ -17,24 +17,21 @@
* under the License.
*/
-package org.apache.iotdb.relational.flink.source.deserializer;
-
-import org.apache.iotdb.isession.SessionDataSet;
+package org.apache.iotdb.relational.flink.source.common;
import java.io.IOException;
import java.io.Serializable;
/**
- * Converts the current row of an IoTDB {@link SessionDataSet.DataIterator}
into the output type of
- * the Flink source.
+ * Converts the current row of an {@link IoTDBDataIterator} into the output
type of the Flink
+ * source.
*
- * <p>The caller owns iteration and must invoke {@link
SessionDataSet.DataIterator#next()} before
- * calling this method. Implementations must only read the current row and
must not advance the
- * iterator.
+ * <p>The caller owns iteration and must invoke {@link
IoTDBDataIterator#next()} before calling this
+ * method. Implementations must only read the current row and must not advance
the iterator.
*
* @param <OUT> output type
*/
public interface IoTDBDeserializationSchema<OUT> extends Serializable {
- OUT deserialize(SessionDataSet.DataIterator iterator) throws IOException;
+ OUT deserialize(IoTDBDataIterator iterator) throws IOException;
}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/RowDataDeserializationSchema.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/RowDataDeserializationSchema.java
new file mode 100644
index 0000000..68c7384
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/RowDataDeserializationSchema.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.relational.flink.source.common;
+
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.types.RowKind;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+
+/**
+ * Table API deserializer that converts the current row of an {@link
IoTDBDataIterator} into {@link
+ * RowData}. Columns are matched to the declared row type by name
(case-insensitive) and fall back
+ * to the same positional index when no name matches.
+ */
+public class RowDataDeserializationSchema implements
IoTDBDeserializationSchema<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final DataType rowDataType;
+ private final List<String> fieldNames;
+ private final List<DataType> fieldTypes;
+
+ private transient List<String> cachedColumnNames;
+ private transient int[] cachedColumnIndexes;
+
+ public RowDataDeserializationSchema(DataType rowDataType) {
+ if (rowDataType.getLogicalType().getTypeRoot() != LogicalTypeRoot.ROW) {
+ throw new IllegalArgumentException("RowDataDeserializationSchema
requires a ROW data type.");
+ }
+ this.rowDataType = rowDataType;
+ this.fieldNames = DataType.getFieldNames(rowDataType);
+ this.fieldTypes = DataType.getFieldDataTypes(rowDataType);
+ }
+
+ @Override
+ public RowData deserialize(IoTDBDataIterator iterator) throws IOException {
+ int[] columnIndexes = resolveColumnIndexes(iterator.getColumnNames());
+ GenericRowData row = new GenericRowData(RowKind.INSERT, fieldNames.size());
+ for (int i = 0; i < fieldNames.size(); i++) {
+ int columnIndex = columnIndexes[i];
+ row.setField(i, columnIndex < 0 ? null : readField(iterator,
columnIndex, fieldTypes.get(i)));
+ }
+ return row;
+ }
+
+ public DataType getRowDataType() {
+ return rowDataType;
+ }
+
+ private int[] resolveColumnIndexes(List<String> columnNames) {
+ if (cachedColumnNames != null && cachedColumnNames.equals(columnNames)) {
+ return cachedColumnIndexes;
+ }
+ int[] indexes = new int[fieldNames.size()];
+ for (int i = 0; i < fieldNames.size(); i++) {
+ int found = indexOfIgnoreCase(columnNames, fieldNames.get(i));
+ if (found < 0 && i < columnNames.size()) {
+ found = i;
+ }
+ indexes[i] = found;
+ }
+ cachedColumnNames = columnNames;
+ cachedColumnIndexes = indexes;
+ return indexes;
+ }
+
+ private static int indexOfIgnoreCase(List<String> columnNames, String name) {
+ String normalized = name.toLowerCase(Locale.ROOT);
+ for (int i = 0; i < columnNames.size(); i++) {
+ if (columnNames.get(i).toLowerCase(Locale.ROOT).equals(normalized)) {
+ return i;
+ }
+ }
+ return -1;
+ }
+
+ private static Object readField(IoTDBDataIterator iterator, int columnIndex,
DataType dataType)
+ throws IOException {
+ if (iterator.isNull(columnIndex)) {
+ return null;
+ }
+
+ LogicalTypeRoot typeRoot = dataType.getLogicalType().getTypeRoot();
+ switch (typeRoot) {
+ case BOOLEAN:
+ return iterator.getBoolean(columnIndex);
+ case TINYINT:
+ return (byte) iterator.getInt(columnIndex);
+ case SMALLINT:
+ return (short) iterator.getInt(columnIndex);
+ case INTEGER:
+ return iterator.getInt(columnIndex);
+ case BIGINT:
+ return iterator.getLong(columnIndex);
+ case FLOAT:
+ return iterator.getFloat(columnIndex);
+ case DOUBLE:
+ return iterator.getDouble(columnIndex);
+ case CHAR:
+ case VARCHAR:
+ return StringData.fromString(iterator.getString(columnIndex));
+ case BINARY:
+ case VARBINARY:
+ return iterator.getBinary(columnIndex);
+ case DATE:
+ return (int) iterator.getDate(columnIndex).toEpochDay();
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return TimestampData.fromTimestamp(iterator.getTimestamp(columnIndex));
+ default:
+ throw new IOException(
+ "Unsupported Flink type at column " + columnIndex + ": " +
dataType.getLogicalType());
+ }
+ }
+}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/deserializer/RowDataDeserializationSchema.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/deserializer/RowDataDeserializationSchema.java
deleted file mode 100644
index 14a953a..0000000
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/deserializer/RowDataDeserializationSchema.java
+++ /dev/null
@@ -1,114 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.relational.flink.source.deserializer;
-
-import org.apache.iotdb.isession.SessionDataSet;
-import org.apache.iotdb.rpc.StatementExecutionException;
-
-import org.apache.flink.table.data.GenericRowData;
-import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.data.StringData;
-import org.apache.flink.table.data.TimestampData;
-import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.types.logical.LogicalTypeRoot;
-import org.apache.flink.types.RowKind;
-import org.apache.tsfile.utils.Binary;
-
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.List;
-
-/**
- * Table API deserializer that converts the current {@link
SessionDataSet.DataIterator} row into
- * {@link RowData}.
- */
-public class RowDataDeserializationSchema implements
IoTDBDeserializationSchema<RowData> {
-
- private static final long serialVersionUID = 1L;
-
- private final DataType rowDataType;
- private final List<DataType> fieldDataTypes;
-
- public RowDataDeserializationSchema(DataType rowDataType) {
- if (rowDataType.getLogicalType().getTypeRoot() != LogicalTypeRoot.ROW) {
- throw new IllegalArgumentException("RowDataDeserializationSchema
requires a ROW data type.");
- }
- this.rowDataType = rowDataType;
- this.fieldDataTypes = DataType.getFieldDataTypes(rowDataType);
- }
-
- @Override
- public RowData deserialize(SessionDataSet.DataIterator iterator) throws
IOException {
- GenericRowData row = new GenericRowData(RowKind.INSERT,
fieldDataTypes.size());
- for (int i = 0; i < fieldDataTypes.size(); i++) {
- row.setField(i, readField(iterator, i + 1, fieldDataTypes.get(i)));
- }
- return row;
- }
-
- public DataType getRowDataType() {
- return rowDataType;
- }
-
- private static Object readField(
- SessionDataSet.DataIterator iterator, int columnIndex, DataType
dataType) throws IOException {
- try {
- if (iterator.isNull(columnIndex)) {
- return null;
- }
-
- LogicalTypeRoot typeRoot = dataType.getLogicalType().getTypeRoot();
- switch (typeRoot) {
- case BOOLEAN:
- return iterator.getBoolean(columnIndex);
- case TINYINT:
- return (byte) iterator.getInt(columnIndex);
- case SMALLINT:
- return (short) iterator.getInt(columnIndex);
- case INTEGER:
- return iterator.getInt(columnIndex);
- case BIGINT:
- return iterator.getLong(columnIndex);
- case FLOAT:
- return iterator.getFloat(columnIndex);
- case DOUBLE:
- return iterator.getDouble(columnIndex);
- case CHAR:
- case VARCHAR:
- return StringData.fromString(iterator.getString(columnIndex));
- case BINARY:
- case VARBINARY:
- Binary binary = iterator.getBlob(columnIndex);
- return binary == null ? null : Arrays.copyOf(binary.getValues(),
binary.getLength());
- case DATE:
- return (int) iterator.getDate(columnIndex).toEpochDay();
- case TIMESTAMP_WITHOUT_TIME_ZONE:
- case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
- return
TimestampData.fromTimestamp(iterator.getTimestamp(columnIndex));
- default:
- throw new IOException(
- "Unsupported Flink type at column " + columnIndex + ": " +
dataType.getLogicalType());
- }
- } catch (StatementExecutionException e) {
- throw new IOException(
- "Failed to read IoTDB column " + columnIndex + " as " +
dataType.getLogicalType(), e);
- }
- }
-}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupReader.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupReader.java
index cd4e6ad..eb973c4 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupReader.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupReader.java
@@ -23,7 +23,9 @@ import org.apache.iotdb.isession.ITableSession;
import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.isession.pool.ITableSessionPool;
import org.apache.iotdb.relational.flink.cfg.IoTDBOptions;
-import
org.apache.iotdb.relational.flink.source.deserializer.RowDataDeserializationSchema;
+import org.apache.iotdb.relational.flink.source.common.IoTDBDataIterator;
+import
org.apache.iotdb.relational.flink.source.common.RowDataDeserializationSchema;
+import org.apache.iotdb.relational.flink.source.scan.SessionScanDataIterator;
import org.apache.iotdb.relational.flink.utils.IoTDBUtils;
import org.apache.iotdb.session.pool.TableSessionPoolBuilder;
@@ -98,7 +100,7 @@ public class IoTDBLookupReader implements AutoCloseable {
try (ITableSession session = pool.getSession();
SessionDataSet dataSet = session.executeQueryStatement(sql)) {
List<RowData> rows = new ArrayList<>();
- SessionDataSet.DataIterator iterator = dataSet.iterator();
+ IoTDBDataIterator iterator = new
SessionScanDataIterator(dataSet.iterator());
while (iterator.next()) {
rows.add(deserializer.deserialize(iterator));
}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/SessionScanDataIterator.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/SessionScanDataIterator.java
new file mode 100644
index 0000000..a9bb2a4
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/SessionScanDataIterator.java
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.relational.flink.source.scan;
+
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.relational.flink.source.common.IoTDBDataIterator;
+
+import org.apache.tsfile.utils.Binary;
+
+import java.io.IOException;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Adapts a scan {@link SessionDataSet.DataIterator} (1-based) to the unified
{@link
+ * IoTDBDataIterator} (0-based).
+ */
+public class SessionScanDataIterator implements IoTDBDataIterator {
+
+ private final SessionDataSet.DataIterator iterator;
+ private final List<String> columnNames;
+
+ public SessionScanDataIterator(SessionDataSet.DataIterator iterator) {
+ this.iterator = iterator;
+ this.columnNames = iterator.getColumnNameList();
+ }
+
+ @Override
+ public boolean next() throws IOException {
+ try {
+ return iterator.next();
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public List<String> getColumnNames() {
+ return columnNames;
+ }
+
+ @Override
+ public boolean isNull(int columnIndex) throws IOException {
+ try {
+ return iterator.isNull(columnIndex + 1);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public boolean getBoolean(int columnIndex) throws IOException {
+ try {
+ return iterator.getBoolean(columnIndex + 1);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public int getInt(int columnIndex) throws IOException {
+ try {
+ return iterator.getInt(columnIndex + 1);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public long getLong(int columnIndex) throws IOException {
+ try {
+ return iterator.getLong(columnIndex + 1);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public float getFloat(int columnIndex) throws IOException {
+ try {
+ return iterator.getFloat(columnIndex + 1);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public double getDouble(int columnIndex) throws IOException {
+ try {
+ return iterator.getDouble(columnIndex + 1);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public String getString(int columnIndex) throws IOException {
+ try {
+ return iterator.getString(columnIndex + 1);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public byte[] getBinary(int columnIndex) throws IOException {
+ try {
+ Binary binary = iterator.getBlob(columnIndex + 1);
+ return binary == null ? null : Arrays.copyOf(binary.getValues(),
binary.getLength());
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public LocalDate getDate(int columnIndex) throws IOException {
+ try {
+ return iterator.getDate(columnIndex + 1);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public Timestamp getTimestamp(int columnIndex) throws IOException {
+ try {
+ return iterator.getTimestamp(columnIndex + 1);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/enumerator/IoTDBSourceEnumeratorState.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/enumerator/IoTDBSourceEnumeratorState.java
similarity index 90%
rename from
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/enumerator/IoTDBSourceEnumeratorState.java
rename to
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/enumerator/IoTDBSourceEnumeratorState.java
index 884ebb7..c2d48b0 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/enumerator/IoTDBSourceEnumeratorState.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/enumerator/IoTDBSourceEnumeratorState.java
@@ -17,9 +17,9 @@
* under the License.
*/
-package org.apache.iotdb.relational.flink.source.enumerator;
+package org.apache.iotdb.relational.flink.source.scan.enumerator;
-import org.apache.iotdb.relational.flink.source.split.IoTDBSourceSplit;
+import org.apache.iotdb.relational.flink.source.scan.split.IoTDBSourceSplit;
import java.io.Serializable;
import java.util.Collections;
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/enumerator/IoTDBSourceEnumeratorStateSerializer.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/enumerator/IoTDBSourceEnumeratorStateSerializer.java
similarity index 92%
rename from
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/enumerator/IoTDBSourceEnumeratorStateSerializer.java
rename to
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/enumerator/IoTDBSourceEnumeratorStateSerializer.java
index 67b559e..cea64f9 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/enumerator/IoTDBSourceEnumeratorStateSerializer.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/enumerator/IoTDBSourceEnumeratorStateSerializer.java
@@ -17,10 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.relational.flink.source.enumerator;
+package org.apache.iotdb.relational.flink.source.scan.enumerator;
-import org.apache.iotdb.relational.flink.source.split.IoTDBSourceSplit;
-import
org.apache.iotdb.relational.flink.source.split.IoTDBSourceSplitSerializer;
+import org.apache.iotdb.relational.flink.source.scan.split.IoTDBSourceSplit;
+import
org.apache.iotdb.relational.flink.source.scan.split.IoTDBSourceSplitSerializer;
import org.apache.flink.core.io.SimpleVersionedSerializer;
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/pushdown/AggregateSpec.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/AggregateSpec.java
similarity index 96%
rename from
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/pushdown/AggregateSpec.java
rename to
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/AggregateSpec.java
index d2b4644..2961e67 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/pushdown/AggregateSpec.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/AggregateSpec.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.relational.flink.source.pushdown;
+package org.apache.iotdb.relational.flink.source.scan.pushdown;
import java.io.Serializable;
import java.util.ArrayList;
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/pushdown/IoTDBAggregatePushDownUtils.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBAggregatePushDownUtils.java
similarity index 99%
rename from
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/pushdown/IoTDBAggregatePushDownUtils.java
rename to
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBAggregatePushDownUtils.java
index 0abb7b3..3e74d95 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/pushdown/IoTDBAggregatePushDownUtils.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBAggregatePushDownUtils.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.relational.flink.source.pushdown;
+package org.apache.iotdb.relational.flink.source.scan.pushdown;
import org.apache.iotdb.relational.flink.utils.IoTDBUtils;
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/pushdown/IoTDBExpressionVisitor.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBExpressionVisitor.java
similarity index 99%
rename from
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/pushdown/IoTDBExpressionVisitor.java
rename to
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBExpressionVisitor.java
index 9d3834d..e40e3cf 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/pushdown/IoTDBExpressionVisitor.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBExpressionVisitor.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.relational.flink.source.pushdown;
+package org.apache.iotdb.relational.flink.source.scan.pushdown;
import org.apache.iotdb.relational.flink.utils.IoTDBUtils;
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/pushdown/IoTDBLiteralUtils.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBLiteralUtils.java
similarity index 98%
rename from
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/pushdown/IoTDBLiteralUtils.java
rename to
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBLiteralUtils.java
index 066093f..d8e7968 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/pushdown/IoTDBLiteralUtils.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBLiteralUtils.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.relational.flink.source.pushdown;
+package org.apache.iotdb.relational.flink.source.scan.pushdown;
import org.apache.flink.table.expressions.ValueLiteralExpression;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/split/IoTDBSourceSplit.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/split/IoTDBSourceSplit.java
similarity index 97%
rename from
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/split/IoTDBSourceSplit.java
rename to
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/split/IoTDBSourceSplit.java
index 23dcf5f..fa0f593 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/split/IoTDBSourceSplit.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/split/IoTDBSourceSplit.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.relational.flink.source.split;
+package org.apache.iotdb.relational.flink.source.scan.split;
import org.apache.flink.api.connector.source.SourceSplit;
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/split/IoTDBSourceSplitSerializer.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/split/IoTDBSourceSplitSerializer.java
similarity index 97%
rename from
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/split/IoTDBSourceSplitSerializer.java
rename to
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/split/IoTDBSourceSplitSerializer.java
index 2cf453f..aaf1343 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/split/IoTDBSourceSplitSerializer.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/split/IoTDBSourceSplitSerializer.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.relational.flink.source.split;
+package org.apache.iotdb.relational.flink.source.scan.split;
import org.apache.flink.core.io.SimpleVersionedSerializer;
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBCDCSource.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBCDCSource.java
new file mode 100644
index 0000000..379b216
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBCDCSource.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.relational.flink.source.cdc;
+
+import org.apache.iotdb.relational.flink.cfg.IoTDBOptions;
+import
org.apache.iotdb.relational.flink.source.cdc.enumerator.IoTDBSubscriptionEnumeratorState;
+import
org.apache.iotdb.relational.flink.source.cdc.enumerator.IoTDBSubscriptionEnumeratorStateSerializer;
+import
org.apache.iotdb.relational.flink.source.cdc.split.IoTDBSubscriptionSplit;
+import
org.apache.iotdb.relational.flink.source.cdc.split.IoTDBSubscriptionSplitSerializer;
+import
org.apache.iotdb.relational.flink.source.common.IoTDBDeserializationSchema;
+import org.apache.iotdb.relational.flink.source.scan.IoTDBSource;
+
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+
+/**
+ * Unbounded CDC source backed by the IoTDB subscription API.
+ *
+ * <p>Unlike the bounded {@link IoTDBSource}, this source never finishes on
its own.
+ *
+ * @param <OUT> source output type
+ */
+public class IoTDBCDCSource<OUT>
+ implements Source<OUT, IoTDBSubscriptionSplit,
IoTDBSubscriptionEnumeratorState> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final IoTDBOptions options;
+ private final IoTDBDeserializationSchema<OUT> deserializer;
+
+ public IoTDBCDCSource(IoTDBOptions options, IoTDBDeserializationSchema<OUT>
deserializer) {
+ this.options = options;
+ this.deserializer = deserializer;
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.CONTINUOUS_UNBOUNDED;
+ }
+
+ @Override
+ public SourceReader<OUT, IoTDBSubscriptionSplit>
createReader(SourceReaderContext readerContext) {
+ return new IoTDBSubscriptionSourceReader<>(readerContext, options,
deserializer);
+ }
+
+ @Override
+ public SplitEnumerator<IoTDBSubscriptionSplit,
IoTDBSubscriptionEnumeratorState> createEnumerator(
+ SplitEnumeratorContext<IoTDBSubscriptionSplit> enumContext) {
+ return new IoTDBSubscriptionEnumerator(enumContext, options);
+ }
+
+ @Override
+ public SplitEnumerator<IoTDBSubscriptionSplit,
IoTDBSubscriptionEnumeratorState>
+ restoreEnumerator(
+ SplitEnumeratorContext<IoTDBSubscriptionSplit> enumContext,
+ IoTDBSubscriptionEnumeratorState checkpoint) {
+ return new IoTDBSubscriptionEnumerator(enumContext, options, checkpoint);
+ }
+
+ @Override
+ public SimpleVersionedSerializer<IoTDBSubscriptionSplit>
getSplitSerializer() {
+ return new IoTDBSubscriptionSplitSerializer();
+ }
+
+ @Override
+ public SimpleVersionedSerializer<IoTDBSubscriptionEnumeratorState>
+ getEnumeratorCheckpointSerializer() {
+ return new IoTDBSubscriptionEnumeratorStateSerializer();
+ }
+}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBSubscriptionEnumerator.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBSubscriptionEnumerator.java
new file mode 100644
index 0000000..b248ab7
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBSubscriptionEnumerator.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.relational.flink.source.cdc;
+
+import org.apache.iotdb.relational.flink.cfg.IoTDBOptions;
+import
org.apache.iotdb.relational.flink.source.cdc.enumerator.IoTDBSubscriptionEnumeratorState;
+import
org.apache.iotdb.relational.flink.source.cdc.split.IoTDBSubscriptionSplit;
+
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Split enumerator for the unbounded CDC source. It owns a single
subscription split and never
+ * signals "no more splits", so the source keeps running.
+ */
+public class IoTDBSubscriptionEnumerator
+ implements SplitEnumerator<IoTDBSubscriptionSplit,
IoTDBSubscriptionEnumeratorState> {
+
+ private final SplitEnumeratorContext<IoTDBSubscriptionSplit> context;
+ private final IoTDBSubscriptionSplit split;
+ private final Deque<IoTDBSubscriptionSplit> pendingSplits = new
ArrayDeque<>();
+ private final Deque<Integer> readersAwaitingSplit = new ArrayDeque<>();
+ private final Set<Integer> assignedReaders = new HashSet<>();
+
+ private boolean allSplitsCreated;
+ private boolean closed;
+
+ public IoTDBSubscriptionEnumerator(
+ SplitEnumeratorContext<IoTDBSubscriptionSplit> context, IoTDBOptions
options) {
+ this(context, options, null);
+ }
+
+ public IoTDBSubscriptionEnumerator(
+ SplitEnumeratorContext<IoTDBSubscriptionSplit> context,
+ IoTDBOptions options,
+ @Nullable IoTDBSubscriptionEnumeratorState checkpoint) {
+ this.context = context;
+ this.split = new IoTDBSubscriptionSplit(options.getCdcTopic(),
options.getCdcConsumerGroup());
+ if (checkpoint != null) {
+ pendingSplits.addAll(checkpoint.getRemainingSplits());
+ allSplitsCreated = checkpoint.isAllSplitsCreated();
+ }
+ }
+
+ @Override
+ public void start() {
+ if (!allSplitsCreated) {
+ pendingSplits.add(split);
+ allSplitsCreated = true;
+ }
+ assignPendingSplits();
+ }
+
+ @Override
+ public void handleSplitRequest(int subtaskId, @Nullable String
requesterHostname) {
+ if (closed) {
+ return;
+ }
+ assignedReaders.remove(subtaskId);
+ readersAwaitingSplit.addLast(subtaskId);
+ assignPendingSplits();
+ }
+
+ @Override
+ public void addSplitsBack(List<IoTDBSubscriptionSplit> splits, int
subtaskId) {
+ assignedReaders.remove(subtaskId);
+ for (int i = splits.size() - 1; i >= 0; i--) {
+ pendingSplits.addFirst(splits.get(i));
+ }
+ assignPendingSplits();
+ }
+
+ @Override
+ public void addReader(int subtaskId) {
+ // Splits are assigned when the reader explicitly requests one.
+ }
+
+ @Override
+ public IoTDBSubscriptionEnumeratorState snapshotState(long checkpointId) {
+ return new IoTDBSubscriptionEnumeratorState(allSplitsCreated, new
ArrayList<>(pendingSplits));
+ }
+
+ @Override
+ public void close() throws IOException {
+ closed = true;
+ pendingSplits.clear();
+ readersAwaitingSplit.clear();
+ assignedReaders.clear();
+ }
+
+ private void assignPendingSplits() {
+ while (!pendingSplits.isEmpty() && !readersAwaitingSplit.isEmpty()) {
+ int subtaskId = readersAwaitingSplit.pollFirst();
+ IoTDBSubscriptionSplit nextSplit = pendingSplits.pollFirst();
+ assignedReaders.add(subtaskId);
+ context.assignSplit(nextSplit, subtaskId);
+ }
+ // Never signal "no more splits": the CDC source is unbounded.
+ }
+}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBSubscriptionSourceReader.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBSubscriptionSourceReader.java
new file mode 100644
index 0000000..a668bc0
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBSubscriptionSourceReader.java
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.relational.flink.source.cdc;
+
+import org.apache.iotdb.relational.flink.cfg.IoTDBOptions;
+import
org.apache.iotdb.relational.flink.source.cdc.client.IoTDBSubscriptionClient;
+import
org.apache.iotdb.relational.flink.source.cdc.split.IoTDBSubscriptionSplit;
+import
org.apache.iotdb.relational.flink.source.common.IoTDBDeserializationSchema;
+import
org.apache.iotdb.session.subscription.consumer.ISubscriptionTablePullConsumer;
+import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
+import org.apache.iotdb.session.subscription.payload.SubscriptionMessageType;
+import
org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler.SubscriptionResultSet;
+
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.tsfile.read.query.dataset.ResultSet;
+
+import java.time.Duration;
+import java.util.ArrayDeque;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Source reader of the unbounded CDC source. It polls the IoTDB subscription
and emits the received
+ * row-level records. The reader never reaches end of input on its own.
+ *
+ * @param <OUT> source output type
+ */
+public class IoTDBSubscriptionSourceReader<OUT>
+ implements SourceReader<OUT, IoTDBSubscriptionSplit> {
+
+ private final SourceReaderContext context;
+ private final IoTDBOptions options;
+ private final IoTDBDeserializationSchema<OUT> deserializer;
+ private final Deque<IoTDBSubscriptionSplit> pendingSplits = new
ArrayDeque<>();
+ private final Deque<OUT> buffer = new ArrayDeque<>();
+
+ private IoTDBSubscriptionSplit currentSplit;
+ private ISubscriptionTablePullConsumer consumer;
+ private boolean noMoreSplits;
+ private boolean closed;
+ private CompletableFuture<Void> availability =
CompletableFuture.completedFuture(null);
+
+ public IoTDBSubscriptionSourceReader(
+ SourceReaderContext context,
+ IoTDBOptions options,
+ IoTDBDeserializationSchema<OUT> deserializer) {
+ this.context = context;
+ this.options = options;
+ this.deserializer = deserializer;
+ }
+
+ @Override
+ public void start() {
+ markUnavailable();
+ context.sendSplitRequest();
+ }
+
+ @Override
+ public InputStatus pollNext(ReaderOutput<OUT> output) throws Exception {
+ if (closed) {
+ return InputStatus.END_OF_INPUT;
+ }
+
+ if (buffer.isEmpty()) {
+ if (currentSplit == null) {
+ currentSplit = pendingSplits.pollFirst();
+ if (currentSplit == null) {
+ if (noMoreSplits) {
+ return InputStatus.END_OF_INPUT;
+ }
+ markUnavailable();
+ context.sendSplitRequest();
+ return InputStatus.NOTHING_AVAILABLE;
+ }
+ openConsumer();
+ }
+ fetchBatch();
+ if (buffer.isEmpty()) {
+ markUnavailable();
+ return InputStatus.NOTHING_AVAILABLE;
+ }
+ }
+
+ output.collect(buffer.pollFirst());
+ return InputStatus.MORE_AVAILABLE;
+ }
+
+ @Override
+ public List<IoTDBSubscriptionSplit> snapshotState(long checkpointId) {
+ if (currentSplit == null) {
+ return Collections.emptyList();
+ }
+ return Collections.singletonList(currentSplit);
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) {
+ // Offsets are committed by the consumer when auto-commit is enabled.
+ }
+
+ @Override
+ public CompletableFuture<Void> isAvailable() {
+ return availability;
+ }
+
+ @Override
+ public void addSplits(List<IoTDBSubscriptionSplit> splits) {
+ if (closed || splits == null || splits.isEmpty()) {
+ return;
+ }
+ pendingSplits.addAll(splits);
+ markAvailable();
+ }
+
+ @Override
+ public void notifyNoMoreSplits() {
+ noMoreSplits = true;
+ markAvailable();
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ if (consumer != null) {
+ consumer.close();
+ consumer = null;
+ }
+ currentSplit = null;
+ pendingSplits.clear();
+ buffer.clear();
+ markAvailable();
+ }
+
+ private void openConsumer() throws Exception {
+ IoTDBSubscriptionClient.createTopicIfNotExists(options);
+ consumer = IoTDBSubscriptionClient.createPullConsumer(options);
+ consumer.open();
+ consumer.subscribe(currentSplit.getTopic());
+ }
+
+ private void fetchBatch() throws Exception {
+ List<SubscriptionMessage> messages =
+ consumer.poll(Duration.ofMillis(options.getCdcPollTimeoutMs()));
+ for (SubscriptionMessage message : messages) {
+ if (message.getMessageType() !=
SubscriptionMessageType.RECORD_HANDLER.getType()) {
+ continue;
+ }
+ for (ResultSet resultSet : message.getResultSets()) {
+ SubscriptionDataIterator iterator =
+ new SubscriptionDataIterator((SubscriptionResultSet) resultSet);
+ while (iterator.next()) {
+ OUT record = deserializer.deserialize(iterator);
+ if (record != null) {
+ buffer.add(record);
+ }
+ }
+ }
+ }
+ }
+
+ private synchronized void markUnavailable() {
+ if (!availability.isDone()) {
+ return;
+ }
+ availability = new CompletableFuture<>();
+ }
+
+ private synchronized void markAvailable() {
+ CompletableFuture<Void> current = availability;
+ if (!current.isDone()) {
+ current.complete(null);
+ }
+ availability = CompletableFuture.completedFuture(null);
+ }
+}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSource.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/scan/IoTDBSource.java
similarity index 86%
rename from
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSource.java
rename to
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/scan/IoTDBSource.java
index 94754e4..ab35114 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSource.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/scan/IoTDBSource.java
@@ -17,15 +17,15 @@
* under the License.
*/
-package org.apache.iotdb.relational.flink.source;
+package org.apache.iotdb.relational.flink.source.scan;
import org.apache.iotdb.relational.flink.cfg.IoTDBOptions;
-import
org.apache.iotdb.relational.flink.source.deserializer.IoTDBDeserializationSchema;
-import
org.apache.iotdb.relational.flink.source.enumerator.IoTDBSourceEnumeratorState;
-import
org.apache.iotdb.relational.flink.source.enumerator.IoTDBSourceEnumeratorStateSerializer;
-import org.apache.iotdb.relational.flink.source.pushdown.AggregateSpec;
-import org.apache.iotdb.relational.flink.source.split.IoTDBSourceSplit;
-import
org.apache.iotdb.relational.flink.source.split.IoTDBSourceSplitSerializer;
+import
org.apache.iotdb.relational.flink.source.common.IoTDBDeserializationSchema;
+import
org.apache.iotdb.relational.flink.source.scan.enumerator.IoTDBSourceEnumeratorState;
+import
org.apache.iotdb.relational.flink.source.scan.enumerator.IoTDBSourceEnumeratorStateSerializer;
+import org.apache.iotdb.relational.flink.source.scan.pushdown.AggregateSpec;
+import org.apache.iotdb.relational.flink.source.scan.split.IoTDBSourceSplit;
+import
org.apache.iotdb.relational.flink.source.scan.split.IoTDBSourceSplitSerializer;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.Source;
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSourceEnumerator.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/scan/IoTDBSourceEnumerator.java
similarity index 94%
rename from
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSourceEnumerator.java
rename to
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/scan/IoTDBSourceEnumerator.java
index 99bf3fb..5f88480 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSourceEnumerator.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/scan/IoTDBSourceEnumerator.java
@@ -17,12 +17,12 @@
* under the License.
*/
-package org.apache.iotdb.relational.flink.source;
+package org.apache.iotdb.relational.flink.source.scan;
import org.apache.iotdb.relational.flink.cfg.IoTDBOptions;
-import
org.apache.iotdb.relational.flink.source.enumerator.IoTDBSourceEnumeratorState;
-import org.apache.iotdb.relational.flink.source.pushdown.AggregateSpec;
-import org.apache.iotdb.relational.flink.source.split.IoTDBSourceSplit;
+import
org.apache.iotdb.relational.flink.source.scan.enumerator.IoTDBSourceEnumeratorState;
+import org.apache.iotdb.relational.flink.source.scan.pushdown.AggregateSpec;
+import org.apache.iotdb.relational.flink.source.scan.split.IoTDBSourceSplit;
import org.apache.iotdb.relational.flink.utils.IoTDBUtils;
import org.apache.flink.api.connector.source.SplitEnumerator;
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSourceReader.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/scan/IoTDBSourceReader.java
similarity index 93%
rename from
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSourceReader.java
rename to
connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/scan/IoTDBSourceReader.java
index 8415756..34566e5 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSourceReader.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/scan/IoTDBSourceReader.java
@@ -17,13 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.relational.flink.source;
+package org.apache.iotdb.relational.flink.source.scan;
import org.apache.iotdb.isession.ITableSession;
import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.relational.flink.cfg.IoTDBOptions;
-import
org.apache.iotdb.relational.flink.source.deserializer.IoTDBDeserializationSchema;
-import org.apache.iotdb.relational.flink.source.split.IoTDBSourceSplit;
+import org.apache.iotdb.relational.flink.source.common.IoTDBDataIterator;
+import
org.apache.iotdb.relational.flink.source.common.IoTDBDeserializationSchema;
+import org.apache.iotdb.relational.flink.source.scan.split.IoTDBSourceSplit;
import org.apache.iotdb.session.TableSessionBuilder;
import org.apache.flink.api.connector.source.ReaderOutput;
@@ -50,7 +51,7 @@ public class IoTDBSourceReader<OUT> implements
SourceReader<OUT, IoTDBSourceSpli
private IoTDBSourceSplit currentSplit;
private ITableSession session;
private SessionDataSet dataSet;
- private SessionDataSet.DataIterator iterator;
+ private IoTDBDataIterator iterator;
private boolean noMoreSplits;
private boolean closed;
private CompletableFuture<Void> availability =
CompletableFuture.completedFuture(null);
@@ -161,7 +162,7 @@ public class IoTDBSourceReader<OUT> implements
SourceReader<OUT, IoTDBSourceSpli
try {
session = builder.build();
dataSet = session.executeQueryStatement(currentSplit.getSql());
- iterator = dataSet.iterator();
+ iterator = new SessionScanDataIterator(dataSet.iterator());
return true;
} catch (Exception e) {
try {
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableFactory.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableFactory.java
index e755082..22cd95d 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableFactory.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableFactory.java
@@ -25,6 +25,7 @@ import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.lookup.LookupOptions;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.factories.FactoryUtil;
@@ -39,8 +40,8 @@ import java.util.Set;
/**
* SPI factory of the IoTDB relational (table model) Flink connector.
*
- * <p>The factory only wires options and the dynamic table source/sink.
Runtime read/write logic is
- * intentionally not implemented yet.
+ * <p>The factory resolves the table options and creates the dynamic table
source/sink. The source
+ * supports bounded scan, lookup and CDC reads, and the sink writes insert
rows into an IoTDB table.
*/
public class IoTDBRelationalDynamicTableFactory
implements DynamicTableSourceFactory, DynamicTableSinkFactory {
@@ -49,8 +50,9 @@ public class IoTDBRelationalDynamicTableFactory
public DynamicTableSource createDynamicTableSource(Context context) {
FactoryUtil.TableFactoryHelper helper =
FactoryUtil.createTableFactoryHelper(this, context);
helper.validate();
+ ReadableConfig config = helper.getOptions();
return new IoTDBRelationalDynamicTableSource(
- toOptions(helper.getOptions()),
context.getCatalogTable().getResolvedSchema());
+ toOptions(config), context.getCatalogTable().getResolvedSchema(),
config);
}
@Override
@@ -82,7 +84,19 @@ public class IoTDBRelationalDynamicTableFactory
IoTDBOptions.TAG_COLUMNS,
IoTDBOptions.ATTRIBUTE_COLUMNS,
IoTDBOptions.LOOKUP_ASYNC,
- IoTDBOptions.LOOKUP_THREAD_SIZE));
+ IoTDBOptions.LOOKUP_THREAD_SIZE,
+ IoTDBOptions.SCAN_MODE,
+ IoTDBOptions.CDC_TOPIC,
+ IoTDBOptions.CDC_CONSUMER_GROUP,
+ IoTDBOptions.CDC_MODE,
+ IoTDBOptions.CDC_START_TIME,
+ IoTDBOptions.CDC_POLL_TIMEOUT_MS,
+ IoTDBOptions.CDC_AUTO_COMMIT,
+ LookupOptions.CACHE_TYPE,
+ LookupOptions.PARTIAL_CACHE_MAX_ROWS,
+ LookupOptions.PARTIAL_CACHE_EXPIRE_AFTER_WRITE,
+ LookupOptions.PARTIAL_CACHE_EXPIRE_AFTER_ACCESS,
+ LookupOptions.PARTIAL_CACHE_CACHE_MISSING_KEY));
}
private static IoTDBOptions toOptions(ReadableConfig config) {
@@ -97,6 +111,13 @@ public class IoTDBRelationalDynamicTableFactory
.withAttributeColumns(parseColumnNames(config.get(IoTDBOptions.ATTRIBUTE_COLUMNS)))
.withLookupAsync(config.get(IoTDBOptions.LOOKUP_ASYNC))
.withLookupThreadSize(config.get(IoTDBOptions.LOOKUP_THREAD_SIZE))
+ .withScanMode(config.get(IoTDBOptions.SCAN_MODE))
+ .withCdcTopic(config.get(IoTDBOptions.CDC_TOPIC))
+ .withCdcConsumerGroup(config.get(IoTDBOptions.CDC_CONSUMER_GROUP))
+ .withCdcMode(config.get(IoTDBOptions.CDC_MODE))
+ .withCdcStartTime(config.get(IoTDBOptions.CDC_START_TIME))
+ .withCdcPollTimeoutMs(config.get(IoTDBOptions.CDC_POLL_TIMEOUT_MS))
+ .withCdcAutoCommit(config.get(IoTDBOptions.CDC_AUTO_COMMIT))
.build();
}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSource.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSource.java
index 7adcb7e..64bbd76 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSource.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSource.java
@@ -20,15 +20,17 @@
package org.apache.iotdb.relational.flink.table;
import org.apache.iotdb.relational.flink.cfg.IoTDBOptions;
-import org.apache.iotdb.relational.flink.source.IoTDBSource;
-import
org.apache.iotdb.relational.flink.source.deserializer.RowDataDeserializationSchema;
+import org.apache.iotdb.relational.flink.source.cdc.IoTDBCDCSource;
+import
org.apache.iotdb.relational.flink.source.common.RowDataDeserializationSchema;
import
org.apache.iotdb.relational.flink.source.lookup.IoTDBAsyncLookupFunction;
import org.apache.iotdb.relational.flink.source.lookup.IoTDBLookupFunction;
-import org.apache.iotdb.relational.flink.source.pushdown.AggregateSpec;
-import
org.apache.iotdb.relational.flink.source.pushdown.IoTDBAggregatePushDownUtils;
-import
org.apache.iotdb.relational.flink.source.pushdown.IoTDBExpressionVisitor;
+import org.apache.iotdb.relational.flink.source.scan.IoTDBSource;
+import org.apache.iotdb.relational.flink.source.scan.pushdown.AggregateSpec;
+import
org.apache.iotdb.relational.flink.source.scan.pushdown.IoTDBAggregatePushDownUtils;
+import
org.apache.iotdb.relational.flink.source.scan.pushdown.IoTDBExpressionVisitor;
import org.apache.iotdb.relational.flink.utils.IoTDBUtils;
+import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.connector.ChangelogMode;
@@ -42,6 +44,11 @@ import
org.apache.flink.table.connector.source.abilities.SupportsLimitPushDown;
import
org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown;
import
org.apache.flink.table.connector.source.lookup.AsyncLookupFunctionProvider;
import org.apache.flink.table.connector.source.lookup.LookupFunctionProvider;
+import org.apache.flink.table.connector.source.lookup.LookupOptions;
+import
org.apache.flink.table.connector.source.lookup.PartialCachingAsyncLookupProvider;
+import
org.apache.flink.table.connector.source.lookup.PartialCachingLookupProvider;
+import org.apache.flink.table.connector.source.lookup.cache.DefaultLookupCache;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.expressions.AggregateExpression;
import org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.types.DataType;
@@ -53,8 +60,16 @@ import java.util.List;
/**
* Dynamic table source of the IoTDB relational (table model) Flink connector.
*
- * <p>Scan reads are fully implemented. Lookup reads are declared through
{@link LookupTableSource}
- * but their runtime behavior is still a stub.
+ * <p>It supports three read paths:
+ *
+ * <ul>
+ * <li>bounded scan reads ({@code scan.mode=snapshot}, the default) with
filter, projection, limit
+ * and aggregate pushdown;
+ * <li>lookup reads, either synchronous or asynchronous ({@code
lookup.async});
+ * <li>unbounded CDC reads backed by the IoTDB subscription API ({@code
scan.mode=cdc}).
+ * </ul>
+ *
+ * <p>All rows are emitted as inserts, so the changelog mode is insert-only.
*/
public class IoTDBRelationalDynamicTableSource
implements ScanTableSource,
@@ -66,14 +81,17 @@ public class IoTDBRelationalDynamicTableSource
private final IoTDBOptions options;
private final ResolvedSchema schema;
+ private final ReadableConfig config;
private DataType physicalRowDataType;
private final List<String> resolvedFilterQueries = new ArrayList<>();
private long limit = -1L;
private AggregateSpec aggregateSpec;
- public IoTDBRelationalDynamicTableSource(IoTDBOptions options,
ResolvedSchema schema) {
+ public IoTDBRelationalDynamicTableSource(
+ IoTDBOptions options, ResolvedSchema schema, ReadableConfig config) {
this.options = options;
this.schema = schema;
+ this.config = config;
this.physicalRowDataType = schema.toPhysicalRowDataType();
}
@@ -84,6 +102,11 @@ public class IoTDBRelationalDynamicTableSource
@Override
public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
+ if (options.isCdc()) {
+ return SourceProvider.of(
+ new IoTDBCDCSource<RowData>(
+ options, new RowDataDeserializationSchema(physicalRowDataType)));
+ }
return SourceProvider.of(
new IoTDBSource<>(
options,
@@ -112,13 +135,24 @@ public class IoTDBRelationalDynamicTableSource
}
keyIndices[i] = keyIndex;
}
+ LookupOptions.LookupCacheType cacheType =
config.get(LookupOptions.CACHE_TYPE);
+ if (cacheType == LookupOptions.LookupCacheType.FULL) {
+ throw new TableException(
+ "The IoTDB connector does not support the lookup FULL cache; use
'partial' or 'none'.");
+ }
+ boolean cacheEnabled = cacheType == LookupOptions.LookupCacheType.PARTIAL;
if (options.isLookupAsync()) {
- return AsyncLookupFunctionProvider.of(
+ IoTDBAsyncLookupFunction function =
new IoTDBAsyncLookupFunction(
- options, lookupRowDataType, keyIndices,
options.getLookupThreadSize()));
+ options, lookupRowDataType, keyIndices,
options.getLookupThreadSize());
+ return cacheEnabled
+ ? PartialCachingAsyncLookupProvider.of(function,
DefaultLookupCache.fromConfig(config))
+ : AsyncLookupFunctionProvider.of(function);
}
- return LookupFunctionProvider.of(
- new IoTDBLookupFunction(options, lookupRowDataType, keyIndices));
+ IoTDBLookupFunction function = new IoTDBLookupFunction(options,
lookupRowDataType, keyIndices);
+ return cacheEnabled
+ ? PartialCachingLookupProvider.of(function,
DefaultLookupCache.fromConfig(config))
+ : LookupFunctionProvider.of(function);
}
@Override
@@ -181,7 +215,8 @@ public class IoTDBRelationalDynamicTableSource
@Override
public DynamicTableSource copy() {
- IoTDBRelationalDynamicTableSource copy = new
IoTDBRelationalDynamicTableSource(options, schema);
+ IoTDBRelationalDynamicTableSource copy =
+ new IoTDBRelationalDynamicTableSource(options, schema, config);
copy.physicalRowDataType = physicalRowDataType;
copy.resolvedFilterQueries.addAll(resolvedFilterQueries);
copy.limit = limit;
diff --git a/pom.xml b/pom.xml
index df51590..707e985 100644
--- a/pom.xml
+++ b/pom.xml
@@ -91,7 +91,7 @@
https://github.com/apache/iotdb-bin-resources/tree/main/iotdb-tools-thrift
-->
<iotdb-tools-thrift.version>0.14.1.0</iotdb-tools-thrift.version>
- <iotdb.version>2.0.5</iotdb.version>
+ <iotdb.version>2.0.7-SNAPSHOT</iotdb.version>
<jackson.version>2.16.2</jackson.version>
<!-- This is the last version to support the javax namespace -->
<jakarta.servlet-api.version>4.0.4</jakarta.servlet-api.version>
@@ -980,6 +980,11 @@
<artifactId>iotdb-session</artifactId>
<version>${iotdb.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-subscription</artifactId>
+ <version>${iotdb.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
