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
commit a5e2d67da5bf72e774c4ff658cebf15d3a201378 Author: shuwenwei <[email protected]> AuthorDate: Fri Sep 18 17:07:14 2026 +0800 sink --- .../flink-iotdb-table-connector-base/pom.xml | 5 + .../flink/catalog/IoTDBCatalogClient.java | 12 +- ...TDBRelationalOptions.java => IoTDBOptions.java} | 10 +- .../flink/sink/RowDataSinkDataConverter.java | 118 ++++++++++ .../relational/flink/sink/SinkDataConverter.java | 81 +++++++ .../sink/serializer/IoTDBTabletSerializer.java | 57 ----- .../serializer/RowDataIoTDBTabletSerializer.java | 64 ----- .../source/pushdown/IoTDBExpressionVisitor.java | 7 +- .../flink/utils/IoTDBIdentifierUtils.java | 39 ---- .../flink/utils/IoTDBRelationalTypeUtils.java | 87 ------- .../relational/flink/utils/IoTDBSQLBuilder.java | 68 ------ .../iotdb/relational/flink/utils/IoTDBUtils.java | 259 +++++++++++++++++++++ .../flink/sink/RowDataSinkDataConverterTest.java | 114 +++++++++ .../relational/flink/catalog/IoTDBCatalog.java | 112 +++------ .../flink/catalog/IoTDBCatalogFactory.java | 26 +-- .../iotdb/relational/flink/sink/IoTDBSink.java | 43 ++-- .../relational/flink/sink/IoTDBSinkWriter.java | 194 ++++++++++++--- .../iotdb/relational/flink/source/IoTDBSource.java | 14 +- .../flink/source/IoTDBSourceEnumerator.java | 12 +- .../relational/flink/source/IoTDBSourceReader.java | 6 +- .../table/IoTDBRelationalDynamicTableFactory.java | 40 ++-- .../table/IoTDBRelationalDynamicTableSink.java | 15 +- .../table/IoTDBRelationalDynamicTableSource.java | 6 +- .../flink/IoTDBRelationalLocalQueryManual.java | 43 +++- 24 files changed, 904 insertions(+), 528 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 984d209..8c3db7b 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 @@ -69,5 +69,10 @@ <artifactId>tsfile</artifactId> <version>${tsfile.version}</version> </dependency> + <dependency> + <groupId>junit</groupId> + <artifactId>junit</artifactId> + <scope>test</scope> + </dependency> </dependencies> </project> diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalogClient.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalogClient.java index c7d3a5a..d34a456 100644 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalogClient.java +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalogClient.java @@ -22,8 +22,8 @@ package org.apache.iotdb.relational.flink.catalog; 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.IoTDBRelationalOptions; -import org.apache.iotdb.relational.flink.utils.IoTDBIdentifierUtils; +import org.apache.iotdb.relational.flink.cfg.IoTDBOptions; +import org.apache.iotdb.relational.flink.utils.IoTDBUtils; import org.apache.iotdb.session.pool.TableSessionPoolBuilder; import org.apache.flink.table.catalog.exceptions.CatalogException; @@ -60,10 +60,10 @@ public class IoTDBCatalogClient implements AutoCloseable { private static final String COLUMN_DATA_TYPE = "DataType"; private static final String COLUMN_CATEGORY = "Category"; - private final IoTDBRelationalOptions options; + private final IoTDBOptions options; private volatile ITableSessionPool sessionPool; - public IoTDBCatalogClient(IoTDBRelationalOptions options) { + public IoTDBCatalogClient(IoTDBOptions options) { this.options = options; } @@ -147,7 +147,7 @@ public class IoTDBCatalogClient implements AutoCloseable { } } - public IoTDBRelationalOptions getOptions() { + public IoTDBOptions getOptions() { return options; } @@ -231,7 +231,7 @@ public class IoTDBCatalogClient implements AutoCloseable { } private static String quoteIdentifier(String identifier) { - return IoTDBIdentifierUtils.quoteIdentifier(identifier); + return IoTDBUtils.quoteIdentifier(identifier); } private ITableSessionPool getSessionPool() { diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/cfg/IoTDBRelationalOptions.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/cfg/IoTDBOptions.java similarity index 95% rename from connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/cfg/IoTDBRelationalOptions.java rename to connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/cfg/IoTDBOptions.java index b056436..fcd21e5 100644 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/cfg/IoTDBRelationalOptions.java +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/cfg/IoTDBOptions.java @@ -32,7 +32,7 @@ import java.util.List; * <p>The static {@link org.apache.flink.configuration.ConfigOption} constants define the keys used * in the Flink table DDL; the instance fields hold the resolved options passed to the runtime. */ -public class IoTDBRelationalOptions implements Serializable { +public class IoTDBOptions implements Serializable { private static final long serialVersionUID = 1L; @@ -75,7 +75,7 @@ public class IoTDBRelationalOptions implements Serializable { private final List<String> tagColumns; private final List<String> attributeColumns; - private IoTDBRelationalOptions(Builder builder) { + private IoTDBOptions(Builder builder) { this.nodeUrls = builder.nodeUrls; this.username = builder.username; this.password = builder.password; @@ -157,7 +157,7 @@ public class IoTDBRelationalOptions implements Serializable { return new Builder(); } - /** Builder of {@link IoTDBRelationalOptions}. */ + /** Builder of {@link IoTDBOptions}. */ public static class Builder { private List<String> nodeUrls; @@ -218,8 +218,8 @@ public class IoTDBRelationalOptions implements Serializable { /** * @return the built options */ - public IoTDBRelationalOptions build() { - return new IoTDBRelationalOptions(this); + public IoTDBOptions build() { + return new IoTDBOptions(this); } } } 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 new file mode 100644 index 0000000..8e378a1 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverter.java @@ -0,0 +1,118 @@ +/* + * 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.sink; + +import org.apache.flink.table.data.RowData; +import org.apache.flink.types.RowKind; +import org.apache.tsfile.utils.Binary; + +import java.io.IOException; +import java.time.LocalDate; + +/** + * {@link SinkDataConverter} over Flink {@link RowData}. + * + * <p>Each input element is a single row, so the returned iterator yields at most one row. Column + * indexes map directly to the RowData field positions. + */ +public class RowDataSinkDataConverter implements SinkDataConverter<RowData> { + + private static final long serialVersionUID = 1L; + + @Override + public Iterator getIterator(RowData record) throws IOException { + return new RowDataIterator(record); + } + + private static class RowDataIterator implements SinkDataConverter.Iterator { + + private final RowData current; + private boolean consumed; + + private RowDataIterator(RowData record) throws IOException { + if (record != null && record.getRowKind() != RowKind.INSERT) { + throw new IOException( + "The IoTDB table sink only accepts INSERT records, but got " + + record.getRowKind() + + "."); + } + this.current = record; + } + + @Override + public boolean next() { + if (current == null || consumed) { + return false; + } + consumed = true; + return true; + } + + @Override + public boolean isNull(int columnIndex) { + return current.isNullAt(columnIndex); + } + + @Override + public boolean getBoolean(int columnIndex) { + return current.getBoolean(columnIndex); + } + + @Override + public int getInt(int columnIndex) { + return current.getInt(columnIndex); + } + + @Override + public long getLong(int columnIndex) { + return current.getLong(columnIndex); + } + + @Override + public float getFloat(int columnIndex) { + return current.getFloat(columnIndex); + } + + @Override + public double getDouble(int columnIndex) { + return current.getDouble(columnIndex); + } + + @Override + public String getString(int columnIndex) { + return current.getString(columnIndex).toString(); + } + + @Override + public Binary getBlob(int columnIndex) { + return new Binary(current.getBinary(columnIndex)); + } + + @Override + public LocalDate getDate(int columnIndex) { + return LocalDate.ofEpochDay(current.getInt(columnIndex)); + } + + @Override + public long getTimestamp(int columnIndex) { + return current.getTimestamp(columnIndex, 3).getMillisecond(); + } + } +} 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 new file mode 100644 index 0000000..a7bbbe4 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/SinkDataConverter.java @@ -0,0 +1,81 @@ +/* + * 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.sink; + +import org.apache.tsfile.utils.Binary; + +import java.io.IOException; +import java.io.Serializable; +import java.time.LocalDate; + +/** + * Converts one sink input element into zero or more IoTDB rows. + * + * <p>Implementations only need to understand the input type {@code IN} and how to expose the + * columns of the IoTDB table; they do not need to know how the rows are buffered or written. + * + * <p>Usage: call {@link #getIterator(Object)} for each input element and consume the returned + * {@link Iterator}; column indexes follow the resolved IoTDB table column order (including the TIME + * column). + * + * @param <IN> input element type + */ +public interface SinkDataConverter<IN> extends Serializable { + + /** Opens the converter before the first input is processed. */ + default void open() throws Exception {} + + /** Creates an iterator that reads the rows produced by the given input element. */ + Iterator getIterator(IN record) throws IOException; + + /** Closes the converter after the last input has been processed. */ + default void close() throws Exception {} + + /** Cursor over the IoTDB rows produced by one input element. */ + interface Iterator extends AutoCloseable { + + /** Advances to the next row; returns {@code false} when there are no more rows. */ + boolean next() throws IOException; + + boolean isNull(int columnIndex); + + 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; + + Binary getBlob(int columnIndex) throws IOException; + + LocalDate getDate(int columnIndex) throws IOException; + + /** Returns the TIME value of the current row as epoch milliseconds. */ + long 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/sink/serializer/IoTDBTabletSerializer.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/serializer/IoTDBTabletSerializer.java deleted file mode 100644 index cbb2e68..0000000 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/serializer/IoTDBTabletSerializer.java +++ /dev/null @@ -1,57 +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.sink.serializer; - -import org.apache.tsfile.write.record.Tablet; - -import java.io.IOException; -import java.io.Serializable; - -/** - * Serializes Flink input records into a writer-owned IoTDB {@link Tablet}. - * - * <p>The sink writer owns tablet creation, batching, and flushing. The serializer only defines how - * an input record is appended to the provided tablet. - * - * @param <IN> input record type - */ -public interface IoTDBTabletSerializer<IN> extends Serializable { - - /** Opens the serializer before the first record is serialized. */ - default void open() throws Exception {} - - /** - * Creates the writer-owned tablet used for batching. - * - * <p>The returned tablet must have a positive maximum row number. The writer owns the returned - * instance and reuses it until it is flushed. - */ - Tablet createTablet(int maxRows) throws IOException; - - /** - * Serializes one input record directly into the writer-owned tablet. - * - * @return {@code true} if the tablet is full after serialization, otherwise {@code false} - */ - boolean serialize(IN record, Tablet tablet) throws IOException; - - /** Closes the serializer after the last record has been processed. */ - default void close() throws Exception {} -} diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/serializer/RowDataIoTDBTabletSerializer.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/serializer/RowDataIoTDBTabletSerializer.java deleted file mode 100644 index 0119cac..0000000 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/serializer/RowDataIoTDBTabletSerializer.java +++ /dev/null @@ -1,64 +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.sink.serializer; - -import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; - -import org.apache.flink.table.catalog.ResolvedSchema; -import org.apache.flink.table.data.RowData; -import org.apache.tsfile.write.record.Tablet; - -import java.io.IOException; - -/** - * Table API serializer that converts {@link RowData} into an IoTDB {@link Tablet}. - * - * <p>TODO: implement the RowData-to-Tablet conversion and keep the runtime state serializable. - */ -public class RowDataIoTDBTabletSerializer implements IoTDBTabletSerializer<RowData> { - - private static final long serialVersionUID = 1L; - - private final IoTDBRelationalOptions options; - private final ResolvedSchema schema; - - public RowDataIoTDBTabletSerializer(IoTDBRelationalOptions options, ResolvedSchema schema) { - this.options = options; - this.schema = schema; - } - - @Override - public Tablet createTablet(int maxRows) throws IOException { - throw new UnsupportedOperationException("Not implemented yet."); - } - - @Override - public boolean serialize(RowData record, Tablet tablet) throws IOException { - throw new UnsupportedOperationException("Not implemented yet."); - } - - public IoTDBRelationalOptions getOptions() { - return options; - } - - public ResolvedSchema getSchema() { - return schema; - } -} 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/pushdown/IoTDBExpressionVisitor.java index b970986..da9c874 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/pushdown/IoTDBExpressionVisitor.java @@ -19,8 +19,7 @@ package org.apache.iotdb.relational.flink.source.pushdown; -import org.apache.iotdb.relational.flink.utils.IoTDBIdentifierUtils; -import org.apache.iotdb.relational.flink.utils.IoTDBRelationalTypeUtils; +import org.apache.iotdb.relational.flink.utils.IoTDBUtils; import org.apache.flink.table.expressions.CallExpression; import org.apache.flink.table.expressions.Expression; @@ -111,7 +110,7 @@ public class IoTDBExpressionVisitor implements ExpressionVisitor<String> { if (fieldReference == null || fieldReference.getInputIndex() != 0) { return null; } - return IoTDBIdentifierUtils.quoteIdentifier(fieldReference.getName()); + return IoTDBUtils.quoteIdentifier(fieldReference.getName()); } @Override @@ -441,7 +440,7 @@ public class IoTDBExpressionVisitor implements ExpressionVisitor<String> { return null; } try { - return IoTDBRelationalTypeUtils.toIoTDBDataType(expression.getOutputDataType()).name(); + return IoTDBUtils.toIoTDBDataType(expression.getOutputDataType()).name(); } catch (RuntimeException e) { return null; } diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBIdentifierUtils.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBIdentifierUtils.java deleted file mode 100644 index a34a437..0000000 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBIdentifierUtils.java +++ /dev/null @@ -1,39 +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.utils; - -/** Utilities for rendering IoTDB table-model identifiers in SQL. */ -public final class IoTDBIdentifierUtils { - - private IoTDBIdentifierUtils() {} - - /** - * Quotes a resolved logical identifier using IoTDB's double-quote syntax. - * - * <p>The input is treated as a logical name, not as SQL text. Backticks are preserved as part of - * the identifier and double quotes are escaped by doubling. - */ - public static String quoteIdentifier(String identifier) { - if (identifier == null) { - throw new IllegalArgumentException("Identifier must not be null."); - } - return "\"" + identifier.replace("\"", "\"\"") + "\""; - } -} diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBRelationalTypeUtils.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBRelationalTypeUtils.java deleted file mode 100644 index 066e3e1..0000000 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBRelationalTypeUtils.java +++ /dev/null @@ -1,87 +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.utils; - -import org.apache.flink.table.api.DataTypes; -import org.apache.flink.table.catalog.exceptions.CatalogException; -import org.apache.flink.table.types.DataType; -import org.apache.tsfile.enums.TSDataType; - -/** Utilities for converting between Flink and IoTDB table-model data types. */ -public final class IoTDBRelationalTypeUtils { - - private IoTDBRelationalTypeUtils() {} - - public static TSDataType toIoTDBDataType(DataType dataType) { - switch (dataType.getLogicalType().getTypeRoot()) { - case BOOLEAN: - return TSDataType.BOOLEAN; - case TINYINT: - case SMALLINT: - case INTEGER: - return TSDataType.INT32; - case BIGINT: - return TSDataType.INT64; - case FLOAT: - return TSDataType.FLOAT; - case DOUBLE: - return TSDataType.DOUBLE; - case CHAR: - case VARCHAR: - return TSDataType.STRING; - case BINARY: - case VARBINARY: - return TSDataType.BLOB; - case DATE: - return TSDataType.DATE; - case TIMESTAMP_WITHOUT_TIME_ZONE: - case TIMESTAMP_WITH_LOCAL_TIME_ZONE: - return TSDataType.TIMESTAMP; - default: - throw new CatalogException("Unsupported Flink data type for IoTDB: " + dataType); - } - } - - public static DataType toFlinkDataType(TSDataType dataType) { - switch (dataType) { - case BOOLEAN: - return DataTypes.BOOLEAN(); - case INT32: - return DataTypes.INT(); - case INT64: - return DataTypes.BIGINT(); - case FLOAT: - return DataTypes.FLOAT(); - case DOUBLE: - return DataTypes.DOUBLE(); - case TEXT: - case STRING: - return DataTypes.STRING(); - case BLOB: - return DataTypes.BYTES(); - case DATE: - return DataTypes.DATE(); - case TIMESTAMP: - return DataTypes.TIMESTAMP(3); - default: - throw new CatalogException("Unsupported IoTDB data type: " + dataType); - } - } -} diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBSQLBuilder.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBSQLBuilder.java deleted file mode 100644 index 12ce03c..0000000 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBSQLBuilder.java +++ /dev/null @@ -1,68 +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.utils; - -import org.apache.flink.table.types.DataType; - -import java.util.List; - -/** Builds reusable IoTDB SQL statements for the relational table connector. */ -public final class IoTDBSQLBuilder { - - private IoTDBSQLBuilder() {} - - /** - * Builds a bounded table-model scan query. - * - * @param table IoTDB table name - * @param rowDataType projected Flink row type whose field names define the SELECT list - * @param filterQueries already-rendered IoTDB predicate fragments - * @param limit maximum number of rows, or a negative value for no limit - * @return IoTDB SELECT SQL - */ - public static String buildSelectQuery( - String table, DataType rowDataType, List<String> filterQueries, long limit) { - List<String> fieldNames = DataType.getFieldNames(rowDataType); - if (fieldNames.isEmpty()) { - throw new IllegalArgumentException("IoTDB source requires at least one selected column."); - } - - StringBuilder columns = new StringBuilder(); - for (String fieldName : fieldNames) { - if (columns.length() > 0) { - columns.append(", "); - } - columns.append(IoTDBIdentifierUtils.quoteIdentifier(fieldName)); - } - - StringBuilder sql = - new StringBuilder("SELECT ") - .append(columns) - .append(" FROM ") - .append(IoTDBIdentifierUtils.quoteIdentifier(table)); - if (filterQueries != null && !filterQueries.isEmpty()) { - sql.append(" WHERE ").append(String.join(" AND ", filterQueries)); - } - if (limit >= 0) { - sql.append(" LIMIT ").append(limit); - } - return sql.toString(); - } -} 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 new file mode 100644 index 0000000..908dd59 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBUtils.java @@ -0,0 +1,259 @@ +/* + * 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 org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.types.DataType; +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; + +/** + * Shared utilities for the IoTDB relational (table model) Flink connector. + * + * <p>This class centralizes identifier quoting, Flink/IoTDB data type conversion and the column + * name/category resolution used by both the Flink catalog (DDL) and the sink serializer. + */ +public final class IoTDBUtils { + + private IoTDBUtils() {} + + /** + * Quotes a resolved logical identifier using IoTDB's double-quote syntax. + * + * <p>The input is treated as a logical name, not as SQL text. Backticks are preserved as part of + * the identifier and double quotes are escaped by doubling. + */ + public static String quoteIdentifier(String identifier) { + if (identifier == null) { + throw new IllegalArgumentException("Identifier must not be null."); + } + return "\"" + identifier.replace("\"", "\"\"") + "\""; + } + + public static TSDataType toIoTDBDataType(DataType dataType) { + switch (dataType.getLogicalType().getTypeRoot()) { + case BOOLEAN: + return TSDataType.BOOLEAN; + case TINYINT: + case SMALLINT: + case INTEGER: + return TSDataType.INT32; + case BIGINT: + return TSDataType.INT64; + case FLOAT: + return TSDataType.FLOAT; + case DOUBLE: + return TSDataType.DOUBLE; + case CHAR: + case VARCHAR: + return TSDataType.STRING; + case BINARY: + case VARBINARY: + return TSDataType.BLOB; + case DATE: + return TSDataType.DATE; + case TIMESTAMP_WITHOUT_TIME_ZONE: + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + return TSDataType.TIMESTAMP; + default: + throw new CatalogException("Unsupported Flink data type for IoTDB: " + dataType); + } + } + + public static DataType toFlinkDataType(TSDataType dataType) { + switch (dataType) { + case BOOLEAN: + return DataTypes.BOOLEAN(); + case INT32: + return DataTypes.INT(); + case INT64: + return DataTypes.BIGINT(); + case FLOAT: + return DataTypes.FLOAT(); + case DOUBLE: + return DataTypes.DOUBLE(); + case TEXT: + case STRING: + return DataTypes.STRING(); + case BLOB: + return DataTypes.BYTES(); + case DATE: + return DataTypes.DATE(); + case TIMESTAMP: + return DataTypes.TIMESTAMP(3); + default: + throw new CatalogException("Unsupported IoTDB data type: " + dataType); + } + } + + /** Normalizes a column name for option matching. */ + public static String normalizeColumnName(String columnName) { + if (columnName == null) { + throw new CatalogException("Column name must not be null."); + } + return columnName.trim().toLowerCase(Locale.ROOT); + } + + /** + * Builds a lookup from normalized column name to {@link TSDataType}. + * + * @param columnNames column names in schema order + * @param dataTypes data types aligned with {@code columnNames} + */ + public static Map<String, TSDataType> normalizeDataTypesByColumn( + List<String> columnNames, List<TSDataType> dataTypes) { + if (columnNames == null || dataTypes == null || columnNames.size() != dataTypes.size()) { + throw new CatalogException("Column names and data types must have the same size."); + } + Map<String, TSDataType> dataTypesByColumn = new HashMap<>(); + for (int i = 0; i < columnNames.size(); i++) { + dataTypesByColumn.put(normalizeColumnName(columnNames.get(i)), dataTypes.get(i)); + } + return dataTypesByColumn; + } + + /** + * Validates the {@code time-column}, {@code tag-columns} and {@code attribute-columns} options + * against an existing table schema. + * + * @param timeColumn normalized TIME column name + * @param tagColumns normalized TAG column names + * @param attributeColumns normalized ATTRIBUTE column names + * @param dataTypesByColumn table columns keyed by normalized name + */ + public static void validateColumnCategories( + String timeColumn, + Collection<String> tagColumns, + Collection<String> attributeColumns, + Map<String, TSDataType> dataTypesByColumn) { + if (timeColumn == null || timeColumn.isEmpty()) { + throw new CatalogException("Table option 'time-column' must specify the IoTDB TIME column."); + } + if (tagColumns.contains(timeColumn) || attributeColumns.contains(timeColumn)) { + throw new CatalogException("The TIME column cannot also be a TAG or ATTRIBUTE column."); + } + Set<String> overlappingColumns = new HashSet<>(tagColumns); + overlappingColumns.retainAll(attributeColumns); + if (!overlappingColumns.isEmpty()) { + throw new CatalogException( + "TAG and ATTRIBUTE columns must not overlap: " + overlappingColumns.iterator().next()); + } + validateColumnExists(timeColumn, "time-column", dataTypesByColumn); + for (String columnName : tagColumns) { + validateColumnExists(columnName, "tag-columns", dataTypesByColumn); + } + for (String columnName : attributeColumns) { + validateColumnExists(columnName, "attribute-columns", dataTypesByColumn); + } + if (dataTypesByColumn.get(timeColumn) != TSDataType.TIMESTAMP) { + throw new CatalogException("The IoTDB TIME column must use the TIMESTAMP data type."); + } + } + + /** + * Resolves the {@link ColumnCategory} of every column, in schema order. The returned list + * includes exactly one {@link ColumnCategory#TIME} entry. + * + * @param columnNames column names in schema order + * @param timeColumn normalized TIME column name + * @param tagColumns normalized TAG column names + * @param attributeColumns normalized ATTRIBUTE column names + */ + public static List<ColumnCategory> resolveColumnCategories( + List<String> columnNames, + String timeColumn, + Collection<String> tagColumns, + Collection<String> attributeColumns) { + List<ColumnCategory> categories = new ArrayList<>(columnNames.size()); + for (String columnName : columnNames) { + String normalizedColumnName = normalizeColumnName(columnName); + if (normalizedColumnName.equals(timeColumn)) { + categories.add(ColumnCategory.TIME); + } else if (tagColumns.contains(normalizedColumnName)) { + categories.add(ColumnCategory.TAG); + } else if (attributeColumns.contains(normalizedColumnName)) { + categories.add(ColumnCategory.ATTRIBUTE); + } else { + categories.add(ColumnCategory.FIELD); + } + } + return categories; + } + + /** + * Builds a bounded table-model scan query. + * + * @param table IoTDB table name + * @param rowDataType projected Flink row type whose field names define the SELECT list + * @param filterQueries already-rendered IoTDB predicate fragments + * @param limit maximum number of rows, or a negative value for no limit + * @return IoTDB SELECT SQL + */ + public static String buildSelectQuery( + String table, DataType rowDataType, List<String> filterQueries, long limit) { + List<String> fieldNames = DataType.getFieldNames(rowDataType); + if (fieldNames.isEmpty()) { + throw new IllegalArgumentException("IoTDB source requires at least one selected column."); + } + + StringBuilder columns = new StringBuilder(); + for (String fieldName : fieldNames) { + if (columns.length() > 0) { + columns.append(", "); + } + columns.append(quoteIdentifier(fieldName)); + } + + StringBuilder sql = + new StringBuilder("SELECT ") + .append(columns) + .append(" FROM ") + .append(quoteIdentifier(table)); + if (filterQueries != null && !filterQueries.isEmpty()) { + sql.append(" WHERE ").append(String.join(" AND ", filterQueries)); + } + if (limit >= 0) { + sql.append(" LIMIT ").append(limit); + } + return sql.toString(); + } + + private static void validateColumnExists( + String columnName, String optionName, Map<String, TSDataType> dataTypesByColumn) { + if (!dataTypesByColumn.containsKey(columnName)) { + throw new CatalogException( + "Column '" + + columnName + + "' declared by table option '" + + optionName + + "' does not exist."); + } + } +} 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 new file mode 100644 index 0000000..d4e93c6 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverterTest.java @@ -0,0 +1,114 @@ +/* + * 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.sink; + +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.types.RowKind; +import org.apache.tsfile.utils.Binary; +import org.junit.Test; + +import java.io.IOException; +import java.time.LocalDate; +import java.util.Arrays; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +public class RowDataSinkDataConverterTest { + + private final RowDataSinkDataConverter converter = new RowDataSinkDataConverter(); + + @Test + public void testYieldsOneRowPerInput() throws IOException { + try (SinkDataConverter.Iterator iterator = converter.getIterator(rowData())) { + assertTrue(iterator.next()); + assertFalse(iterator.next()); + } + + try (SinkDataConverter.Iterator iterator = converter.getIterator(rowData())) { + assertTrue(iterator.next()); + } + } + + @Test + public void testGetters() throws IOException { + try (SinkDataConverter.Iterator iterator = converter.getIterator(rowData())) { + assertTrue(iterator.next()); + + assertEquals(1000L, iterator.getTimestamp(0)); + assertEquals("t", iterator.getString(1)); + assertEquals(7, iterator.getInt(2)); + assertEquals(2.5d, iterator.getDouble(3), 0.0d); + assertEquals(LocalDate.ofEpochDay(10), iterator.getDate(4)); + assertArrayEquals(new byte[] {1, 2}, binary(iterator.getBlob(5))); + } + } + + @Test + public void testNullField() throws IOException { + GenericRowData row = rowData(); + row.setField(2, null); + try (SinkDataConverter.Iterator iterator = converter.getIterator(row)) { + assertTrue(iterator.next()); + + assertTrue(iterator.isNull(2)); + assertFalse(iterator.isNull(1)); + } + } + + @Test + public void testNullInputHasNoRows() throws IOException { + try (SinkDataConverter.Iterator iterator = converter.getIterator(null)) { + assertFalse(iterator.next()); + } + } + + @Test + public void testRejectsNonInsertRowKind() { + GenericRowData row = rowData(); + row.setRowKind(RowKind.DELETE); + + try { + converter.getIterator(row); + fail("Expected an IOException for a non-INSERT record."); + } catch (IOException expected) { + assertTrue(expected.getMessage().contains("INSERT")); + } + } + + private static GenericRowData rowData() { + return GenericRowData.of( + TimestampData.fromEpochMillis(1000L), + StringData.fromString("t"), + 7, + 2.5d, + 10, + new byte[] {1, 2}); + } + + private static byte[] binary(Binary value) { + return Arrays.copyOf(value.getValues(), value.getLength()); + } +} 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 249bd73..3e3d6c0 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 @@ -19,8 +19,8 @@ package org.apache.iotdb.relational.flink.catalog; -import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; -import org.apache.iotdb.relational.flink.utils.IoTDBRelationalTypeUtils; +import org.apache.iotdb.relational.flink.cfg.IoTDBOptions; +import org.apache.iotdb.relational.flink.utils.IoTDBUtils; import org.apache.flink.table.api.Schema; import org.apache.flink.table.catalog.AbstractCatalog; @@ -53,7 +53,6 @@ import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; -import java.util.Locale; import java.util.Map; import java.util.Set; @@ -66,10 +65,10 @@ import java.util.Set; */ public class IoTDBCatalog extends AbstractCatalog { - private final IoTDBRelationalOptions options; + private final IoTDBOptions options; private final IoTDBCatalogClient catalogClient; - public IoTDBCatalog(String catalogName, String defaultDatabase, IoTDBRelationalOptions options) { + public IoTDBCatalog(String catalogName, String defaultDatabase, IoTDBOptions options) { super(catalogName, defaultDatabase); this.options = options; this.catalogClient = new IoTDBCatalogClient(options); @@ -343,10 +342,10 @@ public class IoTDBCatalog extends AbstractCatalog { Map<String, String> tableOptions = table.getOptions(); String timeColumn = getRequiredTimeColumn(tableOptions); Set<String> tagColumns = - parseColumnNames(tableOptions.get(IoTDBRelationalOptions.TAG_COLUMNS.key()), "tag-columns"); + parseColumnNames(tableOptions.get(IoTDBOptions.TAG_COLUMNS.key()), "tag-columns"); Set<String> attributeColumns = parseColumnNames( - tableOptions.get(IoTDBRelationalOptions.ATTRIBUTE_COLUMNS.key()), "attribute-columns"); + tableOptions.get(IoTDBOptions.ATTRIBUTE_COLUMNS.key()), "attribute-columns"); List<String> columnNames = new ArrayList<>(); List<TSDataType> dataTypes = new ArrayList<>(); @@ -363,30 +362,30 @@ public class IoTDBCatalog extends AbstractCatalog { } String columnName = column.getName(); - TSDataType dataType = IoTDBRelationalTypeUtils.toIoTDBDataType((DataType) abstractDataType); + TSDataType dataType = IoTDBUtils.toIoTDBDataType((DataType) abstractDataType); columnNames.add(columnName); dataTypes.add(dataType); - dataTypesByColumn.put(normalizeColumnName(columnName), dataType); + dataTypesByColumn.put(IoTDBUtils.normalizeColumnName(columnName), dataType); } if (columnNames.isEmpty()) { throw new CatalogException("An IoTDB table must contain at least one column."); } - validateColumnCategories(timeColumn, tagColumns, attributeColumns, dataTypesByColumn); + IoTDBUtils.validateColumnCategories( + timeColumn, tagColumns, attributeColumns, dataTypesByColumn); - List<ColumnCategory> categories = new ArrayList<>(); - for (String columnName : columnNames) { - categories.add(toColumnCategory(columnName, timeColumn, tagColumns, attributeColumns)); - } + List<ColumnCategory> categories = + IoTDBUtils.resolveColumnCategories( + columnNames, timeColumn, tagColumns, attributeColumns); return new TableSchema(tablePath.getObjectName(), columnNames, dataTypes, categories); } private static String getRequiredTimeColumn(Map<String, String> tableOptions) { - String timeColumn = tableOptions.get(IoTDBRelationalOptions.TIME_COLUMN.key()); + String timeColumn = tableOptions.get(IoTDBOptions.TIME_COLUMN.key()); if (timeColumn == null || timeColumn.trim().isEmpty()) { throw new CatalogException( "Table option 'time-column' must specify the IoTDB TIME column for CREATE TABLE."); } - return normalizeColumnName(timeColumn); + return IoTDBUtils.normalizeColumnName(timeColumn); } private static Set<String> parseColumnNames(String value, String optionName) { @@ -399,7 +398,7 @@ public class IoTDBCatalog extends AbstractCatalog { throw new CatalogException( "Table option '" + optionName + "' contains an empty column name."); } - String normalizedColumnName = normalizeColumnName(columnName); + String normalizedColumnName = IoTDBUtils.normalizeColumnName(columnName); if (!columnNames.add(normalizedColumnName)) { throw new CatalogException( "Table option '" + optionName + "' contains duplicate column: " + columnName); @@ -408,63 +407,6 @@ public class IoTDBCatalog extends AbstractCatalog { return columnNames; } - private static void validateColumnCategories( - String timeColumn, - Set<String> tagColumns, - Set<String> attributeColumns, - Map<String, TSDataType> dataTypesByColumn) { - if (tagColumns.contains(timeColumn) || attributeColumns.contains(timeColumn)) { - throw new CatalogException("The TIME column cannot also be a TAG or ATTRIBUTE column."); - } - Set<String> overlappingColumns = new HashSet<>(tagColumns); - overlappingColumns.retainAll(attributeColumns); - if (!overlappingColumns.isEmpty()) { - throw new CatalogException( - "TAG and ATTRIBUTE columns must not overlap: " + overlappingColumns.iterator().next()); - } - validateColumnExists(timeColumn, "time-column", dataTypesByColumn); - for (String columnName : tagColumns) { - validateColumnExists(columnName, "tag-columns", dataTypesByColumn); - } - for (String columnName : attributeColumns) { - validateColumnExists(columnName, "attribute-columns", dataTypesByColumn); - } - if (dataTypesByColumn.get(timeColumn) != TSDataType.TIMESTAMP) { - throw new CatalogException("The IoTDB TIME column must use the TIMESTAMP data type."); - } - } - - private static void validateColumnExists( - String columnName, String optionName, Map<String, TSDataType> dataTypesByColumn) { - if (!dataTypesByColumn.containsKey(columnName)) { - throw new CatalogException( - "Column '" - + columnName - + "' declared by table option '" - + optionName - + "' does not exist."); - } - } - - private static ColumnCategory toColumnCategory( - String columnName, String timeColumn, Set<String> tagColumns, Set<String> attributeColumns) { - String normalizedColumnName = normalizeColumnName(columnName); - if (timeColumn.equals(normalizedColumnName)) { - return ColumnCategory.TIME; - } - if (tagColumns.contains(normalizedColumnName)) { - return ColumnCategory.TAG; - } - if (attributeColumns.contains(normalizedColumnName)) { - return ColumnCategory.ATTRIBUTE; - } - return ColumnCategory.FIELD; - } - - private static String normalizeColumnName(String columnName) { - return columnName.trim().toLowerCase(Locale.ROOT); - } - private CatalogTable toCatalogTable( TableSchema tableSchema, String databaseName, String tableName) { Schema.Builder schemaBuilder = Schema.newBuilder(); @@ -476,7 +418,7 @@ public class IoTDBCatalog extends AbstractCatalog { for (int i = 0; i < columns.size(); i++) { IMeasurementSchema column = columns.get(i); schemaBuilder.column( - column.getMeasurementName(), IoTDBRelationalTypeUtils.toFlinkDataType(column.getType())); + column.getMeasurementName(), IoTDBUtils.toFlinkDataType(column.getType())); switch (categories.get(i)) { case TIME: timeColumn = column.getMeasurementName(); @@ -497,26 +439,26 @@ public class IoTDBCatalog extends AbstractCatalog { } Map<String, String> tableOptions = new HashMap<>(); - tableOptions.put(FactoryUtil.CONNECTOR.key(), IoTDBRelationalOptions.IDENTIFIER); + tableOptions.put(FactoryUtil.CONNECTOR.key(), IoTDBOptions.IDENTIFIER); tableOptions.put( - IoTDBRelationalOptions.NODE_URLS.key(), String.join(",", options.getNodeUrls())); - tableOptions.put(IoTDBRelationalOptions.USER.key(), options.getUsername()); - tableOptions.put(IoTDBRelationalOptions.PASSWORD.key(), options.getPassword()); - tableOptions.put(IoTDBRelationalOptions.DATABASE.key(), databaseName); - tableOptions.put(IoTDBRelationalOptions.TABLE.key(), tableName); - tableOptions.put(IoTDBRelationalOptions.TIME_COLUMN.key(), timeColumn); + IoTDBOptions.NODE_URLS.key(), String.join(",", options.getNodeUrls())); + tableOptions.put(IoTDBOptions.USER.key(), options.getUsername()); + tableOptions.put(IoTDBOptions.PASSWORD.key(), options.getPassword()); + tableOptions.put(IoTDBOptions.DATABASE.key(), databaseName); + tableOptions.put(IoTDBOptions.TABLE.key(), tableName); + tableOptions.put(IoTDBOptions.TIME_COLUMN.key(), timeColumn); if (!tagColumns.isEmpty()) { - tableOptions.put(IoTDBRelationalOptions.TAG_COLUMNS.key(), String.join(",", tagColumns)); + tableOptions.put(IoTDBOptions.TAG_COLUMNS.key(), String.join(",", tagColumns)); } if (!attributeColumns.isEmpty()) { tableOptions.put( - IoTDBRelationalOptions.ATTRIBUTE_COLUMNS.key(), String.join(",", attributeColumns)); + IoTDBOptions.ATTRIBUTE_COLUMNS.key(), String.join(",", attributeColumns)); } return CatalogTable.of(schemaBuilder.build(), null, Collections.emptyList(), tableOptions); } - public IoTDBRelationalOptions getOptions() { + public IoTDBOptions getOptions() { return options; } 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 ec088be..b29fde5 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 @@ -19,7 +19,7 @@ package org.apache.iotdb.relational.flink.catalog; -import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; +import org.apache.iotdb.relational.flink.cfg.IoTDBOptions; import org.apache.flink.configuration.ConfigOption; import org.apache.flink.configuration.ReadableConfig; @@ -36,38 +36,38 @@ public class IoTDBCatalogFactory implements CatalogFactory { @Override public String factoryIdentifier() { - return IoTDBRelationalOptions.IDENTIFIER; + return IoTDBOptions.IDENTIFIER; } @Override public Set<ConfigOption<?>> requiredOptions() { - return new HashSet<>(Arrays.asList(IoTDBRelationalOptions.NODE_URLS)); + return new HashSet<>(Arrays.asList(IoTDBOptions.NODE_URLS)); } @Override public Set<ConfigOption<?>> optionalOptions() { return new HashSet<>( Arrays.asList( - IoTDBRelationalOptions.USER, - IoTDBRelationalOptions.PASSWORD, - IoTDBRelationalOptions.DEFAULT_DATABASE)); + IoTDBOptions.USER, + IoTDBOptions.PASSWORD, + IoTDBOptions.DEFAULT_DATABASE)); } @Override public Catalog createCatalog(Context context) { FactoryUtil.CatalogFactoryHelper helper = FactoryUtil.createCatalogFactoryHelper(this, context); helper.validate(); - IoTDBRelationalOptions options = toOptions(helper.getOptions()); + IoTDBOptions options = toOptions(helper.getOptions()); return new IoTDBCatalog(context.getName(), options.getDefaultDatabase(), options); } - private static IoTDBRelationalOptions toOptions(ReadableConfig config) { - return IoTDBRelationalOptions.builder() + private static IoTDBOptions toOptions(ReadableConfig config) { + return IoTDBOptions.builder() .withNodeUrls( - Arrays.asList(((String) config.get(IoTDBRelationalOptions.NODE_URLS)).split(","))) - .withUsername(config.get(IoTDBRelationalOptions.USER)) - .withPassword(config.get(IoTDBRelationalOptions.PASSWORD)) - .withDefaultDatabase(config.get(IoTDBRelationalOptions.DEFAULT_DATABASE)) + Arrays.asList(((String) config.get(IoTDBOptions.NODE_URLS)).split(","))) + .withUsername(config.get(IoTDBOptions.USER)) + .withPassword(config.get(IoTDBOptions.PASSWORD)) + .withDefaultDatabase(config.get(IoTDBOptions.DEFAULT_DATABASE)) .build(); } } diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/sink/IoTDBSink.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/sink/IoTDBSink.java index b676c4a..9b283b8 100644 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/sink/IoTDBSink.java +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/sink/IoTDBSink.java @@ -19,11 +19,11 @@ package org.apache.iotdb.relational.flink.sink; -import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; -import org.apache.iotdb.relational.flink.sink.serializer.IoTDBTabletSerializer; +import org.apache.iotdb.relational.flink.cfg.IoTDBOptions; import org.apache.flink.api.connector.sink2.Sink; import org.apache.flink.api.connector.sink2.SinkWriter; +import org.apache.flink.table.types.DataType; import java.io.IOException; @@ -34,10 +34,12 @@ import java.io.IOException; * * <ul> * <li>Table API/SQL creates {@code IoTDBSink<RowData>}. - * <li>DataStream API can create {@code IoTDBSink<IN>} with a user-provided serializer. + * <li>DataStream API can create {@code IoTDBSink<IN>} with a user-provided {@link + * SinkDataConverter}. * </ul> * - * <p>The writer owns the IoTDB session and batches serialized tablets before insertion. + * <p>The sink only carries serializable state (options, the physical row {@link DataType} and the + * converter); the writer owns the IoTDB session and the buffered tablet. * * @param <IN> input record type */ @@ -45,17 +47,22 @@ public class IoTDBSink<IN> implements Sink<IN> { private static final long serialVersionUID = 1L; - private final IoTDBRelationalOptions options; - private final IoTDBTabletSerializer<IN> serializer; + private final IoTDBOptions options; + private final DataType physicalRowDataType; + private final SinkDataConverter<IN> converter; - public IoTDBSink(IoTDBRelationalOptions options, IoTDBTabletSerializer<IN> serializer) { + public IoTDBSink( + IoTDBOptions options, + DataType physicalRowDataType, + SinkDataConverter<IN> converter) { this.options = options; - this.serializer = serializer; + this.physicalRowDataType = physicalRowDataType; + this.converter = converter; } @Override public SinkWriter<IN> createWriter(InitContext context) throws IOException { - return new IoTDBSinkWriter<>(options, serializer); + return new IoTDBSinkWriter<>(options, physicalRowDataType, converter); } public static <IN> Builder<IN> builder() { @@ -65,21 +72,27 @@ public class IoTDBSink<IN> implements Sink<IN> { /** Builder for the DataStream API entry point. */ public static class Builder<IN> { - private IoTDBRelationalOptions options; - private IoTDBTabletSerializer<IN> serializer; + private IoTDBOptions options; + private DataType physicalRowDataType; + private SinkDataConverter<IN> converter; - public Builder<IN> setOptions(IoTDBRelationalOptions options) { + public Builder<IN> setOptions(IoTDBOptions options) { this.options = options; return this; } - public Builder<IN> setSerializer(IoTDBTabletSerializer<IN> serializer) { - this.serializer = serializer; + public Builder<IN> setPhysicalRowDataType(DataType physicalRowDataType) { + this.physicalRowDataType = physicalRowDataType; + return this; + } + + public Builder<IN> setConverter(SinkDataConverter<IN> converter) { + this.converter = converter; return this; } public IoTDBSink<IN> build() { - return new IoTDBSink<>(options, serializer); + return new IoTDBSink<>(options, physicalRowDataType, converter); } } } 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 681eacb..cbb045c 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 @@ -20,20 +20,30 @@ package org.apache.iotdb.relational.flink.sink; import org.apache.iotdb.isession.ITableSession; -import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; -import org.apache.iotdb.relational.flink.sink.serializer.IoTDBTabletSerializer; +import org.apache.iotdb.relational.flink.cfg.IoTDBOptions; +import org.apache.iotdb.relational.flink.utils.IoTDBUtils; import org.apache.iotdb.session.TableSessionBuilder; import org.apache.flink.api.connector.sink2.SinkWriter; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.types.DataType; +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.write.record.Tablet; import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; /** * Sink writer of the IoTDB relational (table model) Flink connector. * - * <p>The writer owns one IoTDB session and one buffered tablet. Serialized tablets are validated, - * merged into the buffer, and flushed through {@code ITableSession.insert(Tablet)}. + * <p>The writer owns one IoTDB session and one buffered tablet. For each input element it obtains a + * {@link SinkDataConverter.Iterator}, appends every produced row to the tablet, and flushes the + * tablet through {@code ITableSession.insert(Tablet)} when it is full. * * @param <IN> input record type */ @@ -41,17 +51,27 @@ public class IoTDBSinkWriter<IN> implements SinkWriter<IN> { private static final int DEFAULT_BATCH_SIZE = 1024; - private final IoTDBRelationalOptions options; - private final IoTDBTabletSerializer<IN> serializer; + private final IoTDBOptions options; + private final DataType physicalRowDataType; + private final SinkDataConverter<IN> converter; private ITableSession session; private Tablet buffer; private boolean closed; - public IoTDBSinkWriter(IoTDBRelationalOptions options, IoTDBTabletSerializer<IN> serializer) + private List<String> columnNames; + private List<TSDataType> columnTsTypes; + private int[] columnIndexes; + private int timeColumnIndex; + + public IoTDBSinkWriter( + IoTDBOptions options, + DataType physicalRowDataType, + SinkDataConverter<IN> converter) throws IOException { this.options = options; - this.serializer = serializer; + this.physicalRowDataType = physicalRowDataType; + this.converter = converter; open(); } @@ -59,19 +79,13 @@ public class IoTDBSinkWriter<IN> implements SinkWriter<IN> { public void write(IN element, Context context) throws IOException, InterruptedException { ensureOpen(); - if (buffer.getRowSize() >= buffer.getMaxRowNumber()) { - flushBuffer(); - } - - int rowSizeBefore = buffer.getRowSize(); - boolean full = serializer.serialize(element, buffer); - int rowSizeAfter = buffer.getRowSize(); - - if (rowSizeAfter < rowSizeBefore || rowSizeAfter > buffer.getMaxRowNumber()) { - throw new IOException("Serializer produced an invalid tablet row count."); - } - if (full || rowSizeAfter >= buffer.getMaxRowNumber()) { - flushBuffer(); + try (SinkDataConverter.Iterator iterator = converter.getIterator(element)) { + while (iterator.next()) { + if (buffer.getRowSize() >= buffer.getMaxRowNumber()) { + flushBuffer(); + } + appendCurrentRow(iterator); + } } } @@ -98,7 +112,7 @@ public class IoTDBSinkWriter<IN> implements SinkWriter<IN> { } try { - serializer.close(); + converter.close(); } catch (Exception e) { if (failure == null) { failure = e; @@ -136,17 +150,143 @@ public class IoTDBSinkWriter<IN> implements SinkWriter<IN> { builder.database(options.getDatabase()); } session = builder.build(); - serializer.open(); - buffer = serializer.createTablet(DEFAULT_BATCH_SIZE); - if (buffer == null || buffer.getMaxRowNumber() <= 0) { - throw new IOException("Serializer created an invalid tablet buffer."); - } + converter.open(); + initializeTablet(); } catch (Exception e) { closeQuietly(); throw new IOException("Failed to open IoTDB sink writer.", e); } } + private void initializeTablet() throws IOException { + String tableName = options.getTable(); + if (tableName == null || tableName.trim().isEmpty()) { + throw new IOException("Table option 'table' must specify the IoTDB table name."); + } + + String rawTimeColumn = options.getTimeColumn(); + if (rawTimeColumn == null || rawTimeColumn.trim().isEmpty()) { + throw new IOException( + "Table option 'time-column' is required to write into an IoTDB table model table."); + } + + try { + List<String> fieldNames = DataType.getFieldNames(physicalRowDataType); + List<DataType> fieldTypes = DataType.getFieldDataTypes(physicalRowDataType); + if (fieldNames.isEmpty()) { + throw new IOException("The IoTDB table sink requires at least one column."); + } + + List<TSDataType> fieldTsTypes = new ArrayList<>(fieldTypes.size()); + for (DataType fieldType : fieldTypes) { + fieldTsTypes.add(IoTDBUtils.toIoTDBDataType(fieldType)); + } + + String timeColumn = IoTDBUtils.normalizeColumnName(rawTimeColumn); + Set<String> tagColumns = normalizeColumns(options.getTagColumns()); + Set<String> attributeColumns = normalizeColumns(options.getAttributeColumns()); + + IoTDBUtils.validateColumnCategories( + timeColumn, + tagColumns, + attributeColumns, + IoTDBUtils.normalizeDataTypesByColumn(fieldNames, fieldTsTypes)); + + List<ColumnCategory> categories = + IoTDBUtils.resolveColumnCategories(fieldNames, timeColumn, tagColumns, attributeColumns); + + List<String> names = new ArrayList<>(fieldNames.size() - 1); + List<TSDataType> tsTypes = new ArrayList<>(fieldNames.size() - 1); + List<ColumnCategory> nonTimeCategories = new ArrayList<>(fieldNames.size() - 1); + List<Integer> indexes = new ArrayList<>(fieldNames.size() - 1); + int resolvedTimeIndex = -1; + for (int i = 0; i < fieldNames.size(); i++) { + if (categories.get(i) == ColumnCategory.TIME) { + resolvedTimeIndex = i; + continue; + } + names.add(fieldNames.get(i)); + tsTypes.add(fieldTsTypes.get(i)); + nonTimeCategories.add(categories.get(i)); + indexes.add(i); + } + if (resolvedTimeIndex < 0) { + throw new IOException( + "The IoTDB TIME column '" + rawTimeColumn + "' does not exist in the sink schema."); + } + + this.timeColumnIndex = resolvedTimeIndex; + this.columnNames = Collections.unmodifiableList(names); + this.columnTsTypes = Collections.unmodifiableList(tsTypes); + this.columnIndexes = indexes.stream().mapToInt(Integer::intValue).toArray(); + this.buffer = + new Tablet( + tableName, + columnNames, + columnTsTypes, + Collections.unmodifiableList(nonTimeCategories), + DEFAULT_BATCH_SIZE); + if (buffer.getMaxRowNumber() <= 0) { + throw new IOException("Tablet has an invalid max row number."); + } + } catch (CatalogException e) { + throw new IOException("Invalid IoTDB table sink schema: " + e.getMessage(), e); + } + } + + private void appendCurrentRow(SinkDataConverter.Iterator iterator) throws IOException { + if (iterator.isNull(timeColumnIndex)) { + throw new IOException( + "The IoTDB TIME column '" + + options.getTimeColumn() + + "' must not be null for INSERT records."); + } + + int row = buffer.getRowSize(); + buffer.addTimestamp(row, iterator.getTimestamp(timeColumnIndex)); + for (int i = 0; i < columnNames.size(); i++) { + buffer.addValue( + columnNames.get(i), row, readValue(iterator, columnIndexes[i], columnTsTypes.get(i))); + } + } + + private Object readValue( + SinkDataConverter.Iterator iterator, int columnIndex, TSDataType dataType) throws IOException { + if (iterator.isNull(columnIndex)) { + return null; + } + switch (dataType) { + case BOOLEAN: + return iterator.getBoolean(columnIndex); + case INT32: + return iterator.getInt(columnIndex); + case INT64: + return iterator.getLong(columnIndex); + case FLOAT: + return iterator.getFloat(columnIndex); + case DOUBLE: + return iterator.getDouble(columnIndex); + case TEXT: + case STRING: + return iterator.getString(columnIndex); + case BLOB: + return iterator.getBlob(columnIndex); + case DATE: + return iterator.getDate(columnIndex); + case TIMESTAMP: + return iterator.getTimestamp(columnIndex); + default: + throw new IOException("Unsupported IoTDB data type for sink: " + dataType); + } + } + + private static Set<String> normalizeColumns(List<String> columns) { + if (columns == null) { + return Collections.emptySet(); + } + return columns.stream().map(IoTDBUtils::normalizeColumnName).collect(Collectors.toSet()); + } + private void ensureOpen() throws IOException { if (closed || session == null) { throw new IOException("IoTDB sink writer is already closed."); 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/IoTDBSource.java index ecaeb0a..ea4abc3 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/IoTDBSource.java @@ -19,7 +19,7 @@ package org.apache.iotdb.relational.flink.source; -import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; +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; @@ -49,22 +49,14 @@ public class IoTDBSource<OUT> implements Source<OUT, IoTDBSourceSplit, IoTDBSour private static final long serialVersionUID = 1L; - private final IoTDBRelationalOptions options; + private final IoTDBOptions options; private final DataType rowDataType; private final IoTDBDeserializationSchema<OUT> deserializer; private final List<String> filterQueries; private final long limit; public IoTDBSource( - IoTDBRelationalOptions options, - DataType rowDataType, - IoTDBDeserializationSchema<OUT> deserializer, - List<String> filterQueries) { - this(options, rowDataType, deserializer, filterQueries, -1L); - } - - public IoTDBSource( - IoTDBRelationalOptions options, + IoTDBOptions options, DataType rowDataType, IoTDBDeserializationSchema<OUT> deserializer, List<String> filterQueries, 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/IoTDBSourceEnumerator.java index ef0dd5d..29ea548 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/IoTDBSourceEnumerator.java @@ -19,10 +19,10 @@ package org.apache.iotdb.relational.flink.source; -import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; +import org.apache.iotdb.relational.flink.cfg.IoTDBOptions; import org.apache.iotdb.relational.flink.source.enumerator.IoTDBSourceEnumeratorState; import org.apache.iotdb.relational.flink.source.split.IoTDBSourceSplit; -import org.apache.iotdb.relational.flink.utils.IoTDBSQLBuilder; +import org.apache.iotdb.relational.flink.utils.IoTDBUtils; import org.apache.flink.api.connector.source.SplitEnumerator; import org.apache.flink.api.connector.source.SplitEnumeratorContext; @@ -44,7 +44,7 @@ public class IoTDBSourceEnumerator implements SplitEnumerator<IoTDBSourceSplit, IoTDBSourceEnumeratorState> { private final SplitEnumeratorContext<IoTDBSourceSplit> context; - private final IoTDBRelationalOptions options; + private final IoTDBOptions options; private final DataType rowDataType; private final List<String> filterQueries; private final long limit; @@ -57,7 +57,7 @@ public class IoTDBSourceEnumerator public IoTDBSourceEnumerator( SplitEnumeratorContext<IoTDBSourceSplit> context, - IoTDBRelationalOptions options, + IoTDBOptions options, DataType rowDataType, List<String> filterQueries, long limit) { @@ -66,7 +66,7 @@ public class IoTDBSourceEnumerator public IoTDBSourceEnumerator( SplitEnumeratorContext<IoTDBSourceSplit> context, - IoTDBRelationalOptions options, + IoTDBOptions options, DataType rowDataType, List<String> filterQueries, long limit, @@ -146,7 +146,7 @@ public class IoTDBSourceEnumerator private IoTDBSourceSplit createSingleSplit() { String splitId = UUID.randomUUID().toString(); String sql = - IoTDBSQLBuilder.buildSelectQuery( + IoTDBUtils.buildSelectQuery( options.getTable(), rowDataType, filterQueries, limit); return new IoTDBSourceSplit(splitId, options.getDatabase(), options.getTable(), sql); } 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/IoTDBSourceReader.java index 5bceafd..8415756 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/IoTDBSourceReader.java @@ -21,7 +21,7 @@ package org.apache.iotdb.relational.flink.source; import org.apache.iotdb.isession.ITableSession; import org.apache.iotdb.isession.SessionDataSet; -import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; +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.session.TableSessionBuilder; @@ -42,7 +42,7 @@ import java.util.concurrent.CompletableFuture; public class IoTDBSourceReader<OUT> implements SourceReader<OUT, IoTDBSourceSplit> { private final SourceReaderContext context; - private final IoTDBRelationalOptions options; + private final IoTDBOptions options; private final DataType rowDataType; private final IoTDBDeserializationSchema<OUT> deserializer; private final Deque<IoTDBSourceSplit> pendingSplits = new ArrayDeque<>(); @@ -57,7 +57,7 @@ public class IoTDBSourceReader<OUT> implements SourceReader<OUT, IoTDBSourceSpli public IoTDBSourceReader( SourceReaderContext context, - IoTDBRelationalOptions options, + IoTDBOptions options, DataType rowDataType, IoTDBDeserializationSchema<OUT> deserializer) { this.context = context; 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 a7a865f..5c45b73 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 @@ -19,7 +19,7 @@ package org.apache.iotdb.relational.flink.table; -import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; +import org.apache.iotdb.relational.flink.cfg.IoTDBOptions; import org.apache.flink.configuration.ConfigOption; import org.apache.flink.configuration.ReadableConfig; @@ -63,40 +63,40 @@ public class IoTDBRelationalDynamicTableFactory @Override public String factoryIdentifier() { - return IoTDBRelationalOptions.IDENTIFIER; + return IoTDBOptions.IDENTIFIER; } @Override public Set<ConfigOption<?>> requiredOptions() { return new HashSet<>( Arrays.asList( - IoTDBRelationalOptions.NODE_URLS, - IoTDBRelationalOptions.DATABASE, - IoTDBRelationalOptions.TABLE)); + IoTDBOptions.NODE_URLS, + IoTDBOptions.DATABASE, + IoTDBOptions.TABLE)); } @Override public Set<ConfigOption<?>> optionalOptions() { return new HashSet<>( Arrays.asList( - IoTDBRelationalOptions.USER, - IoTDBRelationalOptions.PASSWORD, - IoTDBRelationalOptions.TIME_COLUMN, - IoTDBRelationalOptions.TAG_COLUMNS, - IoTDBRelationalOptions.ATTRIBUTE_COLUMNS)); + IoTDBOptions.USER, + IoTDBOptions.PASSWORD, + IoTDBOptions.TIME_COLUMN, + IoTDBOptions.TAG_COLUMNS, + IoTDBOptions.ATTRIBUTE_COLUMNS)); } - private static IoTDBRelationalOptions toOptions(ReadableConfig config) { - return IoTDBRelationalOptions.builder() - .withNodeUrls(Arrays.asList(config.get(IoTDBRelationalOptions.NODE_URLS).split(","))) - .withUsername(config.get(IoTDBRelationalOptions.USER)) - .withPassword(config.get(IoTDBRelationalOptions.PASSWORD)) - .withDatabase(config.get(IoTDBRelationalOptions.DATABASE)) - .withTable(config.get(IoTDBRelationalOptions.TABLE)) - .withTimeColumn(config.get(IoTDBRelationalOptions.TIME_COLUMN)) - .withTagColumns(parseColumnNames(config.get(IoTDBRelationalOptions.TAG_COLUMNS))) + private static IoTDBOptions toOptions(ReadableConfig config) { + return IoTDBOptions.builder() + .withNodeUrls(Arrays.asList(config.get(IoTDBOptions.NODE_URLS).split(","))) + .withUsername(config.get(IoTDBOptions.USER)) + .withPassword(config.get(IoTDBOptions.PASSWORD)) + .withDatabase(config.get(IoTDBOptions.DATABASE)) + .withTable(config.get(IoTDBOptions.TABLE)) + .withTimeColumn(config.get(IoTDBOptions.TIME_COLUMN)) + .withTagColumns(parseColumnNames(config.get(IoTDBOptions.TAG_COLUMNS))) .withAttributeColumns( - parseColumnNames(config.get(IoTDBRelationalOptions.ATTRIBUTE_COLUMNS))) + parseColumnNames(config.get(IoTDBOptions.ATTRIBUTE_COLUMNS))) .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 6f55e63..eb52dfe 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 @@ -19,10 +19,10 @@ package org.apache.iotdb.relational.flink.table; -import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; +import org.apache.iotdb.relational.flink.cfg.IoTDBOptions; import org.apache.iotdb.relational.flink.sink.IoTDBSink; -import org.apache.iotdb.relational.flink.sink.serializer.IoTDBTabletSerializer; -import org.apache.iotdb.relational.flink.sink.serializer.RowDataIoTDBTabletSerializer; +import org.apache.iotdb.relational.flink.sink.RowDataSinkDataConverter; +import org.apache.iotdb.relational.flink.sink.SinkDataConverter; import org.apache.flink.table.catalog.ResolvedSchema; import org.apache.flink.table.connector.ChangelogMode; @@ -38,10 +38,10 @@ import org.apache.flink.table.data.RowData; */ public class IoTDBRelationalDynamicTableSink implements DynamicTableSink { - private final IoTDBRelationalOptions options; + private final IoTDBOptions options; private final ResolvedSchema schema; - public IoTDBRelationalDynamicTableSink(IoTDBRelationalOptions options, ResolvedSchema schema) { + public IoTDBRelationalDynamicTableSink(IoTDBOptions options, ResolvedSchema schema) { this.options = options; this.schema = schema; } @@ -53,8 +53,9 @@ public class IoTDBRelationalDynamicTableSink implements DynamicTableSink { @Override public SinkRuntimeProvider getSinkRuntimeProvider(Context context) { - IoTDBTabletSerializer<RowData> serializer = new RowDataIoTDBTabletSerializer(options, schema); - return SinkV2Provider.of(new IoTDBSink<>(options, serializer)); + SinkDataConverter<RowData> converter = new RowDataSinkDataConverter(); + return SinkV2Provider.of( + new IoTDBSink<>(options, schema.toPhysicalRowDataType(), converter)); } @Override 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 87d359d..a3cd50e 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 @@ -19,7 +19,7 @@ package org.apache.iotdb.relational.flink.table; -import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; +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.pushdown.IoTDBExpressionVisitor; @@ -52,13 +52,13 @@ public class IoTDBRelationalDynamicTableSource SupportsLimitPushDown, SupportsProjectionPushDown { - private final IoTDBRelationalOptions options; + private final IoTDBOptions options; private final ResolvedSchema schema; private DataType physicalRowDataType; private final List<String> resolvedFilterQueries = new ArrayList<>(); private long limit = -1L; - public IoTDBRelationalDynamicTableSource(IoTDBRelationalOptions options, ResolvedSchema schema) { + public IoTDBRelationalDynamicTableSource(IoTDBOptions options, ResolvedSchema schema) { this.options = options; this.schema = schema; this.physicalRowDataType = schema.toPhysicalRowDataType(); diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/IoTDBRelationalLocalQueryManual.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/IoTDBRelationalLocalQueryManual.java index 6423d40..02b4b5e 100644 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/IoTDBRelationalLocalQueryManual.java +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/IoTDBRelationalLocalQueryManual.java @@ -29,6 +29,9 @@ import org.apache.flink.util.CloseableIterator; * Temporary manual verification class. This is intentionally kept in test sources and should not be * committed as a production test. * + * <p>The flow is write first and then query: rows are inserted through the connector and read back + * with a SELECT. + * * <p>Run with system properties such as: * * <pre> @@ -38,6 +41,9 @@ import org.apache.flink.util.CloseableIterator; * -Diotdb.database=test * -Diotdb.table=sensor * </pre> + * + * <p>The IoTDB table {@code <database>.<table>} must already exist and its columns must match the + * DDL below: {@code time} (TIME), {@code device_id} (TAG), {@code temperature} (FIELD). */ public class IoTDBRelationalLocalQueryManual { @@ -50,12 +56,12 @@ public class IoTDBRelationalLocalQueryManual { TableEnvironment tableEnvironment = TableEnvironment.create(EnvironmentSettings.inBatchMode()); - tableEnvironment.executeSql("DROP TABLE IF EXISTS iotdb_source"); + tableEnvironment.executeSql("DROP TABLE IF EXISTS iotdb_table"); - // Replace these columns with the actual columns and types in the local IoTDB table. + // Replace these columns with the actual columns and categories in the local IoTDB table. String ddl = String.format( - "CREATE TABLE iotdb_source (\n" + "CREATE TABLE iotdb_table (\n" + " `time` TIMESTAMP(3),\n" + " `device_id` STRING,\n" + " `temperature` DOUBLE\n" @@ -65,24 +71,45 @@ public class IoTDBRelationalLocalQueryManual { + " 'user' = '%s',\n" + " 'password' = '%s',\n" + " 'database' = '%s',\n" - + " 'table' = '%s'\n" + + " 'table' = '%s',\n" + + " 'time-column' = 'time',\n" + + " 'tag-columns' = 'device_id'\n" + ")", nodeUrls, user, password, database, table); System.out.println("DDL:\n" + ddl); tableEnvironment.executeSql(ddl); - run(tableEnvironment, "SELECT * FROM iotdb_source LIMIT 5"); + // 1) Write: the connector turns the selected rows into IoTDB tablets. + String insert = + "INSERT INTO iotdb_table\n" + + "SELECT ts, device_id, temperature FROM (\n" + + " VALUES\n" + + " (CAST(TIMESTAMP '2024-01-01 00:00:00' AS TIMESTAMP(3)), 'd1', 20.5),\n" + + " (CAST(TIMESTAMP '2024-01-01 00:00:01' AS TIMESTAMP(3)), 'd1', 21.0),\n" + + " (CAST(TIMESTAMP '2024-01-01 00:00:02' AS TIMESTAMP(3)), 'd2', 19.5)\n" + + ") AS source_table(ts, device_id, temperature)"; + execute(tableEnvironment, insert); + + // 2) Read: the same connector table can be used as a source. + run(tableEnvironment, "SELECT * FROM iotdb_table"); run( tableEnvironment, - "SELECT device_id, temperature FROM iotdb_source WHERE temperature > 0 LIMIT 5"); + "SELECT device_id, temperature FROM iotdb_table WHERE temperature > 0 LIMIT 1"); run( tableEnvironment, - "SELECT device_id, temperature FROM iotdb_source " + "WHERE temperature + 1 > 0 LIMIT 5"); + "SELECT temperature FROM iotdb_table WHERE temperature + 1 > 21 LIMIT 5"); + } + + private static void execute(TableEnvironment tableEnvironment, String sql) throws Exception { + System.out.println("\n=== WRITE ==="); + System.out.println(sql); + tableEnvironment.executeSql(sql).await(); + System.out.println("Write finished."); } private static void run(TableEnvironment tableEnvironment, String sql) throws Exception { - System.out.println("\n=== EXECUTE ==="); + System.out.println("\n=== READ ==="); System.out.println(sql); TableResult result = tableEnvironment.executeSql(sql);
