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 991ac3d feat: support configurable timestamp precision
991ac3d is described below
commit 991ac3d7e1ec29f193b257419d4ec7713e38ca3c
Author: shuwenwei <[email protected]>
AuthorDate: Thu Oct 8 11:36:02 2026 +0800
feat: support configurable timestamp precision
Add iotdb.timestamp-precision (ms/us/ns) and convert timestamps on write,
CDC read, catalog mapping and pushed literals; add iotdb.cdc.end-time with
start/end-time validation; drop iotdb.cdc.topic and iotdb.cdc.consumer-group in
favor of derived names.
---
.../iotdb/relational/flink/cfg/IoTDBOptions.java | 74 +++++++------
.../flink/sink/RowDataSinkDataConverter.java | 38 +++++--
.../relational/flink/sink/SinkDataConverter.java | 8 +-
.../flink/source/cdc/SubscriptionDataIterator.java | 9 +-
.../source/cdc/client/IoTDBSubscriptionClient.java | 58 +++++++++-
.../flink/source/lookup/IoTDBLookupReader.java | 4 +-
.../scan/pushdown/IoTDBExpressionVisitor.java | 9 +-
.../iotdb/relational/flink/utils/IoTDBUtils.java | 50 +++++----
.../relational/flink/utils/TimestampPrecision.java | 120 +++++++++++++++++++++
.../flink/sink/RowDataSinkDataConverterTest.java | 15 ++-
.../flink/source/lookup/IoTDBLookupReaderTest.java | 2 +-
.../relational/flink/catalog/IoTDBCatalog.java | 3 +-
.../flink/catalog/IoTDBCatalogFactory.java | 4 +-
.../relational/flink/sink/IoTDBSinkWriter.java | 11 +-
.../source/cdc/IoTDBSubscriptionSourceReader.java | 5 +-
.../table/IoTDBRelationalDynamicTableFactory.java | 8 +-
.../table/IoTDBRelationalDynamicTableSink.java | 3 +-
.../table/IoTDBRelationalDynamicTableSource.java | 9 +-
18 files changed, 345 insertions(+), 85 deletions(-)
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 1463679..75ae43c 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
@@ -19,6 +19,8 @@
package org.apache.iotdb.relational.flink.cfg;
+import org.apache.iotdb.relational.flink.utils.TimestampPrecision;
+
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
@@ -76,24 +78,26 @@ public class IoTDBOptions implements Serializable {
.enumType(ScanMode.class)
.defaultValue(ScanMode.SNAPSHOT);
- public static final ConfigOption<String> CDC_TOPIC =
- ConfigOptions.key("iotdb.cdc.topic").stringType().noDefaultValue();
-
- public static final ConfigOption<String> CDC_CONSUMER_GROUP =
-
ConfigOptions.key("iotdb.cdc.consumer-group").stringType().noDefaultValue();
-
public static final ConfigOption<String> CDC_MODE =
ConfigOptions.key("iotdb.cdc.mode").stringType().defaultValue("live");
public static final ConfigOption<String> CDC_START_TIME =
ConfigOptions.key("iotdb.cdc.start-time").stringType().noDefaultValue();
+ public static final ConfigOption<String> CDC_END_TIME =
+ ConfigOptions.key("iotdb.cdc.end-time").stringType().noDefaultValue();
+
public static final ConfigOption<Long> CDC_POLL_TIMEOUT_MS =
ConfigOptions.key("iotdb.cdc.poll-timeout-ms").longType().defaultValue(1000L);
public static final ConfigOption<Boolean> CDC_AUTO_COMMIT =
ConfigOptions.key("iotdb.cdc.auto-commit").booleanType().defaultValue(true);
+ public static final ConfigOption<TimestampPrecision> TIMESTAMP_PRECISION =
+ ConfigOptions.key("iotdb.timestamp-precision")
+ .enumType(TimestampPrecision.class)
+ .defaultValue(TimestampPrecision.MS);
+
private static final String CDC_TOPIC_PREFIX = "flink_iotdb_table_";
/** Scan mode of the table source. */
@@ -114,12 +118,12 @@ public class IoTDBOptions implements Serializable {
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 String cdcEndTime;
private final long cdcPollTimeoutMs;
private final boolean cdcAutoCommit;
+ private final TimestampPrecision timestampPrecision;
private IoTDBOptions(Builder builder) {
this.nodeUrls = builder.nodeUrls;
@@ -134,12 +138,12 @@ public class IoTDBOptions implements Serializable {
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.cdcEndTime = builder.cdcEndTime;
this.cdcPollTimeoutMs = builder.cdcPollTimeoutMs;
this.cdcAutoCommit = builder.cdcAutoCommit;
+ this.timestampPrecision = builder.timestampPrecision;
}
/**
@@ -205,6 +209,13 @@ public class IoTDBOptions implements Serializable {
return attributeColumns;
}
+ /**
+ * @return the timestamp precision of the target IoTDB server.
+ */
+ public TimestampPrecision getTimestampPrecision() {
+ return timestampPrecision;
+ }
+
/**
* @return whether an asynchronous lookup function is used for lookup joins.
*/
@@ -234,23 +245,17 @@ public class IoTDBOptions implements Serializable {
}
/**
- * @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.
+ * @return the CDC topic name, 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.
+ * @return the CDC consumer group id, derived from the topic name.
*/
public String getCdcConsumerGroup() {
- if (cdcConsumerGroup != null && !cdcConsumerGroup.isEmpty()) {
- return cdcConsumerGroup;
- }
return getCdcTopic() + "_group";
}
@@ -268,6 +273,13 @@ public class IoTDBOptions implements Serializable {
return cdcStartTime;
}
+ /**
+ * @return the CDC history end time, or {@code null} when unset.
+ */
+ public String getCdcEndTime() {
+ return cdcEndTime;
+ }
+
/**
* @return the CDC poll timeout in milliseconds.
*/
@@ -308,12 +320,12 @@ public class IoTDBOptions implements Serializable {
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 String cdcEndTime;
private long cdcPollTimeoutMs = 1000L;
private boolean cdcAutoCommit = true;
+ private TimestampPrecision timestampPrecision = TimestampPrecision.MS;
public Builder withNodeUrls(List<String> nodeUrls) {
this.nodeUrls = nodeUrls;
@@ -375,16 +387,6 @@ public class IoTDBOptions implements Serializable {
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;
@@ -395,6 +397,11 @@ public class IoTDBOptions implements Serializable {
return this;
}
+ public Builder withCdcEndTime(String cdcEndTime) {
+ this.cdcEndTime = cdcEndTime;
+ return this;
+ }
+
public Builder withCdcPollTimeoutMs(long cdcPollTimeoutMs) {
this.cdcPollTimeoutMs = cdcPollTimeoutMs;
return this;
@@ -405,6 +412,11 @@ public class IoTDBOptions implements Serializable {
return this;
}
+ public Builder withTimestampPrecision(TimestampPrecision
timestampPrecision) {
+ this.timestampPrecision = timestampPrecision;
+ 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/sink/RowDataSinkDataConverter.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverter.java
index 8e378a1..976afdd 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverter.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverter.java
@@ -9,22 +9,26 @@
*
* 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.
+ * 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.sink;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.types.RowKind;
import org.apache.tsfile.utils.Binary;
import java.io.IOException;
+import java.sql.Timestamp;
import java.time.LocalDate;
+import java.util.List;
/**
* {@link SinkDataConverter} over Flink {@link RowData}.
@@ -36,17 +40,30 @@ public class RowDataSinkDataConverter implements
SinkDataConverter<RowData> {
private static final long serialVersionUID = 1L;
+ private final int[] timestampPrecisions;
+
+ public RowDataSinkDataConverter(DataType rowDataType) {
+ List<DataType> fieldTypes = DataType.getFieldDataTypes(rowDataType);
+ this.timestampPrecisions = new int[fieldTypes.size()];
+ for (int i = 0; i < fieldTypes.size(); i++) {
+ LogicalType logicalType = fieldTypes.get(i).getLogicalType();
+ timestampPrecisions[i] =
+ logicalType instanceof TimestampType ? ((TimestampType)
logicalType).getPrecision() : -1;
+ }
+ }
+
@Override
public Iterator getIterator(RowData record) throws IOException {
- return new RowDataIterator(record);
+ return new RowDataIterator(record, timestampPrecisions);
}
private static class RowDataIterator implements SinkDataConverter.Iterator {
private final RowData current;
+ private final int[] timestampPrecisions;
private boolean consumed;
- private RowDataIterator(RowData record) throws IOException {
+ private RowDataIterator(RowData record, int[] timestampPrecisions) throws
IOException {
if (record != null && record.getRowKind() != RowKind.INSERT) {
throw new IOException(
"The IoTDB table sink only accepts INSERT records, but got "
@@ -54,6 +71,7 @@ public class RowDataSinkDataConverter implements
SinkDataConverter<RowData> {
+ ".");
}
this.current = record;
+ this.timestampPrecisions = timestampPrecisions;
}
@Override
@@ -111,8 +129,8 @@ public class RowDataSinkDataConverter implements
SinkDataConverter<RowData> {
}
@Override
- public long getTimestamp(int columnIndex) {
- return current.getTimestamp(columnIndex, 3).getMillisecond();
+ public Timestamp getTimestamp(int columnIndex) {
+ return current.getTimestamp(columnIndex,
timestampPrecisions[columnIndex]).toTimestamp();
}
}
}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/SinkDataConverter.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/SinkDataConverter.java
index a7bbbe4..0963588 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/SinkDataConverter.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/SinkDataConverter.java
@@ -23,6 +23,7 @@ import org.apache.tsfile.utils.Binary;
import java.io.IOException;
import java.io.Serializable;
+import java.sql.Timestamp;
import java.time.LocalDate;
/**
@@ -72,8 +73,11 @@ public interface SinkDataConverter<IN> extends Serializable {
LocalDate getDate(int columnIndex) throws IOException;
- /** Returns the TIME value of the current row as epoch milliseconds. */
- long getTimestamp(int columnIndex) throws IOException;
+ /**
+ * Returns the timestamp value of the current row, keeping the
sub-millisecond part. The sink
+ * converts it to the target IoTDB timestamp precision.
+ */
+ Timestamp getTimestamp(int columnIndex) throws IOException;
@Override
default void close() throws IOException {}
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
index bee63a7..3e36c4e 100644
---
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
@@ -20,6 +20,7 @@
package org.apache.iotdb.relational.flink.source.cdc;
import org.apache.iotdb.relational.flink.source.common.IoTDBDataIterator;
+import org.apache.iotdb.relational.flink.utils.TimestampPrecision;
import
org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler.SubscriptionResultSet;
import java.io.IOException;
@@ -44,12 +45,16 @@ public class SubscriptionDataIterator implements
IoTDBDataIterator {
private final SubscriptionResultSet resultSet;
private final List<String> columnNames;
private final int[] sourceIndexes;
+ private final TimestampPrecision timestampPrecision;
public SubscriptionDataIterator(
- SubscriptionResultSet resultSet, List<String> projectedColumns) {
+ SubscriptionResultSet resultSet,
+ List<String> projectedColumns,
+ TimestampPrecision timestampPrecision) {
this.resultSet = resultSet;
this.columnNames = Collections.unmodifiableList(new
ArrayList<>(projectedColumns));
this.sourceIndexes = resolveSourceIndexes(resultSet.getColumnNames(),
projectedColumns);
+ this.timestampPrecision = timestampPrecision;
}
private static int[] resolveSourceIndexes(
@@ -133,6 +138,6 @@ public class SubscriptionDataIterator implements
IoTDBDataIterator {
@Override
public Timestamp getTimestamp(int columnIndex) {
- return new Timestamp(resultSet.getLong(sourceIndexes[columnIndex]));
+ return
timestampPrecision.toTimestamp(resultSet.getLong(sourceIndexes[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
index e76b8ed..6c15dfd 100644
---
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
@@ -26,6 +26,8 @@ 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.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
import java.util.List;
import java.util.Properties;
@@ -47,9 +49,13 @@ public final class IoTDBSubscriptionClient {
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());
- }
+ setTimeProperty(
+ properties,
+ TopicConstant.START_TIME_KEY,
+ "iotdb.cdc.start-time",
+ options.getCdcStartTime());
+ setTimeProperty(
+ properties, TopicConstant.END_TIME_KEY, "iotdb.cdc.end-time",
options.getCdcEndTime());
try (ISubscriptionTableSession session =
new SubscriptionTableSessionBuilder()
@@ -74,6 +80,52 @@ public final class IoTDBSubscriptionClient {
.build();
}
+ private static void setTimeProperty(
+ Properties properties, String key, String optionName, String value) {
+ if (value == null || value.trim().isEmpty()) {
+ return;
+ }
+ properties.setProperty(key, validateTimeValue(value.trim(), optionName));
+ }
+
+ private static String validateTimeValue(String value, String optionName) {
+ if ("now".equalsIgnoreCase(value)) {
+ return value;
+ }
+ try {
+ Long.parseLong(value);
+ return value;
+ } catch (NumberFormatException ignored) {
+ // Not a raw long timestamp; fall through to the datetime formats.
+ }
+ if (isIsoDateTime(value)) {
+ return value;
+ }
+ throw new IllegalArgumentException(
+ "Invalid value for '"
+ + optionName
+ + "': "
+ + value
+ + ". Expected 'now', an ISO datetime (e.g. 2011-12-03T10:15:30),
or a raw long"
+ + " timestamp in the IoTDB timestamp precision.");
+ }
+
+ private static boolean isIsoDateTime(String value) {
+ return parseable(value, DateTimeFormatter.ISO_LOCAL_DATE_TIME)
+ || parseable(value, DateTimeFormatter.ISO_OFFSET_DATE_TIME)
+ || parseable(value, DateTimeFormatter.ISO_LOCAL_DATE)
+ || parseable(value.replace(' ', 'T'),
DateTimeFormatter.ISO_LOCAL_DATE_TIME);
+ }
+
+ private static boolean parseable(String value, DateTimeFormatter formatter) {
+ try {
+ formatter.parse(value);
+ return true;
+ } catch (DateTimeParseException e) {
+ return false;
+ }
+ }
+
private static String[] splitNodeUrl(String nodeUrl) {
String[] parts = nodeUrl.split(":");
if (parts.length != 2) {
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 c92e6c6..44df320 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
@@ -117,7 +117,9 @@ public class IoTDBLookupReader implements AutoCloseable {
List<String> predicates = new ArrayList<>(keyIndices.length);
for (int position = 0; position < keyIndices.length; position++) {
int fieldIndex = keyIndices[position];
- String literal = IoTDBUtils.renderLiteral(keyRow, position,
fieldTypes.get(fieldIndex));
+ String literal =
+ IoTDBUtils.renderLiteral(
+ keyRow, position, fieldTypes.get(fieldIndex),
options.getTimestampPrecision());
if (literal == null) {
return null;
}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/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
index 4884e39..432bb61 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/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
@@ -20,6 +20,7 @@
package org.apache.iotdb.relational.flink.source.scan.pushdown;
import org.apache.iotdb.relational.flink.utils.IoTDBUtils;
+import org.apache.iotdb.relational.flink.utils.TimestampPrecision;
import org.apache.flink.table.expressions.CallExpression;
import org.apache.flink.table.expressions.Expression;
@@ -99,6 +100,12 @@ public class IoTDBExpressionVisitor implements
ExpressionVisitor<String> {
FLINK_TO_IOTDB_FUNCTION_NAMES = Collections.unmodifiableMap(functionNames);
}
+ private final TimestampPrecision timestampPrecision;
+
+ public IoTDBExpressionVisitor(TimestampPrecision timestampPrecision) {
+ this.timestampPrecision = timestampPrecision;
+ }
+
@Override
public String visit(CallExpression call) {
if (call == null || call.getFunctionDefinition().getKind() !=
FunctionKind.SCALAR) {
@@ -114,7 +121,7 @@ public class IoTDBExpressionVisitor implements
ExpressionVisitor<String> {
@Override
public String visit(ValueLiteralExpression valueLiteral) {
- return IoTDBUtils.renderLiteral(valueLiteral);
+ return IoTDBUtils.renderLiteral(valueLiteral, timestampPrecision);
}
@Override
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBUtils.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBUtils.java
index 183b1ee..fbb5ba0 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBUtils.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBUtils.java
@@ -103,7 +103,8 @@ public final class IoTDBUtils {
}
}
- public static DataType toFlinkDataType(TSDataType dataType) {
+ public static DataType toFlinkDataType(
+ TSDataType dataType, TimestampPrecision timestampPrecision) {
switch (dataType) {
case BOOLEAN:
return DataTypes.BOOLEAN();
@@ -123,7 +124,7 @@ public final class IoTDBUtils {
case DATE:
return DataTypes.DATE();
case TIMESTAMP:
- return DataTypes.TIMESTAMP(3);
+ return DataTypes.TIMESTAMP(timestampPrecision.getFlinkPrecision());
default:
throw new CatalogException("Unsupported IoTDB data type: " + dataType);
}
@@ -299,7 +300,8 @@ public final class IoTDBUtils {
* Renders a Flink planner literal as an IoTDB SQL literal. Returns {@code
null} if the literal
* type or value cannot be represented in IoTDB SQL.
*/
- public static String renderLiteral(ValueLiteralExpression literal) {
+ public static String renderLiteral(
+ ValueLiteralExpression literal, TimestampPrecision timestampPrecision) {
if (literal == null || literal.isNull()) {
return null;
}
@@ -340,16 +342,18 @@ public final class IoTDBUtils {
.getValueAs(LocalDate.class)
.map(value -> "CAST('" + value + "' AS DATE)")
.orElse(null);
- case TIMESTAMP_WITHOUT_TIME_ZONE:
- return literal
- .getValueAs(LocalDateTime.class)
- .map(value -> "CAST('" + value + "' AS TIMESTAMP)")
- .orElse(null);
- case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
- return literal
- .getValueAs(Instant.class)
- .map(value -> "CAST('" + value + "' AS TIMESTAMP)")
- .orElse(null);
+ case TIMESTAMP_WITHOUT_TIME_ZONE: {
+ LocalDateTime value =
literal.getValueAs(LocalDateTime.class).orElse(null);
+ return value == null
+ ? null
+ : renderTimestampLiteral(TimestampData.fromLocalDateTime(value),
timestampPrecision);
+ }
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE: {
+ Instant value = literal.getValueAs(Instant.class).orElse(null);
+ return value == null
+ ? null
+ : renderTimestampLiteral(TimestampData.fromInstant(value),
timestampPrecision);
+ }
default:
return null;
}
@@ -362,7 +366,8 @@ public final class IoTDBUtils {
* Renders a runtime {@link RowData} field as an IoTDB SQL literal. Returns
{@code null} when the
* value is {@code null} or cannot be represented in IoTDB SQL.
*/
- public static String renderLiteral(RowData row, int position, DataType
dataType) {
+ public static String renderLiteral(
+ RowData row, int position, DataType dataType, TimestampPrecision
timestampPrecision) {
if (row == null || row.isNullAt(position) || dataType == null) {
return null;
}
@@ -395,11 +400,8 @@ public final class IoTDBUtils {
case DATE:
return "CAST('" + LocalDate.ofEpochDay(row.getInt(position)) + "' AS
DATE)";
case TIMESTAMP_WITHOUT_TIME_ZONE:
- return "CAST('"
- + getTimestamp(row, position, dataType).toLocalDateTime()
- + "' AS TIMESTAMP)";
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
- return "CAST('" + getTimestamp(row, position, dataType).toInstant() +
"' AS TIMESTAMP)";
+ return renderTimestampLiteral(getTimestamp(row, position, dataType),
timestampPrecision);
default:
return null;
}
@@ -417,7 +419,8 @@ public final class IoTDBUtils {
List<int[]> groupingSets,
List<AggregateExpression> aggregateExpressions,
DataType sourceRowDataType,
- DataType producedDataType) {
+ DataType producedDataType,
+ TimestampPrecision timestampPrecision) {
if (groupingSets == null
|| groupingSets.size() != 1
|| aggregateExpressions == null
@@ -450,7 +453,7 @@ public final class IoTDBUtils {
selectExpressions.add(column);
}
- IoTDBExpressionVisitor visitor = new IoTDBExpressionVisitor();
+ IoTDBExpressionVisitor visitor = new
IoTDBExpressionVisitor(timestampPrecision);
for (int i = 0; i < aggregateExpressions.size(); i++) {
int producedIndex = grouping.length + i;
if (producedIndex >= producedFieldTypes.size()) {
@@ -566,6 +569,13 @@ public final class IoTDBUtils {
return row.getTimestamp(position, precision);
}
+ private static String renderTimestampLiteral(
+ TimestampData timestamp, TimestampPrecision timestampPrecision) {
+ long units =
+ timestampPrecision.toUnits(timestamp.getMillisecond(),
timestamp.getNanoOfMillisecond());
+ return "CAST(" + units + " AS TIMESTAMP)";
+ }
+
private static boolean isFinite(float value) {
return !Float.isNaN(value) && !Float.isInfinite(value);
}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/TimestampPrecision.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/TimestampPrecision.java
new file mode 100644
index 0000000..bed585c
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/TimestampPrecision.java
@@ -0,0 +1,120 @@
+/*
+ * 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.utils;
+
+import java.sql.Timestamp;
+import java.util.Locale;
+
+/**
+ * Timestamp precision used by the target IoTDB server.
+ *
+ * <p>The values mirror IoTDB's {@code timestamp_precision}: {@code ms},
{@code us} and {@code ns}.
+ * The connector converts between Flink's epoch-millisecond based timestamps
and the precision of
+ * the IoTDB server with the helpers below.
+ */
+public enum TimestampPrecision {
+ MS("ms", 1L, 3),
+ US("us", 1_000L, 6),
+ NS("ns", 1_000_000L, 9);
+
+ private final String value;
+ private final long timeFactor;
+ private final int flinkPrecision;
+
+ TimestampPrecision(String value, long timeFactor, int flinkPrecision) {
+ this.value = value;
+ this.timeFactor = timeFactor;
+ this.flinkPrecision = flinkPrecision;
+ }
+
+ /** @return the option value, e.g. {@code ms}. */
+ public String getValue() {
+ return value;
+ }
+
+ /** @return the number of units in one millisecond. */
+ public long getTimeFactor() {
+ return timeFactor;
+ }
+
+ /** @return the equivalent Flink {@code TIMESTAMP(p)} precision. */
+ public int getFlinkPrecision() {
+ return flinkPrecision;
+ }
+
+ /**
+ * Converts an epoch-millisecond value (with its sub-millisecond
nanoseconds) into this
+ * precision's units.
+ *
+ * @param epochMillis milliseconds since the epoch
+ * @param nanoOfMilli nanoseconds within the millisecond, in {@code [0,
1_000_000)}
+ * @return the value in this precision's units
+ */
+ public long toUnits(long epochMillis, int nanoOfMilli) {
+ long nanoPerUnit = 1_000_000L / timeFactor;
+ return epochMillis * timeFactor + nanoOfMilli / nanoPerUnit;
+ }
+
+ /**
+ * Converts a value in this precision's units into epoch milliseconds,
truncating the
+ * sub-millisecond part.
+ *
+ * @param raw value in this precision's units
+ * @return milliseconds since the epoch
+ */
+ public long toEpochMillis(long raw) {
+ return Math.floorDiv(raw, timeFactor);
+ }
+
+ /**
+ * Converts a value in this precision's units into a {@link Timestamp},
keeping the
+ * sub-millisecond part.
+ *
+ * @param raw value in this precision's units
+ * @return the corresponding timestamp
+ */
+ public Timestamp toTimestamp(long raw) {
+ long millis = Math.floorDiv(raw, timeFactor);
+ long sub = Math.floorMod(raw, timeFactor);
+ long nanoPerUnit = 1_000_000L / timeFactor;
+ Timestamp timestamp = new Timestamp(millis);
+ timestamp.setNanos((int) (sub * nanoPerUnit));
+ return timestamp;
+ }
+
+ /**
+ * Parses a precision value, e.g. {@code ms}, case-insensitively.
+ *
+ * @param value option value
+ * @return the matching precision
+ * @throws IllegalArgumentException when the value is {@code null} or
unsupported
+ */
+ public static TimestampPrecision fromString(String value) {
+ if (value == null) {
+ throw new IllegalArgumentException("Timestamp precision must not be
null.");
+ }
+ String normalized = value.trim().toLowerCase(Locale.ROOT);
+ for (TimestampPrecision precision : values()) {
+ if (precision.value.equals(normalized)) {
+ return precision;
+ }
+ }
+ throw new IllegalArgumentException("Unsupported timestamp precision: " +
value);
+ }
+}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverterTest.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverterTest.java
index d4e93c6..7ac51b0 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverterTest.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverterTest.java
@@ -19,9 +19,11 @@
package org.apache.iotdb.relational.flink.sink;
+import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.data.GenericRowData;
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.types.RowKind;
import org.apache.tsfile.utils.Binary;
import org.junit.Test;
@@ -38,7 +40,16 @@ import static org.junit.Assert.fail;
public class RowDataSinkDataConverterTest {
- private final RowDataSinkDataConverter converter = new
RowDataSinkDataConverter();
+ private static final DataType ROW_TYPE =
+ DataTypes.ROW(
+ DataTypes.FIELD("time", DataTypes.TIMESTAMP(3)),
+ DataTypes.FIELD("s", DataTypes.STRING()),
+ DataTypes.FIELD("i", DataTypes.INT()),
+ DataTypes.FIELD("d", DataTypes.DOUBLE()),
+ DataTypes.FIELD("date", DataTypes.DATE()),
+ DataTypes.FIELD("b", DataTypes.BYTES()));
+
+ private final RowDataSinkDataConverter converter = new
RowDataSinkDataConverter(ROW_TYPE);
@Test
public void testYieldsOneRowPerInput() throws IOException {
@@ -57,7 +68,7 @@ public class RowDataSinkDataConverterTest {
try (SinkDataConverter.Iterator iterator =
converter.getIterator(rowData())) {
assertTrue(iterator.next());
- assertEquals(1000L, iterator.getTimestamp(0));
+ assertEquals(1000L, iterator.getTimestamp(0).getTime());
assertEquals("t", iterator.getString(1));
assertEquals(7, iterator.getInt(2));
assertEquals(2.5d, iterator.getDouble(3), 0.0d);
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupReaderTest.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupReaderTest.java
index f2b6134..b773791 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupReaderTest.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupReaderTest.java
@@ -61,7 +61,7 @@ public class IoTDBLookupReaderTest {
assertEquals(
"SELECT \"time\", \"device_id\", \"temperature\" FROM \"sensor\" "
- + "WHERE \"device_id\" = 'd1' AND \"time\" =
CAST('2024-01-01T12:30' AS TIMESTAMP)",
+ + "WHERE \"device_id\" = 'd1' AND \"time\" = CAST(1704112200000 AS
TIMESTAMP)",
reader.buildLookupSql(keyRow));
}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalog.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalog.java
index 25f4ebd..a31cd32 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalog.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalog.java
@@ -418,7 +418,8 @@ public class IoTDBCatalog extends AbstractCatalog {
for (int i = 0; i < columns.size(); i++) {
IMeasurementSchema column = columns.get(i);
schemaBuilder.column(
- column.getMeasurementName(),
IoTDBUtils.toFlinkDataType(column.getType()));
+ column.getMeasurementName(),
+ IoTDBUtils.toFlinkDataType(column.getType(),
options.getTimestampPrecision()));
switch (categories.get(i)) {
case TIME:
timeColumn = column.getMeasurementName();
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalogFactory.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalogFactory.java
index b29fde5..553b945 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalogFactory.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalogFactory.java
@@ -50,7 +50,8 @@ public class IoTDBCatalogFactory implements CatalogFactory {
Arrays.asList(
IoTDBOptions.USER,
IoTDBOptions.PASSWORD,
- IoTDBOptions.DEFAULT_DATABASE));
+ IoTDBOptions.DEFAULT_DATABASE,
+ IoTDBOptions.TIMESTAMP_PRECISION));
}
@Override
@@ -68,6 +69,7 @@ public class IoTDBCatalogFactory implements CatalogFactory {
.withUsername(config.get(IoTDBOptions.USER))
.withPassword(config.get(IoTDBOptions.PASSWORD))
.withDefaultDatabase(config.get(IoTDBOptions.DEFAULT_DATABASE))
+ .withTimestampPrecision(config.get(IoTDBOptions.TIMESTAMP_PRECISION))
.build();
}
}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/sink/IoTDBSinkWriter.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/sink/IoTDBSinkWriter.java
index 0422e34..ee8cddd 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/sink/IoTDBSinkWriter.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/sink/IoTDBSinkWriter.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.relational.flink.sink;
import org.apache.iotdb.isession.ITableSession;
import org.apache.iotdb.relational.flink.cfg.IoTDBOptions;
import org.apache.iotdb.relational.flink.utils.IoTDBUtils;
+import org.apache.iotdb.relational.flink.utils.TimestampPrecision;
import org.apache.iotdb.session.TableSessionBuilder;
import org.apache.flink.api.connector.sink2.SinkWriter;
@@ -32,6 +33,7 @@ import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.write.record.Tablet;
import java.io.IOException;
+import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -241,7 +243,7 @@ public class IoTDBSinkWriter<IN> implements SinkWriter<IN> {
}
int row = buffer.getRowSize();
- buffer.addTimestamp(row, iterator.getTimestamp(timeColumnIndex));
+ buffer.addTimestamp(row, toUnits(iterator.getTimestamp(timeColumnIndex)));
for (int i = 0; i < columnNames.size(); i++) {
buffer.addValue(
columnNames.get(i), row, readValue(iterator, columnIndexes[i],
columnTsTypes.get(i)));
@@ -273,12 +275,17 @@ public class IoTDBSinkWriter<IN> implements
SinkWriter<IN> {
case DATE:
return iterator.getDate(columnIndex);
case TIMESTAMP:
- return iterator.getTimestamp(columnIndex);
+ return toUnits(iterator.getTimestamp(columnIndex));
default:
throw new IOException("Unsupported IoTDB data type for sink: " +
dataType);
}
}
+ private long toUnits(Timestamp timestamp) {
+ TimestampPrecision precision = options.getTimestampPrecision();
+ return precision.toUnits(timestamp.getTime(), timestamp.getNanos() %
1_000_000);
+ }
+
private static Set<String> normalizeColumns(List<String> columns) {
if (columns == null) {
return Collections.emptySet();
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
index 878759d..8673275 100644
---
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
@@ -176,7 +176,10 @@ public class IoTDBSubscriptionSourceReader<OUT>
}
for (ResultSet resultSet : message.getResultSets()) {
SubscriptionDataIterator iterator =
- new SubscriptionDataIterator((SubscriptionResultSet) resultSet,
projectedColumns);
+ new SubscriptionDataIterator(
+ (SubscriptionResultSet) resultSet,
+ projectedColumns,
+ options.getTimestampPrecision());
while (iterator.next()) {
OUT record = deserializer.deserialize(iterator);
if (record != null) {
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 22cd95d..7ffa32d 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
@@ -86,12 +86,12 @@ public class IoTDBRelationalDynamicTableFactory
IoTDBOptions.LOOKUP_ASYNC,
IoTDBOptions.LOOKUP_THREAD_SIZE,
IoTDBOptions.SCAN_MODE,
- IoTDBOptions.CDC_TOPIC,
- IoTDBOptions.CDC_CONSUMER_GROUP,
IoTDBOptions.CDC_MODE,
IoTDBOptions.CDC_START_TIME,
+ IoTDBOptions.CDC_END_TIME,
IoTDBOptions.CDC_POLL_TIMEOUT_MS,
IoTDBOptions.CDC_AUTO_COMMIT,
+ IoTDBOptions.TIMESTAMP_PRECISION,
LookupOptions.CACHE_TYPE,
LookupOptions.PARTIAL_CACHE_MAX_ROWS,
LookupOptions.PARTIAL_CACHE_EXPIRE_AFTER_WRITE,
@@ -112,12 +112,12 @@ public class IoTDBRelationalDynamicTableFactory
.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))
+ .withCdcEndTime(config.get(IoTDBOptions.CDC_END_TIME))
.withCdcPollTimeoutMs(config.get(IoTDBOptions.CDC_POLL_TIMEOUT_MS))
.withCdcAutoCommit(config.get(IoTDBOptions.CDC_AUTO_COMMIT))
+ .withTimestampPrecision(config.get(IoTDBOptions.TIMESTAMP_PRECISION))
.build();
}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSink.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSink.java
index eb52dfe..d42994a 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSink.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSink.java
@@ -53,7 +53,8 @@ public class IoTDBRelationalDynamicTableSink implements
DynamicTableSink {
@Override
public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
- SinkDataConverter<RowData> converter = new RowDataSinkDataConverter();
+ SinkDataConverter<RowData> converter =
+ new RowDataSinkDataConverter(schema.toPhysicalRowDataType());
return SinkV2Provider.of(
new IoTDBSink<>(options, schema.toPhysicalRowDataType(), converter));
}
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 b7cbaac..c424c88 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
@@ -179,7 +179,8 @@ public class IoTDBRelationalDynamicTableSource
List<ResolvedExpression> acceptedFilters = new ArrayList<>();
List<ResolvedExpression> remainingFilters = new ArrayList<>();
- IoTDBExpressionVisitor expressionVisitor = new IoTDBExpressionVisitor();
+ IoTDBExpressionVisitor expressionVisitor =
+ new IoTDBExpressionVisitor(options.getTimestampPrecision());
for (ResolvedExpression filter : filters) {
String filterQuery = filter.accept(expressionVisitor);
if (filterQuery == null || filterQuery.trim().isEmpty()) {
@@ -205,7 +206,11 @@ public class IoTDBRelationalDynamicTableSource
// after any projection that has already been pushed into this source.
AggregateSpec spec =
IoTDBUtils.translateAggregate(
- groupingSets, aggregateExpressions, physicalRowDataType,
producedDataType);
+ groupingSets,
+ aggregateExpressions,
+ physicalRowDataType,
+ producedDataType,
+ options.getTimestampPrecision());
if (spec == null) {
return false;
}
