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 bc896cf4e76bdf26f6835e0885a543fab539c59a Author: shuwenwei <[email protected]> AuthorDate: Mon Sep 14 18:39:56 2026 +0800 sink --- .../flink-iotdb-table-connector-base/pom.xml | 73 +++ .../flink/catalog/IoTDBCatalogClient.java | 265 +++++++++++ .../flink/cfg/IoTDBRelationalOptions.java | 219 +++++++++ .../sink/serializer/IoTDBTabletSerializer.java | 49 ++ .../serializer/RowDataIoTDBTabletSerializer.java | 59 +++ .../flink/utils/IoTDBRelationalTypeUtils.java | 87 ++++ .../flink-iotdb-table-connector-flink1/pom.xml | 61 +++ .../relational/flink/catalog/IoTDBCatalog.java | 525 +++++++++++++++++++++ .../flink/catalog/IoTDBCatalogFactory.java | 73 +++ .../iotdb/relational/flink/sink/IoTDBSink.java | 86 ++++ .../relational/flink/sink/IoTDBSinkWriter.java | 62 +++ .../table/IoTDBRelationalDynamicTableFactory.java | 115 +++++ .../table/IoTDBRelationalDynamicTableSink.java | 70 +++ .../table/IoTDBRelationalDynamicTableSource.java | 71 +++ .../org.apache.flink.table.factories.Factory | 20 + .../flink-iotdb-table-connector-flink2/pom.xml | 61 +++ .../flink/IoTDBRelationalDynamicTableFactory.java | 64 +++ .../org.apache.flink.table.factories.Factory | 19 + connectors/flink-iotdb-table-connector/pom.xml | 108 +++++ connectors/pom.xml | 2 + pom.xml | 2 +- 21 files changed, 2090 insertions(+), 1 deletion(-) 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 new file mode 100644 index 0000000..984d209 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/pom.xml @@ -0,0 +1,73 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + 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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.iotdb</groupId> + <artifactId>flink-iotdb-table-connector</artifactId> + <version>2.0.4-SNAPSHOT</version> + <relativePath>../pom.xml</relativePath> + </parent> + <artifactId>flink-iotdb-table-connector-base</artifactId> + <packaging>jar</packaging> + <name>IoTDB: Connector: Apache Flink Table Base</name> + <description>Common code for the IoTDB table model Flink connector</description> + <dependencies> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-api-java-bridge</artifactId> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-common</artifactId> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-runtime</artifactId> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-base</artifactId> + </dependency> + <dependency> + <groupId>org.apache.iotdb</groupId> + <artifactId>iotdb-session</artifactId> + <version>${iotdb.version}</version> + </dependency> + <dependency> + <groupId>org.apache.iotdb</groupId> + <artifactId>isession</artifactId> + <version>${iotdb.version}</version> + </dependency> + <dependency> + <groupId>org.apache.iotdb</groupId> + <artifactId>service-rpc</artifactId> + <version>${iotdb.version}</version> + </dependency> + <dependency> + <groupId>org.apache.tsfile</groupId> + <artifactId>common</artifactId> + <version>${tsfile.version}</version> + </dependency> + <dependency> + <groupId>org.apache.tsfile</groupId> + <artifactId>tsfile</artifactId> + <version>${tsfile.version}</version> + </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 new file mode 100644 index 0000000..7919dad --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalogClient.java @@ -0,0 +1,265 @@ +/* + * 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.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.session.pool.TableSessionPoolBuilder; + +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.read.common.Field; +import org.apache.tsfile.read.common.RowRecord; +import org.apache.tsfile.write.schema.IMeasurementSchema; + +import java.util.ArrayList; +import java.util.List; +import java.util.Locale; + +/** + * Common IoTDB catalog metadata access layer. + * + * <p>This class implements metadata and basic DDL operations used by the Flink catalog. It talks + * to IoTDB through the table model session and the following SQL statements: + * + * <pre> + * SHOW DATABASES + * SHOW TABLES FROM <database> + * DESC <database>.<table> + * CREATE/DROP DATABASE + * CREATE/DROP TABLE + * </pre> + */ +public class IoTDBCatalogClient implements AutoCloseable { + + private static final String COLUMN_DATABASE = "Database"; + private static final String COLUMN_TABLE_NAME = "TableName"; + private static final String COLUMN_COLUMN_NAME = "ColumnName"; + private static final String COLUMN_DATA_TYPE = "DataType"; + private static final String COLUMN_CATEGORY = "Category"; + + private final IoTDBRelationalOptions options; + private volatile ITableSessionPool sessionPool; + + public IoTDBCatalogClient(IoTDBRelationalOptions options) { + this.options = options; + } + + public List<String> listDatabases() { + return querySingleColumn("SHOW DATABASES", COLUMN_DATABASE); + } + + public List<String> listTables(String database) { + return querySingleColumn( + "SHOW TABLES FROM " + quoteIdentifier(database), COLUMN_TABLE_NAME); + } + + public boolean databaseExists(String database) { + return listDatabases().contains(database); + } + + public boolean tableExists(String database, String table) { + return databaseExists(database) && listTables(database).contains(table); + } + + public void createDatabase(String database) { + executeNonQuery("CREATE DATABASE " + quoteIdentifier(database)); + } + + public void dropDatabase(String database) { + executeNonQuery("DROP DATABASE " + quoteIdentifier(database)); + } + + public void createTable(String database, String table, TableSchema tableSchema) { + StringBuilder sql = + new StringBuilder("CREATE TABLE ") + .append(quoteIdentifier(database)) + .append(".") + .append(quoteIdentifier(table)) + .append(" ("); + List<IMeasurementSchema> columns = tableSchema.getColumnSchemas(); + List<ColumnCategory> categories = tableSchema.getColumnTypes(); + for (int i = 0; i < columns.size(); i++) { + if (i > 0) { + sql.append(", "); + } + IMeasurementSchema column = columns.get(i); + sql.append(quoteIdentifier(column.getMeasurementName())) + .append(" ") + .append(toSqlDataType(column.getType())) + .append(" ") + .append(categories.get(i).name()); + } + sql.append(")"); + executeNonQuery(sql.toString()); + } + + public void dropTable(String database, String table) { + executeNonQuery( + "DROP TABLE " + + quoteIdentifier(database) + + "." + + quoteIdentifier(table)); + } + + public TableSchema getTable(String database, String table) { + String sql = "DESC " + quoteIdentifier(database) + "." + quoteIdentifier(table); + try (ITableSession session = getSessionPool().getSession(); + SessionDataSet dataSet = session.executeQueryStatement(sql)) { + int nameIndex = findColumnIndex(dataSet, COLUMN_COLUMN_NAME, 0); + int typeIndex = findColumnIndex(dataSet, COLUMN_DATA_TYPE, 1); + int categoryIndex = findColumnIndex(dataSet, COLUMN_CATEGORY, 2); + + List<String> columnNames = new ArrayList<>(); + List<TSDataType> dataTypes = new ArrayList<>(); + List<ColumnCategory> categories = new ArrayList<>(); + while (dataSet.hasNext()) { + RowRecord row = dataSet.next(); + String columnName = getString(row, nameIndex); + String dataTypeName = getString(row, typeIndex); + String categoryName = getString(row, categoryIndex); + columnNames.add(columnName); + dataTypes.add(TSDataType.valueOf(dataTypeName.toUpperCase(Locale.ROOT))); + categories.add(ColumnCategory.valueOf(categoryName.toUpperCase(Locale.ROOT))); + } + return new TableSchema(table, columnNames, dataTypes, categories); + } catch (CatalogException e) { + throw e; + } catch (Exception e) { + throw new CatalogException("Failed to describe IoTDB table: " + database + "." + table, e); + } + } + + public IoTDBRelationalOptions getOptions() { + return options; + } + + /** Opens the session pool. Called by the owning Flink catalog during {@code Catalog.open()}. */ + public synchronized void open() { + if (sessionPool != null) { + return; + } + try { + TableSessionPoolBuilder builder = + new TableSessionPoolBuilder() + .nodeUrls(options.getNodeUrls()) + .user(options.getUsername()) + .password(options.getPassword()); + if (options.getDatabase() != null) { + builder.database(options.getDatabase()); + } + sessionPool = builder.build(); + } catch (Exception e) { + throw new CatalogException("Failed to open IoTDB table session pool.", e); + } + } + + /** Closes the session pool. Called by the owning Flink catalog during {@code Catalog.close()}. */ + @Override + public synchronized void close() { + if (sessionPool != null) { + try { + sessionPool.close(); + } finally { + sessionPool = null; + } + } + } + + private List<String> querySingleColumn(String sql, String columnName) { + try (ITableSession session = getSessionPool().getSession(); + SessionDataSet dataSet = session.executeQueryStatement(sql)) { + int columnIndex = findColumnIndex(dataSet, columnName, 0); + List<String> values = new ArrayList<>(); + while (dataSet.hasNext()) { + String value = getString(dataSet.next(), columnIndex); + if (value != null) { + values.add(value); + } + } + return values; + } catch (CatalogException e) { + throw e; + } catch (Exception e) { + throw new CatalogException("Failed to execute IoTDB catalog query: " + sql, e); + } + } + + private void executeNonQuery(String sql) { + try (ITableSession session = getSessionPool().getSession()) { + session.executeNonQueryStatement(sql); + } catch (CatalogException e) { + throw e; + } catch (Exception e) { + throw new CatalogException("Failed to execute IoTDB catalog statement: " + sql, e); + } + } + + private static String toSqlDataType(TSDataType dataType) { + switch (dataType) { + case BOOLEAN: + case INT32: + case INT64: + case FLOAT: + case DOUBLE: + case TEXT: + case STRING: + case BLOB: + case DATE: + case TIMESTAMP: + return dataType.name(); + default: + throw new CatalogException("Unsupported IoTDB data type: " + dataType); + } + } + + private static String quoteIdentifier(String identifier) { + return "\"" + identifier.replace("\"", "\"\"") + "\""; + } + + private ITableSessionPool getSessionPool() { + ITableSessionPool pool = sessionPool; + if (pool == null) { + throw new CatalogException("IoTDB catalog is not open."); + } + return pool; + } + + private static int findColumnIndex(SessionDataSet dataSet, String columnName, int fallbackIndex) { + List<String> columnNames = dataSet.getColumnNames(); + if (columnNames != null) { + for (int i = 0; i < columnNames.size(); i++) { + if (columnName.equalsIgnoreCase(columnNames.get(i))) { + return i; + } + } + } + return fallbackIndex; + } + + private static String getString(RowRecord row, int index) { + Field field = row.getField(index); + return field == null ? null : field.getStringValue(); + } +} 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/IoTDBRelationalOptions.java new file mode 100644 index 0000000..3a656c1 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/cfg/IoTDBRelationalOptions.java @@ -0,0 +1,219 @@ +/* + * 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.cfg; + +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; + +import java.io.Serializable; +import java.util.Collections; +import java.util.List; + +/** + * Options and Flink table option keys of the IoTDB relational (table model) Flink connector. + * + * <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 { + + private static final long serialVersionUID = 1L; + + public static final String IDENTIFIER = "iotdb-relational"; + + public static final ConfigOption<String> NODE_URLS = + ConfigOptions.key("nodeUrls").stringType().noDefaultValue(); + + public static final ConfigOption<String> USER = + ConfigOptions.key("user").stringType().defaultValue("root"); + + public static final ConfigOption<String> PASSWORD = + ConfigOptions.key("password").stringType().defaultValue("root"); + + public static final ConfigOption<String> DATABASE = + ConfigOptions.key("database").stringType().noDefaultValue(); + + public static final ConfigOption<String> TABLE = + ConfigOptions.key("table").stringType().noDefaultValue(); + + public static final ConfigOption<String> DEFAULT_DATABASE = + ConfigOptions.key("default-database").stringType().defaultValue("public"); + + public static final ConfigOption<String> TIME_COLUMN = + ConfigOptions.key("time-column").stringType().noDefaultValue(); + + public static final ConfigOption<String> TAG_COLUMNS = + ConfigOptions.key("tag-columns").stringType().defaultValue(""); + + public static final ConfigOption<String> ATTRIBUTE_COLUMNS = + ConfigOptions.key("attribute-columns").stringType().defaultValue(""); + + private final List<String> nodeUrls; + private final String username; + private final String password; + private final String database; + private final String table; + private final String defaultDatabase; + private final String timeColumn; + private final List<String> tagColumns; + private final List<String> attributeColumns; + + private IoTDBRelationalOptions(Builder builder) { + this.nodeUrls = builder.nodeUrls; + this.username = builder.username; + this.password = builder.password; + this.database = builder.database; + this.table = builder.table; + this.defaultDatabase = builder.defaultDatabase; + this.timeColumn = builder.timeColumn; + this.tagColumns = builder.tagColumns; + this.attributeColumns = builder.attributeColumns; + } + + /** + * @return IoTDB node urls, e.g. {@code ["127.0.0.1:6667"]}. + */ + public List<String> getNodeUrls() { + return nodeUrls; + } + + /** + * @return IoTDB username. + */ + public String getUsername() { + return username; + } + + /** + * @return IoTDB password. + */ + public String getPassword() { + return password; + } + + /** + * @return IoTDB database name. + */ + public String getDatabase() { + return database; + } + + /** + * @return IoTDB table name. + */ + public String getTable() { + return table; + } + + /** + * @return the default database used by the catalog. + */ + public String getDefaultDatabase() { + return defaultDatabase; + } + + /** @return the configured IoTDB time-column name, or {@code null} when unspecified. */ + public String getTimeColumn() { + return timeColumn; + } + + /** @return configured IoTDB TAG column names. */ + public List<String> getTagColumns() { + return tagColumns; + } + + /** @return configured IoTDB ATTRIBUTE column names. */ + public List<String> getAttributeColumns() { + return attributeColumns; + } + + /** + * @return a new builder + */ + public static Builder builder() { + return new Builder(); + } + + /** Builder of {@link IoTDBRelationalOptions}. */ + public static class Builder { + + private List<String> nodeUrls; + private String username; + private String password; + private String database; + private String table; + private String defaultDatabase = "public"; + private String timeColumn; + private List<String> tagColumns = Collections.emptyList(); + private List<String> attributeColumns = Collections.emptyList(); + + public Builder withNodeUrls(List<String> nodeUrls) { + this.nodeUrls = nodeUrls; + return this; + } + + public Builder withUsername(String username) { + this.username = username; + return this; + } + + public Builder withPassword(String password) { + this.password = password; + return this; + } + + public Builder withDatabase(String database) { + this.database = database; + return this; + } + + public Builder withTable(String table) { + this.table = table; + return this; + } + + public Builder withDefaultDatabase(String defaultDatabase) { + this.defaultDatabase = defaultDatabase; + return this; + } + + public Builder withTimeColumn(String timeColumn) { + this.timeColumn = timeColumn; + return this; + } + + public Builder withTagColumns(List<String> tagColumns) { + this.tagColumns = tagColumns; + return this; + } + + public Builder withAttributeColumns(List<String> attributeColumns) { + this.attributeColumns = attributeColumns; + return this; + } + + /** + * @return the built options + */ + public IoTDBRelationalOptions build() { + return new IoTDBRelationalOptions(this); + } + } +} 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 new file mode 100644 index 0000000..83441e3 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/serializer/IoTDBTabletSerializer.java @@ -0,0 +1,49 @@ +/* + * 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; + +/** + * Converts one Flink input record into an IoTDB {@link Tablet}. + * + * <p>The returned tablet may contain zero, one, or multiple rows. The sink writer owns batching and + * flushing, while the serializer only defines how an input record is represented as tablet rows. + * + * @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 {} + + /** + * Serializes one input record. + * + * <p>TODO: define the exact owner of the returned tablet and the batch-size contract. + */ + Tablet serialize(IN record) 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 new file mode 100644 index 0000000..0a30dda --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/serializer/RowDataIoTDBTabletSerializer.java @@ -0,0 +1,59 @@ +/* + * 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 serialize(RowData record) 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/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 new file mode 100644 index 0000000..066e3e1 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBRelationalTypeUtils.java @@ -0,0 +1,87 @@ +/* + * 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-flink1/pom.xml b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/pom.xml new file mode 100644 index 0000000..c1a292a --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/pom.xml @@ -0,0 +1,61 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + 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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.iotdb</groupId> + <artifactId>flink-iotdb-table-connector</artifactId> + <version>2.0.4-SNAPSHOT</version> + <relativePath>../pom.xml</relativePath> + </parent> + <artifactId>flink-iotdb-table-connector-${flink.major.version}</artifactId> + <packaging>jar</packaging> + <name>IoTDB: Connector: Apache Flink Table (Flink ${flink.version})</name> + <description>Flink-version-specific module for the IoTDB table model connector</description> + <dependencies> + <dependency> + <groupId>org.apache.iotdb</groupId> + <artifactId>flink-iotdb-table-connector-base</artifactId> + <version>${project.version}</version> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-core</artifactId> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-streaming-java</artifactId> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-api-java-bridge</artifactId> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-common</artifactId> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-runtime</artifactId> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-base</artifactId> + </dependency> + </dependencies> +</project> 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 new file mode 100644 index 0000000..cf4a5b9 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalog.java @@ -0,0 +1,525 @@ +/* + * 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.catalog; + +import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; +import org.apache.iotdb.relational.flink.utils.IoTDBRelationalTypeUtils; + +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.catalog.AbstractCatalog; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogDatabase; +import org.apache.flink.table.catalog.CatalogDatabaseImpl; +import org.apache.flink.table.catalog.CatalogFunction; +import org.apache.flink.table.catalog.CatalogPartition; +import org.apache.flink.table.catalog.CatalogPartitionSpec; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.catalog.stats.CatalogColumnStatistics; +import org.apache.flink.table.catalog.stats.CatalogTableStatistics; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.types.DataType; +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.write.schema.IMeasurementSchema; + +import java.util.ArrayList; +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; + +/** + * Flink 1.x Catalog adapter for IoTDB relational tables. + * + * <p>Flink Catalog API adaptation belongs here. IoTDB metadata access belongs in {@link + * IoTDBCatalogClient}; the current implementation supports database/table discovery, table + * schema resolution, and basic database/table DDL. + */ +public class IoTDBCatalog extends AbstractCatalog { + + private final IoTDBRelationalOptions options; + private final IoTDBCatalogClient catalogClient; + + public IoTDBCatalog(String catalogName, String defaultDatabase, IoTDBRelationalOptions options) { + super(catalogName, defaultDatabase); + this.options = options; + this.catalogClient = new IoTDBCatalogClient(options); + } + + @Override + public void open() throws CatalogException { + catalogClient.open(); + } + + @Override + public void close() throws CatalogException { + catalogClient.close(); + } + + @Override + public List<String> listDatabases() { + return catalogClient.listDatabases(); + } + + @Override + public CatalogDatabase getDatabase(String databaseName) throws DatabaseNotExistException { + if (!catalogClient.databaseExists(databaseName)) { + throw new DatabaseNotExistException(getName(), databaseName); + } + return new CatalogDatabaseImpl(Collections.emptyMap(), null); + } + + @Override + public boolean databaseExists(String databaseName) { + return catalogClient.databaseExists(databaseName); + } + + @Override + public void createDatabase(String name, CatalogDatabase database, boolean ignoreIfExists) + throws DatabaseAlreadyExistException, CatalogException { + if (catalogClient.databaseExists(name)) { + if (ignoreIfExists) { + return; + } + throw new DatabaseAlreadyExistException(getName(), name); + } + catalogClient.createDatabase(name); + } + + @Override + public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade) + throws DatabaseNotEmptyException, DatabaseNotExistException, CatalogException { + if (!catalogClient.databaseExists(name)) { + if (ignoreIfNotExists) { + return; + } + throw new DatabaseNotExistException(getName(), name); + } + if (!cascade && !catalogClient.listTables(name).isEmpty()) { + throw new DatabaseNotEmptyException(getName(), name); + } + catalogClient.dropDatabase(name); + } + + @Override + public void alterDatabase(String name, CatalogDatabase newDatabase, boolean ignoreIfNotExists) { + throw unsupported(); + } + + @Override + public List<String> listTables(String databaseName) throws DatabaseNotExistException { + if (!catalogClient.databaseExists(databaseName)) { + throw new DatabaseNotExistException(getName(), databaseName); + } + return catalogClient.listTables(databaseName); + } + + @Override + public List<String> listViews(String databaseName) { + throw unsupported(); + } + + @Override + public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException { + String databaseName = tablePath.getDatabaseName(); + String tableName = tablePath.getObjectName(); + if (!catalogClient.tableExists(databaseName, tableName)) { + throw new TableNotExistException(getName(), tablePath); + } + return toCatalogTable(catalogClient.getTable(databaseName, tableName), databaseName, tableName); + } + + @Override + public boolean tableExists(ObjectPath tablePath) { + return catalogClient.tableExists(tablePath.getDatabaseName(), tablePath.getObjectName()); + } + + @Override + public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists) + throws TableNotExistException, CatalogException { + if (!catalogClient.tableExists(tablePath.getDatabaseName(), tablePath.getObjectName())) { + if (ignoreIfNotExists) { + return; + } + throw new TableNotExistException(getName(), tablePath); + } + catalogClient.dropTable(tablePath.getDatabaseName(), tablePath.getObjectName()); + } + + @Override + public void renameTable(ObjectPath tablePath, String newTableName, boolean ignoreIfNotExists) { + throw unsupported(); + } + + @Override + public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) + throws TableAlreadyExistException, DatabaseNotExistException, CatalogException { + String databaseName = tablePath.getDatabaseName(); + if (!catalogClient.databaseExists(databaseName)) { + throw new DatabaseNotExistException(getName(), databaseName); + } + if (catalogClient.tableExists(databaseName, tablePath.getObjectName())) { + if (ignoreIfExists) { + return; + } + throw new TableAlreadyExistException(getName(), tablePath); + } + catalogClient.createTable( + databaseName, tablePath.getObjectName(), toTableSchema(tablePath, table)); + } + + @Override + public void alterTable( + ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists) { + throw unsupported(); + } + + @Override + public List<CatalogPartitionSpec> listPartitions(ObjectPath tablePath) { + throw unsupported(); + } + + @Override + public List<CatalogPartitionSpec> listPartitions( + ObjectPath tablePath, CatalogPartitionSpec partitionSpec) { + throw unsupported(); + } + + @Override + public List<CatalogPartitionSpec> listPartitionsByFilter( + ObjectPath tablePath, List<Expression> filters) { + throw unsupported(); + } + + @Override + public CatalogPartition getPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) { + throw unsupported(); + } + + @Override + public boolean partitionExists(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) { + throw unsupported(); + } + + @Override + public void createPartition( + ObjectPath tablePath, + CatalogPartitionSpec partitionSpec, + CatalogPartition partition, + boolean ignoreIfExists) { + throw unsupported(); + } + + @Override + public void dropPartition( + ObjectPath tablePath, CatalogPartitionSpec partitionSpec, boolean ignoreIfNotExists) { + throw unsupported(); + } + + @Override + public void alterPartition( + ObjectPath tablePath, + CatalogPartitionSpec partitionSpec, + CatalogPartition newPartition, + boolean ignoreIfNotExists) { + throw unsupported(); + } + + @Override + public List<String> listFunctions(String dbName) { + throw unsupported(); + } + + @Override + public CatalogFunction getFunction(ObjectPath functionPath) { + throw unsupported(); + } + + @Override + public boolean functionExists(ObjectPath functionPath) { + throw unsupported(); + } + + @Override + public void createFunction( + ObjectPath functionPath, CatalogFunction function, boolean ignoreIfExists) { + throw unsupported(); + } + + @Override + public void alterFunction( + ObjectPath functionPath, CatalogFunction newFunction, boolean ignoreIfNotExists) { + throw unsupported(); + } + + @Override + public void dropFunction(ObjectPath functionPath, boolean ignoreIfNotExists) { + throw unsupported(); + } + + @Override + public CatalogTableStatistics getTableStatistics(ObjectPath tablePath) { + throw unsupported(); + } + + @Override + public CatalogColumnStatistics getTableColumnStatistics(ObjectPath tablePath) { + throw unsupported(); + } + + @Override + public CatalogTableStatistics getPartitionStatistics( + ObjectPath tablePath, CatalogPartitionSpec partitionSpec) { + throw unsupported(); + } + + @Override + public CatalogColumnStatistics getPartitionColumnStatistics( + ObjectPath tablePath, CatalogPartitionSpec partitionSpec) { + throw unsupported(); + } + + @Override + public void alterTableStatistics( + ObjectPath tablePath, CatalogTableStatistics tableStatistics, boolean ignoreIfNotExists) { + throw unsupported(); + } + + @Override + public void alterTableColumnStatistics( + ObjectPath tablePath, CatalogColumnStatistics columnStatistics, boolean ignoreIfNotExists) { + throw unsupported(); + } + + @Override + public void alterPartitionStatistics( + ObjectPath tablePath, + CatalogPartitionSpec partitionSpec, + CatalogTableStatistics partitionStatistics, + boolean ignoreIfNotExists) { + throw unsupported(); + } + + @Override + public void alterPartitionColumnStatistics( + ObjectPath tablePath, + CatalogPartitionSpec partitionSpec, + CatalogColumnStatistics partitionColumnStatistics, + boolean ignoreIfNotExists) { + throw unsupported(); + } + + private TableSchema toTableSchema(ObjectPath tablePath, CatalogBaseTable table) { + Schema schema = table.getUnresolvedSchema(); + Map<String, String> tableOptions = table.getOptions(); + String timeColumn = getRequiredTimeColumn(tableOptions); + Set<String> tagColumns = + parseColumnNames(tableOptions.get(IoTDBRelationalOptions.TAG_COLUMNS.key()), "tag-columns"); + Set<String> attributeColumns = + parseColumnNames( + tableOptions.get(IoTDBRelationalOptions.ATTRIBUTE_COLUMNS.key()), "attribute-columns"); + + List<String> columnNames = new ArrayList<>(); + List<TSDataType> dataTypes = new ArrayList<>(); + Map<String, TSDataType> dataTypesByColumn = new HashMap<>(); + for (Schema.UnresolvedColumn column : schema.getColumns()) { + if (!(column instanceof Schema.UnresolvedPhysicalColumn)) { + throw new CatalogException( + "Only physical columns are supported by the IoTDB table model catalog."); + } + Object abstractDataType = ((Schema.UnresolvedPhysicalColumn) column).getDataType(); + if (!(abstractDataType instanceof DataType)) { + throw new CatalogException( + "Unresolved column data type is not supported: " + column.getName()); + } + + String columnName = column.getName(); + TSDataType dataType = + IoTDBRelationalTypeUtils.toIoTDBDataType((DataType) abstractDataType); + columnNames.add(columnName); + dataTypes.add(dataType); + dataTypesByColumn.put(normalizeColumnName(columnName), dataType); + } + if (columnNames.isEmpty()) { + throw new CatalogException("An IoTDB table must contain at least one column."); + } + validateColumnCategories(timeColumn, tagColumns, attributeColumns, dataTypesByColumn); + + List<ColumnCategory> categories = new ArrayList<>(); + for (String columnName : columnNames) { + categories.add(toColumnCategory(columnName, 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()); + 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); + } + + private static Set<String> parseColumnNames(String value, String optionName) { + Set<String> columnNames = new HashSet<>(); + if (value == null || value.trim().isEmpty()) { + return columnNames; + } + for (String columnName : value.split(",", -1)) { + if (columnName.trim().isEmpty()) { + throw new CatalogException("Table option '" + optionName + "' contains an empty column name."); + } + String normalizedColumnName = normalizeColumnName(columnName); + if (!columnNames.add(normalizedColumnName)) { + throw new CatalogException( + "Table option '" + optionName + "' contains duplicate column: " + columnName); + } + } + 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(); + List<IMeasurementSchema> columns = tableSchema.getColumnSchemas(); + List<ColumnCategory> categories = tableSchema.getColumnTypes(); + String timeColumn = null; + List<String> tagColumns = new ArrayList<>(); + List<String> attributeColumns = new ArrayList<>(); + for (int i = 0; i < columns.size(); i++) { + IMeasurementSchema column = columns.get(i); + schemaBuilder.column( + column.getMeasurementName(), + IoTDBRelationalTypeUtils.toFlinkDataType(column.getType())); + switch (categories.get(i)) { + case TIME: + timeColumn = column.getMeasurementName(); + break; + case TAG: + tagColumns.add(column.getMeasurementName()); + break; + case ATTRIBUTE: + attributeColumns.add(column.getMeasurementName()); + break; + default: + break; + } + } + if (timeColumn == null) { + throw new CatalogException("IoTDB table has no TIME column: " + databaseName + "." + tableName); + } + + Map<String, String> tableOptions = new HashMap<>(); + tableOptions.put(FactoryUtil.CONNECTOR.key(), IoTDBRelationalOptions.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); + if (!tagColumns.isEmpty()) { + tableOptions.put(IoTDBRelationalOptions.TAG_COLUMNS.key(), String.join(",", tagColumns)); + } + if (!attributeColumns.isEmpty()) { + tableOptions.put( + IoTDBRelationalOptions.ATTRIBUTE_COLUMNS.key(), String.join(",", attributeColumns)); + } + + return CatalogTable.of(schemaBuilder.build(), null, Collections.emptyList(), tableOptions); + } + + public IoTDBRelationalOptions getOptions() { + return options; + } + + private static UnsupportedOperationException unsupported() { + return new UnsupportedOperationException("Not implemented yet."); + } +} 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 new file mode 100644 index 0000000..ec088be --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalogFactory.java @@ -0,0 +1,73 @@ +/* + * 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.catalog; + +import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; + +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.factories.CatalogFactory; +import org.apache.flink.table.factories.FactoryUtil; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; + +/** Flink CatalogFactory entry point for the IoTDB relational table model. */ +public class IoTDBCatalogFactory implements CatalogFactory { + + @Override + public String factoryIdentifier() { + return IoTDBRelationalOptions.IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + return new HashSet<>(Arrays.asList(IoTDBRelationalOptions.NODE_URLS)); + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return new HashSet<>( + Arrays.asList( + IoTDBRelationalOptions.USER, + IoTDBRelationalOptions.PASSWORD, + IoTDBRelationalOptions.DEFAULT_DATABASE)); + } + + @Override + public Catalog createCatalog(Context context) { + FactoryUtil.CatalogFactoryHelper helper = FactoryUtil.createCatalogFactoryHelper(this, context); + helper.validate(); + IoTDBRelationalOptions options = toOptions(helper.getOptions()); + return new IoTDBCatalog(context.getName(), options.getDefaultDatabase(), options); + } + + private static IoTDBRelationalOptions toOptions(ReadableConfig config) { + return IoTDBRelationalOptions.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)) + .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 new file mode 100644 index 0000000..46101af --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/sink/IoTDBSink.java @@ -0,0 +1,86 @@ +/* + * 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.iotdb.relational.flink.cfg.IoTDBRelationalOptions; +import org.apache.iotdb.relational.flink.sink.serializer.IoTDBTabletSerializer; + +import org.apache.flink.api.connector.sink2.Sink; +import org.apache.flink.api.connector.sink2.SinkWriter; + +import java.io.IOException; + +/** + * Sink v2 entry of the IoTDB relational (table model) Flink connector. + * + * <p>The generic input type is used by both APIs: + * + * <ul> + * <li>Table API/SQL creates {@code IoTDBSink<RowData>}. + * <li>DataStream API can create {@code IoTDBSink<IN>} with a user-provided serializer. + * </ul> + * + * <p>TODO: implement batching and {@code ITableSession.insert(Tablet)} in the writer. + * + * @param <IN> input record type + */ +public class IoTDBSink<IN> implements Sink<IN> { + + private static final long serialVersionUID = 1L; + + private final IoTDBRelationalOptions options; + private final IoTDBTabletSerializer<IN> serializer; + + public IoTDBSink( + IoTDBRelationalOptions options, IoTDBTabletSerializer<IN> serializer) { + this.options = options; + this.serializer = serializer; + } + + @Override + public SinkWriter<IN> createWriter(InitContext context) throws IOException { + return new IoTDBSinkWriter<>(options, serializer); + } + + public static <IN> Builder<IN> builder() { + return new Builder<>(); + } + + /** Builder for the DataStream API entry point. */ + public static class Builder<IN> { + + private IoTDBRelationalOptions options; + private IoTDBTabletSerializer<IN> serializer; + + public Builder<IN> setOptions(IoTDBRelationalOptions options) { + this.options = options; + return this; + } + + public Builder<IN> setSerializer(IoTDBTabletSerializer<IN> serializer) { + this.serializer = serializer; + return this; + } + + public IoTDBSink<IN> build() { + return new IoTDBSink<>(options, serializer); + } + } +} 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 new file mode 100644 index 0000000..d0a8e30 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/sink/IoTDBSinkWriter.java @@ -0,0 +1,62 @@ +/* + * 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.iotdb.relational.flink.cfg.IoTDBRelationalOptions; +import org.apache.iotdb.relational.flink.sink.serializer.IoTDBTabletSerializer; + +import org.apache.flink.api.connector.sink2.SinkWriter; + +import java.io.IOException; + +/** + * Sink writer of the IoTDB relational (table model) Flink connector. + * + * <p>TODO: open one {@code ITableSession}, buffer serialized tablets, flush through {@code + * ITableSession.insert(Tablet)}, and close all resources. + * + * @param <IN> input record type + */ +public class IoTDBSinkWriter<IN> implements SinkWriter<IN> { + + private final IoTDBRelationalOptions options; + private final IoTDBTabletSerializer<IN> serializer; + + public IoTDBSinkWriter( + IoTDBRelationalOptions options, IoTDBTabletSerializer<IN> serializer) { + this.options = options; + this.serializer = serializer; + } + + @Override + public void write(IN element, Context context) throws IOException, InterruptedException { + throw new UnsupportedOperationException("Not implemented yet."); + } + + @Override + public void flush(boolean endOfInput) throws IOException, InterruptedException { + throw new UnsupportedOperationException("Not implemented yet."); + } + + @Override + public void close() throws Exception { + // TODO: flush remaining rows and close the IoTDB session. + } +} 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 new file mode 100644 index 0000000..5b7bfd9 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableFactory.java @@ -0,0 +1,115 @@ +/* + * 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.table; + +import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; + +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.table.connector.sink.DynamicTableSink; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.factories.DynamicTableSinkFactory; +import org.apache.flink.table.factories.DynamicTableSourceFactory; +import org.apache.flink.table.factories.FactoryUtil; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.HashSet; +import java.util.Set; + +/** + * SPI factory of the IoTDB relational (table model) Flink connector. + * + * <p>The factory only wires options and the dynamic table source/sink. Runtime read/write logic is + * intentionally not implemented yet. + */ +public class IoTDBRelationalDynamicTableFactory + implements DynamicTableSourceFactory, DynamicTableSinkFactory { + + @Override + public DynamicTableSource createDynamicTableSource(Context context) { + FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context); + helper.validate(); + return new IoTDBRelationalDynamicTableSource( + toOptions(helper.getOptions()), context.getCatalogTable().getResolvedSchema()); + } + + @Override + public DynamicTableSink createDynamicTableSink(Context context) { + FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context); + helper.validate(); + return new IoTDBRelationalDynamicTableSink( + toOptions(helper.getOptions()), context.getCatalogTable().getResolvedSchema()); + } + + @Override + public String factoryIdentifier() { + return IoTDBRelationalOptions.IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + return new HashSet<>( + Arrays.asList( + IoTDBRelationalOptions.NODE_URLS, + IoTDBRelationalOptions.DATABASE, + IoTDBRelationalOptions.TABLE)); + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return new HashSet<>( + Arrays.asList( + IoTDBRelationalOptions.USER, + IoTDBRelationalOptions.PASSWORD, + IoTDBRelationalOptions.TIME_COLUMN, + IoTDBRelationalOptions.TAG_COLUMNS, + IoTDBRelationalOptions.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))) + .withAttributeColumns(parseColumnNames(config.get(IoTDBRelationalOptions.ATTRIBUTE_COLUMNS))) + .build(); + } + + private static List<String> parseColumnNames(String value) { + if (value == null || value.trim().isEmpty()) { + return Collections.emptyList(); + } + List<String> columnNames = new ArrayList<>(); + for (String columnName : value.split(",")) { + String trimmedColumnName = columnName.trim(); + if (!trimmedColumnName.isEmpty()) { + columnNames.add(trimmedColumnName); + } + } + return columnNames; + } +} 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 new file mode 100644 index 0000000..82a3d59 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSink.java @@ -0,0 +1,70 @@ +/* + * 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.table; + +import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; +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.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.connector.sink.DynamicTableSink; +import org.apache.flink.table.connector.sink.SinkV2Provider; +import org.apache.flink.table.data.RowData; + +/** + * Dynamic table sink of the IoTDB relational (table model) Flink connector. + * + * <p>Mirrors the structure of Doris' {@code DorisDynamicTableSink}: the Table layer only wraps the + * DataStream Sink v2 implementation through {@link SinkV2Provider}. + */ +public class IoTDBRelationalDynamicTableSink implements DynamicTableSink { + + private final IoTDBRelationalOptions options; + private final ResolvedSchema schema; + + public IoTDBRelationalDynamicTableSink(IoTDBRelationalOptions options, ResolvedSchema schema) { + this.options = options; + this.schema = schema; + } + + @Override + public ChangelogMode getChangelogMode(ChangelogMode requestedMode) { + return ChangelogMode.insertOnly(); + } + + @Override + public SinkRuntimeProvider getSinkRuntimeProvider(Context context) { + IoTDBTabletSerializer<RowData> serializer = + new RowDataIoTDBTabletSerializer(options, schema); + return SinkV2Provider.of(new IoTDBSink<>(options, serializer)); + } + + @Override + public DynamicTableSink copy() { + return new IoTDBRelationalDynamicTableSink(options, schema); + } + + @Override + public String asSummaryString() { + return "IoTDB Relational Dynamic Table Sink"; + } +} 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 new file mode 100644 index 0000000..59bda40 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSource.java @@ -0,0 +1,71 @@ +/* + * 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.table; + +import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; + +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.connector.source.LookupTableSource; +import org.apache.flink.table.connector.source.ScanTableSource; + +/** + * Dynamic table source of the IoTDB relational (table model) Flink connector, covering scan and + * lookup reading. + * + * <p>Mirrors the structure of Doris' {@code DorisDynamicTableSource}. TODO: implement the scan and + * lookup runtime providers. + */ +public class IoTDBRelationalDynamicTableSource implements ScanTableSource, LookupTableSource { + + private final IoTDBRelationalOptions options; + private final ResolvedSchema schema; + + public IoTDBRelationalDynamicTableSource(IoTDBRelationalOptions options, ResolvedSchema schema) { + this.options = options; + this.schema = schema; + } + + @Override + public ChangelogMode getChangelogMode() { + throw new UnsupportedOperationException("Not implemented yet."); + } + + @Override + public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { + throw new UnsupportedOperationException("Not implemented yet."); + } + + @Override + public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext lookupContext) { + throw new UnsupportedOperationException("Not implemented yet."); + } + + @Override + public DynamicTableSource copy() { + return new IoTDBRelationalDynamicTableSource(options, schema); + } + + @Override + public String asSummaryString() { + return "IoTDB Relational Dynamic Table Source"; + } +} diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory new file mode 100644 index 0000000..4281287 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory @@ -0,0 +1,20 @@ +# +# 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. +# +org.apache.iotdb.relational.flink.table.IoTDBRelationalDynamicTableFactory +org.apache.iotdb.relational.flink.catalog.IoTDBCatalogFactory diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink2/pom.xml b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink2/pom.xml new file mode 100644 index 0000000..c1a292a --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink2/pom.xml @@ -0,0 +1,61 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + 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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.iotdb</groupId> + <artifactId>flink-iotdb-table-connector</artifactId> + <version>2.0.4-SNAPSHOT</version> + <relativePath>../pom.xml</relativePath> + </parent> + <artifactId>flink-iotdb-table-connector-${flink.major.version}</artifactId> + <packaging>jar</packaging> + <name>IoTDB: Connector: Apache Flink Table (Flink ${flink.version})</name> + <description>Flink-version-specific module for the IoTDB table model connector</description> + <dependencies> + <dependency> + <groupId>org.apache.iotdb</groupId> + <artifactId>flink-iotdb-table-connector-base</artifactId> + <version>${project.version}</version> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-core</artifactId> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-streaming-java</artifactId> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-api-java-bridge</artifactId> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-common</artifactId> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-runtime</artifactId> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-base</artifactId> + </dependency> + </dependencies> +</project> diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink2/src/main/java/org/apache/iotdb/relational/flink/IoTDBRelationalDynamicTableFactory.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink2/src/main/java/org/apache/iotdb/relational/flink/IoTDBRelationalDynamicTableFactory.java new file mode 100644 index 0000000..c56ca31 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink2/src/main/java/org/apache/iotdb/relational/flink/IoTDBRelationalDynamicTableFactory.java @@ -0,0 +1,64 @@ +/* + * 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; + +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.table.connector.sink.DynamicTableSink; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.factories.DynamicTableSinkFactory; +import org.apache.flink.table.factories.DynamicTableSourceFactory; + +import java.util.Collections; +import java.util.Set; + +/** + * SPI factory of the IoTDB relational (table model) Flink connector. + * + * <p>TODO: fill in the factory identifier, option definitions and source/sink creation logic. + */ +public class IoTDBRelationalDynamicTableFactory + implements DynamicTableSourceFactory, DynamicTableSinkFactory { + + @Override + public DynamicTableSource createDynamicTableSource(Context context) { + throw new UnsupportedOperationException("Not implemented yet."); + } + + @Override + public DynamicTableSink createDynamicTableSink(Context context) { + throw new UnsupportedOperationException("Not implemented yet."); + } + + @Override + public String factoryIdentifier() { + throw new UnsupportedOperationException("Not implemented yet."); + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + return Collections.emptySet(); + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return Collections.emptySet(); + } +} + diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink2/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink2/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory new file mode 100644 index 0000000..6efb5d0 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink2/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory @@ -0,0 +1,19 @@ +# +# 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. +# +org.apache.iotdb.relational.flink.IoTDBRelationalDynamicTableFactory diff --git a/connectors/flink-iotdb-table-connector/pom.xml b/connectors/flink-iotdb-table-connector/pom.xml new file mode 100644 index 0000000..fe0c318 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/pom.xml @@ -0,0 +1,108 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + 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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.iotdb</groupId> + <artifactId>connectors</artifactId> + <version>2.0.4-SNAPSHOT</version> + </parent> + <artifactId>flink-iotdb-table-connector</artifactId> + <packaging>pom</packaging> + <name>IoTDB: Connector: Apache Flink Table</name> + <description>IoTDB table model connector for Apache Flink</description> + <dependencyManagement> + <dependencies> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-core</artifactId> + <version>${flink.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-streaming-java</artifactId> + <version>${flink.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-api-java</artifactId> + <version>${flink.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-api-java-bridge</artifactId> + <version>${flink.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-common</artifactId> + <version>${flink.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-runtime</artifactId> + <version>${flink.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-base</artifactId> + <version>${flink.version}</version> + <scope>provided</scope> + </dependency> + </dependencies> + </dependencyManagement> + <modules> + <module>flink-iotdb-table-connector-base</module> + </modules> + <profiles> + <profile> + <id>flink1</id> + <activation> + <activeByDefault>true</activeByDefault> + </activation> + <properties> + <flink.version>1.17.1</flink.version> + <flink.major.version>1.17</flink.major.version> + <maven.compiler.source>1.8</maven.compiler.source> + <maven.compiler.target>1.8</maven.compiler.target> + </properties> + <modules> + <module>flink-iotdb-table-connector-flink1</module> + </modules> + </profile> + <profile> + <id>flink2</id> + <properties> + <flink.version>2.2.0</flink.version> + <flink.major.version>2.2</flink.major.version> + <maven.compiler.source>17</maven.compiler.source> + <maven.compiler.target>17</maven.compiler.target> + <maven.compiler.release>17</maven.compiler.release> + </properties> + <modules> + <module>flink-iotdb-table-connector-flink2</module> + </modules> + </profile> + </profiles> +</project> diff --git a/connectors/pom.xml b/connectors/pom.xml index 19c4dee..0396da0 100644 --- a/connectors/pom.xml +++ b/connectors/pom.xml @@ -68,6 +68,7 @@ <id>with-all-connectors</id> <modules> <module>flink-iotdb-connector</module> + <module>flink-iotdb-table-connector</module> <module>flink-sql-iotdb-connector</module> <module>flink-tsfile-connector</module> <module>hadoop</module> @@ -101,6 +102,7 @@ <id>with-flink</id> <modules> <module>flink-iotdb-connector</module> + <module>flink-iotdb-table-connector</module> <module>flink-sql-iotdb-connector</module> <module>flink-tsfile-connector</module> </modules> diff --git a/pom.xml b/pom.xml index b155418..df51590 100644 --- a/pom.xml +++ b/pom.xml @@ -184,7 +184,7 @@ <thrift.version>0.14.1</thrift.version> <!-- This was the last version to support Java 8 --> <tomcat.version>9.0.86</tomcat.version> - <tsfile.version>2.1.1</tsfile.version> + <tsfile.version>2.4.0</tsfile.version> <xz.version>1.9</xz.version> <zeppelin.version>0.11.1</zeppelin.version> <zstd-jni.version>1.5.5-5</zstd-jni.version>
