This is an automated email from the ASF dual-hosted git repository. JackieTien97 pushed a commit to branch ty/federated-query-plugin in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 7d12ec967c15b2f84f732c342b1e33468d469740 Author: JackieTien97 <[email protected]> AuthorDate: Fri Sep 18 16:06:45 2026 +0800 Add federated query plugin with GaussDB and openGauss connectors Migrate the JDBC table functions from ty/mysql-connector to the current master table-function API, keeping all changes inside library-udf. Add native GaussDB and openGauss driver adapters, preserve JDBC column aliases and null values, and close resources on execution and cleanup failures. Document deployment, authentication compatibility and limits. Validate both locale compiles, six unit tests in each locale, and fourteen end-to-end checks against openGauss, including local/remote joins and connection cleanup after early query termination. --- library-udf/FEDERATED_QUERY.md | 164 ++++++++++ library-udf/pom.xml | 54 +++- .../iotdb/library/i18n/LibraryUdfMessages.java | 29 +- .../iotdb/library/i18n/LibraryUdfMessages.java | 29 +- .../connector/BaseJDBCConnectorTableFunction.java | 334 +++++++++++++++++++++ .../connector/CassandraConnectorTableFunction.java | 22 ++ .../ClickhouseConnectorTableFunction.java | 96 ++++++ .../connector/DorisConnectorTableFunction.java | 22 ++ .../connector/GaussDBConnectorTableFunction.java | 47 +++ .../connector/JDBCConnectionPool.java | 163 ++++++++++ .../connector/MongoDBConnectorTableFunction.java | 22 ++ .../connector/MySqlConnectorTableFunction.java | 47 +++ .../connector/OpenGaussConnectorTableFunction.java | 47 +++ .../PostgreSqlConnectorTableFunction.java | 47 +++ .../connector/RedisConnectorTableFunction.java | 22 ++ .../connector/SnowflakeConnectorTableFunction.java | 22 ++ .../connector/converter/BinaryConverter.java | 39 +++ .../connector/converter/BlobConverter.java | 45 +++ .../connector/converter/BooleanConverter.java | 38 +++ .../connector/converter/DateConverter.java | 40 +++ .../connector/converter/DoubleConverter.java | 38 +++ .../connector/converter/FloatConverter.java | 38 +++ .../connector/converter/Int32Converter.java | 38 +++ .../connector/converter/Int64Converter.java | 38 +++ .../connector/converter/ResultSetConverter.java | 30 ++ .../connector/converter/StringConverter.java | 41 +++ .../connector/converter/TimeConverter.java | 38 +++ .../connector/converter/TimestampConverter.java | 38 +++ .../exception/CloseFailedInExternalDB.java | 32 ++ .../exception/ExecutionFailedInExternalDB.java | 32 ++ .../connector/JDBCConnectorTableFunctionTest.java | 325 ++++++++++++++++++++ 31 files changed, 2014 insertions(+), 3 deletions(-) diff --git a/library-udf/FEDERATED_QUERY.md b/library-udf/FEDERATED_QUERY.md new file mode 100644 index 00000000000..69ab199cf38 --- /dev/null +++ b/library-udf/FEDERATED_QUERY.md @@ -0,0 +1,164 @@ +<!-- + + 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. + +--> + +# Federated query table functions + +This plugin exposes remote JDBC query results as table-model relations in IoTDB. +It was migrated from `ty/mysql-connector` at `b7dc14dd490289add5fa2a92f09082a8e5339a99` +onto master at `87a6d198a901f46630a3455f59b264fc9368ca08`. +The plugin includes its own external-database exception classes and requires no +changes to the server or UDF API. The current master already provides the table-function lifecycle callbacks and UDF +error handling required by the original branch. Those existing implementations are +used without restoring the obsolete operators or duplicate status codes. The old +ConfigNode Ratis buffer-size adjustment is unrelated to this plugin and is not included. + +## Connectors and drivers + +All class names below have the prefix +`org.apache.iotdb.library.relational.tablefunction.connector.`. + +| Class | JDBC driver | URL prefix | +| --- | --- | --- | +| `MySqlConnectorTableFunction` | `com.mysql.cj.jdbc.Driver` | `jdbc:mysql:` | +| `PostgreSqlConnectorTableFunction` | `org.postgresql.Driver` | `jdbc:postgresql:` | +| `ClickhouseConnectorTableFunction` | `com.clickhouse.jdbc.ClickHouseDriver` | `jdbc:ch:` | +| `OpenGaussConnectorTableFunction` | `org.opengauss.Driver` | `jdbc:opengauss:` | +| `GaussDBConnectorTableFunction` | `com.huawei.gaussdb.jdbc.Driver` | `jdbc:gaussdb:` | + +Cassandra, Doris, MongoDB, Redis, and Snowflake classes inherited from the old +branch are empty placeholders, not usable connectors. + +GaussDB and openGauss have separate adapters. Huawei explicitly advises against +using the PostgreSQL JDBC driver for GaussDB, even when a particular version can +connect. See the [GaussDB driver documentation](https://support.huaweicloud.com/distributed-devg-v2-gaussdb/gaussdb-12-0056.html). + +openGauss supports PostgreSQL-compatible authentication when the server and user +are configured for MD5. Its normal SHA-256 authentication requires an openGauss +compatible driver. The native adapter uses `org.opengauss:opengauss-jdbc:6.0.3-og` +(the `-og` variant has the `org.opengauss` namespace), avoiding collisions with +PostgreSQL. See the [openGauss Java guide](https://docs.opengauss.org/en/docs/latest/getting_started/java.html). +An openGauss test does not establish compatibility with every commercial GaussDB +release. Use the GaussDB driver version supplied for the target deployment. + +## Build and installation + +Build from the repository root: + +```sh +mvn clean package -pl library-udf -am -DskipTests +``` + +Copy the resulting `library-udf-<version>.jar` and the driver JARs needed by your +connectors into `ext/udf/` on every ConfigNode and DataNode, before registering +functions. JDBC dependencies are `provided` and `optional`: they are not bundled +into the plugin or propagated into IoTDB distributions. + +Driver versions declared for compilation/testing are MySQL 9.3.0, PostgreSQL +42.7.7, ClickHouse 0.8.2 (`shaded-all`), openGauss 6.0.3-og, and GaussDB +506.0.0.b058-jdk7. Install the appropriate vendor driver and its runtime dependencies +for the database being queried. Do not install GaussDB's `gsjdbc4.jar` or the +PostgreSQL-namespaced openGauss driver alongside PostgreSQL JDBC; these share class +names. The adapters in this plugin deliberately use distinct vendor namespaces. + +In a table-model CLI session: + +```sql +CREATE FUNCTION query_opengauss AS + 'org.apache.iotdb.library.relational.tablefunction.connector.OpenGaussConnectorTableFunction'; +CREATE FUNCTION query_gaussdb AS + 'org.apache.iotdb.library.relational.tablefunction.connector.GaussDBConnectorTableFunction'; +CREATE FUNCTION query_pg AS + 'org.apache.iotdb.library.relational.tablefunction.connector.PostgreSqlConnectorTableFunction'; + +SELECT * FROM query_opengauss( + SQL => 'SELECT id, name FROM public.devices ORDER BY id', + URL => 'jdbc:opengauss://127.0.0.1:5432/postgres', + USERNAME => 'reader', "PASSWORD" => '<password>'); + +-- The remote result can be joined with a local IoTDB table. +SELECT l.time, l.device_id, r.name, l.temperature +FROM measurements l +JOIN query_opengauss( + SQL => 'SELECT id, name FROM public.devices', + URL => 'jdbc:opengauss://127.0.0.1:5432/postgres', + USERNAME => 'reader', "PASSWORD" => '<password>') r +ON l.device_id = r.id; +``` + +`PASSWORD` is a reserved SQL keyword, so quote its name as `"PASSWORD"` when +using named arguments (or pass all four arguments positionally). + +The `SQL` argument is executed in the external database's dialect. Use a dedicated +read-only database account. Column aliases are preserved. SQL, URL, USERNAME, and +PASSWORD are scalar arguments; URL/user/password defaults depend on the connector. +Passwords default to the empty string. Credentials are passed in the query and +serialized execution handle, so apply the deployment's existing query-log and +transport protections. + +## Type mapping and current limits + +Integer types map to INT32/INT64, floating-point and DECIMAL/NUMERIC to FLOAT/DOUBLE, +character types to STRING, BOOLEAN/BIT to BOOLEAN, DATE to DATE, TIME/TIMESTAMP +(including JDBC TIMESTAMP_WITH_TIMEZONE) to TIMESTAMP, and binary/BLOB to BLOB. +NULL values remain NULL. Unsupported JDBC types produce a UDF type error; cast +arrays, JSON, UUID, or vendor-specific objects in the remote SQL when necessary. +DECIMAL/NUMERIC conversion to DOUBLE can lose precision. Temporal conversion uses +JDBC epoch milliseconds; the deployment must use IoTDB's default millisecond +precision. The default JDBC timezone follows the DataNode JVM, so configure the +same timezone on all nodes when reading timestamps without a timezone. + +The analyzer opens a connection for result metadata; execution opens another +connection. The remote query runs as one table-function source, without automatic +predicate/join pushdown or parallel split discovery. Put remote filters in `SQL`. +Rows are emitted in bounded TsBlocks, but a JDBC driver may buffer the complete +remote result unless its URL/configuration enables streaming. ClickHouse's +metadata fallback executes the original SQL with `setMaxRows(1)` rather than +rewriting its LIMIT clause. + +## Unit verification + +```sh +mvn clean test -pl library-udf -am \ + -Dtest=JDBCConnectorTableFunctionTest \ + -Dsurefire.failIfNoSpecifiedTests=false -DfailIfNoTests=false +``` + +These tests cover alias metadata and serialized handles, batching, resource +cleanup after query/close failures, argument validation, and vendor driver +namespace selection. End-to-end verification additionally requires real IoTDB +and external database instances. + +## Verified configuration + +On 2026-09-18, IoTDB 2.0.11-SNAPSHOT based on master `87a6d198a9` was tested +against a real openGauss 5.0.0 ARM64 instance, using a UTF-8 database in PG +compatibility mode. All 14 end-to-end assertions passed: native SHA-256 login, +expected rejection of that login by stock PostgreSQL JDBC, native and MD5 +compatible queries, aliases, NULL values, numeric/date/timestamp/binary values, +5001 rows across batches, local/remote JOIN, and recovery after SQL/authentication +failures. The GaussDB adapter with its Huawei driver also queried this openGauss +instance successfully. Twenty additional `LIMIT 1` queries left no sessions for +either external test user in `pg_stat_activity`. + +Both full-reactor locale compiles and the six connector unit tests passed. +A commercial GaussDB server, and live MySQL/ClickHouse servers, were not part of +this verification. For commercial GaussDB, select the official driver matching +the deployment and verify against that actual server before rollout. diff --git a/library-udf/pom.xml b/library-udf/pom.xml index ef592895cc3..90ac8c261be 100644 --- a/library-udf/pom.xml +++ b/library-udf/pom.xml @@ -35,8 +35,45 @@ <dependencies> <dependency> <groupId>org.apache.tsfile</groupId> - <artifactId>common</artifactId> + <artifactId>tsfile</artifactId> <version>${tsfile.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>com.mysql</groupId> + <artifactId>mysql-connector-j</artifactId> + <version>9.3.0</version> + <scope>provided</scope> + <optional>true</optional> + </dependency> + <dependency> + <groupId>org.postgresql</groupId> + <artifactId>postgresql</artifactId> + <version>42.7.7</version> + <scope>provided</scope> + <optional>true</optional> + </dependency> + <dependency> + <groupId>com.clickhouse</groupId> + <artifactId>clickhouse-jdbc</artifactId> + <version>0.8.2</version> + <scope>provided</scope> + <optional>true</optional> + <classifier>shaded-all</classifier> + </dependency> + <dependency> + <groupId>org.opengauss</groupId> + <artifactId>opengauss-jdbc</artifactId> + <version>6.0.3-og</version> + <scope>provided</scope> + <optional>true</optional> + </dependency> + <dependency> + <groupId>com.huaweicloud.gaussdb</groupId> + <artifactId>gaussdbjdbc</artifactId> + <version>506.0.0.b058-jdk7</version> + <scope>provided</scope> + <optional>true</optional> </dependency> <dependency> <groupId>org.apache.iotdb</groupId> @@ -82,6 +119,21 @@ </dependencies> <build> <plugins> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-dependency-plugin</artifactId> + <configuration> + <!-- JDBC drivers are loaded by class name through the UDF class loader. --> + <usedDependencies> + <usedDependency>com.mysql:mysql-connector-j</usedDependency> + <usedDependency>org.postgresql:postgresql</usedDependency> + <usedDependency>org.opengauss:opengauss-jdbc</usedDependency> + <usedDependency>com.huaweicloud.gaussdb:gaussdbjdbc</usedDependency> + <!-- The shaded ClickHouse driver also contains SLF4J API classes. --> + <usedDependency>org.slf4j:slf4j-api</usedDependency> + </usedDependencies> + </configuration> + </plugin> <plugin> <groupId>org.codehaus.mojo</groupId> <artifactId>build-helper-maven-plugin</artifactId> diff --git a/library-udf/src/main/i18n/en/org/apache/iotdb/library/i18n/LibraryUdfMessages.java b/library-udf/src/main/i18n/en/org/apache/iotdb/library/i18n/LibraryUdfMessages.java index 3b77edc96af..0a1b309c986 100644 --- a/library-udf/src/main/i18n/en/org/apache/iotdb/library/i18n/LibraryUdfMessages.java +++ b/library-udf/src/main/i18n/en/org/apache/iotdb/library/i18n/LibraryUdfMessages.java @@ -21,6 +21,32 @@ package org.apache.iotdb.library.i18n; public final class LibraryUdfMessages { + public static final String EXCEPTION_QUERY_EXECUTION_FAILED_IN_ARG_E4F52371 = + "Query execution failed in %s."; + + public static final String EXCEPTION_CLOSING_JDBC_RESOURCES_FAILED_IN_ARG_A300EDC2 = + "Closing JDBC resources failed in %s."; + public static final String EXCEPTION_SQL_AND_URL_MUST_BE_NON_EMPTY_STRINGS_USERNAME_AND_PASSWORD_MUST_NOT_BE_NULL_A253F77B = + "SQL and URL must be non-empty strings; USERNAME and PASSWORD must not be null."; + + public static final String EXCEPTION_SQL_MUST_BE_A_QUERY_THAT_RETURNS_AT_LEAST_ONE_COLUMN_1AF6E91B = + "SQL must be a query that returns at least one column."; + + public static final String EXCEPTION_JDBC_URL_IS_NOT_ACCEPTED_BY_DRIVER_ARG_78687D88 = + "JDBC URL is not accepted by driver %s."; + + public static final String EXCEPTION_FAILED_TO_LOAD_JDBC_DRIVER_ARG_INSTALL_ITS_JAR_IN_THE_UDF_LIBRARY_DIRECTORY_8B1BB953 = + "Failed to load JDBC driver %s. Install its JAR in the UDF library directory."; + + public static final String EXCEPTION_UNSUPPORTED_JDBC_TYPE_ARG_D8792616 = + "Unsupported JDBC type: %s"; + + public static final String EXCEPTION_FAILED_TO_READ_JDBC_RESULT_METADATA_ARG_16D4E50C = + "Failed to read JDBC result metadata: %s"; + + public static final String EXCEPTION_FAILED_TO_SERIALIZE_JDBC_TABLE_FUNCTION_HANDLE_ARG_1023C78A = + "Failed to serialize JDBC table function handle: %s"; + // UDTFRange, UDTFTwoSidedFilter public static final String NO_SUCH_DATA_TYPE = "No such kind of data type."; @@ -144,7 +170,8 @@ public final class LibraryUdfMessages { "Illegal parameter, valuePattern must be double,double..."; // OnePassBucketizer - public static final String EXCEPTION_CAN_T_PRODUCE_1EF5D1BB = "Can't produce "; + public static final String EXCEPTION_CAN_T_PRODUCE_1EF5D1BB = + "Can't produce "; public static final String EXCEPTION_CAN_T_PRODUCE_ARG_BUCKETS_FROM_AN_INPUT_SERIES_OF_ARG_ELEMENTS_D17D0135 = "Can't produce %d buckets from an input series of %d elements"; diff --git a/library-udf/src/main/i18n/zh/org/apache/iotdb/library/i18n/LibraryUdfMessages.java b/library-udf/src/main/i18n/zh/org/apache/iotdb/library/i18n/LibraryUdfMessages.java index d4fdafddb29..20b2d48de54 100644 --- a/library-udf/src/main/i18n/zh/org/apache/iotdb/library/i18n/LibraryUdfMessages.java +++ b/library-udf/src/main/i18n/zh/org/apache/iotdb/library/i18n/LibraryUdfMessages.java @@ -21,6 +21,32 @@ package org.apache.iotdb.library.i18n; public final class LibraryUdfMessages { + public static final String EXCEPTION_QUERY_EXECUTION_FAILED_IN_ARG_E4F52371 = + "在 %s 中执行查询失败。"; + + public static final String EXCEPTION_CLOSING_JDBC_RESOURCES_FAILED_IN_ARG_A300EDC2 = + "关闭 %s 的 JDBC 资源失败。"; + public static final String EXCEPTION_SQL_AND_URL_MUST_BE_NON_EMPTY_STRINGS_USERNAME_AND_PASSWORD_MUST_NOT_BE_NULL_A253F77B = + "SQL 和 URL 必须为非空字符串；USERNAME 和 PASSWORD 不能为 null。"; + + public static final String EXCEPTION_SQL_MUST_BE_A_QUERY_THAT_RETURNS_AT_LEAST_ONE_COLUMN_1AF6E91B = + "SQL 必须是返回至少一列的查询语句。"; + + public static final String EXCEPTION_JDBC_URL_IS_NOT_ACCEPTED_BY_DRIVER_ARG_78687D88 = + "JDBC URL 不被驱动 %s 接受。"; + + public static final String EXCEPTION_FAILED_TO_LOAD_JDBC_DRIVER_ARG_INSTALL_ITS_JAR_IN_THE_UDF_LIBRARY_DIRECTORY_8B1BB953 = + "加载 JDBC 驱动 %s 失败，请将驱动 JAR 安装到 UDF 库目录。"; + + public static final String EXCEPTION_UNSUPPORTED_JDBC_TYPE_ARG_D8792616 = + "不支持的 JDBC 类型：%s"; + + public static final String EXCEPTION_FAILED_TO_READ_JDBC_RESULT_METADATA_ARG_16D4E50C = + "读取 JDBC 结果元数据失败：%s"; + + public static final String EXCEPTION_FAILED_TO_SERIALIZE_JDBC_TABLE_FUNCTION_HANDLE_ARG_1023C78A = + "序列化 JDBC 表函数句柄失败：%s"; + // UDTFRange, UDTFTwoSidedFilter public static final String NO_SUCH_DATA_TYPE = "不支持此数据类型。"; @@ -133,7 +159,8 @@ public final class LibraryUdfMessages { "非法参数，valuePattern 必须为 double,double..."; // OnePassBucketizer - public static final String EXCEPTION_CAN_T_PRODUCE_1EF5D1BB = "无法生成 "; + public static final String EXCEPTION_CAN_T_PRODUCE_1EF5D1BB = + "无法生成 "; public static final String EXCEPTION_CAN_T_PRODUCE_ARG_BUCKETS_FROM_AN_INPUT_SERIES_OF_ARG_ELEMENTS_D17D0135 = "无法生成 %d 个 bucket，输入序列包含 %d 个元素"; diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/BaseJDBCConnectorTableFunction.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/BaseJDBCConnectorTableFunction.java new file mode 100644 index 00000000000..fde6b279e8f --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/BaseJDBCConnectorTableFunction.java @@ -0,0 +1,334 @@ +/* + * 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.library.relational.tablefunction.connector; + +import org.apache.iotdb.library.i18n.LibraryUdfMessages; +import org.apache.iotdb.library.relational.tablefunction.connector.converter.ResultSetConverter; +import org.apache.iotdb.library.relational.tablefunction.connector.exception.CloseFailedInExternalDB; +import org.apache.iotdb.library.relational.tablefunction.connector.exception.ExecutionFailedInExternalDB; +import org.apache.iotdb.udf.api.exception.UDFException; +import org.apache.iotdb.udf.api.relational.TableFunction; +import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis; +import org.apache.iotdb.udf.api.relational.table.TableFunctionHandle; +import org.apache.iotdb.udf.api.relational.table.TableFunctionProcessorProvider; +import org.apache.iotdb.udf.api.relational.table.argument.Argument; +import org.apache.iotdb.udf.api.relational.table.argument.DescribedSchema; +import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument; +import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionLeafProcessor; +import org.apache.iotdb.udf.api.relational.table.specification.ParameterSpecification; +import org.apache.iotdb.udf.api.relational.table.specification.ScalarParameterSpecification; +import org.apache.iotdb.udf.api.type.Type; + +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.common.conf.TSFileDescriptor; +import org.apache.tsfile.utils.PublicBAOS; +import org.apache.tsfile.utils.ReadWriteIOUtils; + +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; + +import static org.apache.iotdb.library.relational.tablefunction.connector.JDBCConnectionPool.translateJDBCTypeToUDFType; + +abstract class BaseJDBCConnectorTableFunction implements TableFunction { + + static class BaseJDBCConnectorTableFunctionHandle implements TableFunctionHandle { + String sql; + String url; + String userName; + String password; + int[] types; + + public BaseJDBCConnectorTableFunctionHandle() {} + + public BaseJDBCConnectorTableFunctionHandle( + String sql, String url, String userName, String password, int[] types) { + this.sql = sql; + this.url = url; + this.userName = userName; + this.password = password; + this.types = types; + } + + List<ResultSetConverter> getConverters() { + List<ResultSetConverter> converters = new ArrayList<>(types.length); + for (int type : types) { + converters.add(JDBCConnectionPool.getResultSetConverter(type)); + } + return converters; + } + + @Override + public byte[] serialize() { + try (PublicBAOS publicBAOS = new PublicBAOS(); + DataOutputStream outputStream = new DataOutputStream(publicBAOS)) { + ReadWriteIOUtils.write(sql, outputStream); + ReadWriteIOUtils.write(url, outputStream); + ReadWriteIOUtils.write(userName, outputStream); + ReadWriteIOUtils.write(password, outputStream); + ReadWriteIOUtils.write(types.length, outputStream); + for (int type : types) { + ReadWriteIOUtils.write(type, outputStream); + } + outputStream.flush(); + return publicBAOS.toByteArray(); + } catch (IOException e) { + throw new UDFException( + String.format( + LibraryUdfMessages + .EXCEPTION_FAILED_TO_SERIALIZE_JDBC_TABLE_FUNCTION_HANDLE_ARG_1023C78A, + e.getMessage())); + } + } + + @Override + public void deserialize(byte[] bytes) { + ByteBuffer buffer = ByteBuffer.wrap(bytes); + this.sql = ReadWriteIOUtils.readString(buffer); + this.url = ReadWriteIOUtils.readString(buffer); + this.userName = ReadWriteIOUtils.readString(buffer); + this.password = ReadWriteIOUtils.readString(buffer); + this.types = new int[ReadWriteIOUtils.readInt(buffer)]; + for (int i = 0; i < types.length; i++) { + types[i] = ReadWriteIOUtils.readInt(buffer); + } + } + } + + private static final String SQL = "SQL"; + private static final String URL = "URL"; + private static final String USERNAME = "USERNAME"; + private static final String PASSWORD = "PASSWORD"; + + @Override + public List<ParameterSpecification> getArgumentsSpecifications() { + return Arrays.asList( + ScalarParameterSpecification.builder().name(SQL).type(Type.STRING).build(), + ScalarParameterSpecification.builder() + .name(URL) + .type(Type.STRING) + .defaultValue(getDefaultUrl()) + .build(), + ScalarParameterSpecification.builder() + .name(USERNAME) + .type(Type.STRING) + .defaultValue(getDefaultUser()) + .build(), + ScalarParameterSpecification.builder() + .name(PASSWORD) + .type(Type.STRING) + .defaultValue(getDefaultPassword()) + .build()); + } + + abstract String getDefaultUrl(); + + abstract String getDefaultUser(); + + abstract String getDefaultPassword(); + + abstract String getDriverClassName(); + + abstract String getDBName(); + + @Override + public TableFunctionAnalysis analyze(Map<String, Argument> arguments) throws UDFException { + + String sql = (String) ((ScalarArgument) arguments.get(SQL)).getValue(); + String url = (String) ((ScalarArgument) arguments.get(URL)).getValue(); + String userName = (String) ((ScalarArgument) arguments.get(USERNAME)).getValue(); + String password = (String) ((ScalarArgument) arguments.get(PASSWORD)).getValue(); + + if (sql == null + || sql.trim().isEmpty() + || url == null + || url.trim().isEmpty() + || userName == null + || password == null) { + throw new UDFException( + LibraryUdfMessages + .EXCEPTION_SQL_AND_URL_MUST_BE_NON_EMPTY_STRINGS_USERNAME_AND_PASSWORD_MUST_NOT_BE_NULL_A253F77B); + } + DescribedSchema.Builder schemaBuilder = DescribedSchema.builder(); + int[] types = buildResultHeaders(schemaBuilder, sql, url, userName, password); + BaseJDBCConnectorTableFunctionHandle handle = + new BaseJDBCConnectorTableFunctionHandle(sql, url, userName, password, types); + return TableFunctionAnalysis.builder() + .properColumnSchema(schemaBuilder.build()) + .handle(handle) + .build(); + } + + int[] buildResultHeaders( + DescribedSchema.Builder schemaBuilder, + String sql, + String url, + String userName, + String password) { + int[] types; + try (Connection connection = + JDBCConnectionPool.getConnection(getDriverClassName(), url, userName, password); + PreparedStatement statement = connection.prepareStatement(sql)) { + ResultSetMetaData metaData = statement.getMetaData(); + if (metaData == null || metaData.getColumnCount() == 0) { + throw new UDFException( + LibraryUdfMessages + .EXCEPTION_SQL_MUST_BE_A_QUERY_THAT_RETURNS_AT_LEAST_ONE_COLUMN_1AF6E91B); + } + types = new int[metaData.getColumnCount()]; + for (int i = 1, size = metaData.getColumnCount(); i <= size; i++) { + int type = metaData.getColumnType(i); + schemaBuilder.addField(metaData.getColumnLabel(i), translateJDBCTypeToUDFType(type)); + types[i - 1] = type; + } + return types; + } catch (SQLException e) { + throw new UDFException( + String.format( + LibraryUdfMessages.EXCEPTION_FAILED_TO_READ_JDBC_RESULT_METADATA_ARG_16D4E50C, + e.getMessage()), + e); + } + } + + @Override + public TableFunctionHandle createTableFunctionHandle() { + return new BaseJDBCConnectorTableFunctionHandle(); + } + + @Override + public TableFunctionProcessorProvider getProcessorProvider( + TableFunctionHandle tableFunctionHandle) { + return new TableFunctionProcessorProvider() { + @Override + public TableFunctionLeafProcessor getSplitProcessor() { + return getProcessor((BaseJDBCConnectorTableFunctionHandle) tableFunctionHandle); + } + }; + } + + JDBCProcessor getProcessor(BaseJDBCConnectorTableFunctionHandle tableFunctionHandle) { + return new JDBCProcessor(tableFunctionHandle, getDriverClassName(), getDBName()); + } + + static class JDBCProcessor implements TableFunctionLeafProcessor { + private static final int MAX_TSBLOCK_SIZE_IN_BYTES = + TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes(); + private static final int MAX_TSBLOCK_LINE_NUMBER = + TSFileDescriptor.getInstance().getConfig().getMaxTsBlockLineNumber(); + private final BaseJDBCConnectorTableFunctionHandle handle; + private final List<ResultSetConverter> converters; + private final String driverClassName; + private final String dbName; + private Connection connection; + private Statement statement; + private ResultSet resultSet; + private boolean finished = false; + + JDBCProcessor( + BaseJDBCConnectorTableFunctionHandle handle, String driverClassName, String dbName) { + this.driverClassName = driverClassName; + this.dbName = dbName; + this.handle = handle; + this.converters = handle.getConverters(); + } + + @Override + public void beforeStart() { + try { + this.connection = + JDBCConnectionPool.getConnection( + driverClassName, handle.url, handle.userName, handle.password); + this.statement = connection.createStatement(); + this.resultSet = statement.executeQuery(handle.sql); + } catch (SQLException | RuntimeException e) { + try { + beforeDestroy(); + } catch (RuntimeException closeError) { + e.addSuppressed(closeError); + } + throw new ExecutionFailedInExternalDB(dbName, e); + } + } + + @Override + public void process(List<ColumnBuilder> columnBuilders) { + finished = true; + try { + int count = 0; + while (resultSet.next()) { + long retainedSize = 0; + for (int i = 0, size = columnBuilders.size(); i < size; i++) { + converters.get(i).append(resultSet, i + 1, columnBuilders.get(i)); + retainedSize += columnBuilders.get(i).getRetainedSizeInBytes(); + } + count++; + if (retainedSize >= MAX_TSBLOCK_SIZE_IN_BYTES || count >= MAX_TSBLOCK_LINE_NUMBER) { + finished = false; + break; + } + } + } catch (SQLException e) { + throw new ExecutionFailedInExternalDB(dbName, e); + } + } + + @Override + public boolean isFinish() { + return finished; + } + + @Override + public void beforeDestroy() { + finished = true; + SQLException failure = null; + // Attempt all closes even if the result set or statement fails to close. + for (AutoCloseable resource : new AutoCloseable[] {resultSet, statement, connection}) { + if (resource != null) { + try { + resource.close(); + } catch (Exception e) { + if (failure == null) { + failure = new SQLException(e); + } else { + failure.addSuppressed(e); + } + } + } + } + resultSet = null; + statement = null; + connection = null; + if (failure != null) { + throw new CloseFailedInExternalDB(dbName, failure); + } + } + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/CassandraConnectorTableFunction.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/CassandraConnectorTableFunction.java new file mode 100644 index 00000000000..b8a23163458 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/CassandraConnectorTableFunction.java @@ -0,0 +1,22 @@ +/* + * 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.library.relational.tablefunction.connector; + +public class CassandraConnectorTableFunction {} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/ClickhouseConnectorTableFunction.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/ClickhouseConnectorTableFunction.java new file mode 100644 index 00000000000..2f0ea240bf2 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/ClickhouseConnectorTableFunction.java @@ -0,0 +1,96 @@ +/* + * 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.library.relational.tablefunction.connector; + +import org.apache.iotdb.library.i18n.LibraryUdfMessages; +import org.apache.iotdb.udf.api.exception.UDFException; +import org.apache.iotdb.udf.api.relational.table.argument.DescribedSchema; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.sql.Statement; + +import static org.apache.iotdb.library.relational.tablefunction.connector.JDBCConnectionPool.translateJDBCTypeToUDFType; + +public class ClickhouseConnectorTableFunction extends BaseJDBCConnectorTableFunction { + + private static final String DEFAULT_URL = "jdbc:ch://localhost:8123"; + private static final String DEFAULT_USERNAME = "default"; + private static final String DEFAULT_PASSWORD = ""; + private static final String CLICKHOUSE = "CLICKHOUSE"; + + @Override + String getDefaultUrl() { + return DEFAULT_URL; + } + + @Override + String getDefaultUser() { + return DEFAULT_USERNAME; + } + + @Override + String getDefaultPassword() { + return DEFAULT_PASSWORD; + } + + @Override + int[] buildResultHeaders( + DescribedSchema.Builder schemaBuilder, + String sql, + String url, + String userName, + String password) { + int[] types; + try (Connection connection = + JDBCConnectionPool.getConnection(getDriverClassName(), url, userName, password); + Statement statement = connection.createStatement()) { + statement.setMaxRows(1); + try (ResultSet resultSet = statement.executeQuery(sql)) { + ResultSetMetaData metaData = resultSet.getMetaData(); + types = new int[metaData.getColumnCount()]; + for (int i = 1, size = metaData.getColumnCount(); i <= size; i++) { + int type = metaData.getColumnType(i); + schemaBuilder.addField(metaData.getColumnLabel(i), translateJDBCTypeToUDFType(type)); + types[i - 1] = type; + } + return types; + } + } catch (SQLException e) { + throw new UDFException( + String.format( + LibraryUdfMessages.EXCEPTION_FAILED_TO_READ_JDBC_RESULT_METADATA_ARG_16D4E50C, + e.getMessage()), + e); + } + } + + @Override + String getDriverClassName() { + return "com.clickhouse.jdbc.ClickHouseDriver"; + } + + @Override + String getDBName() { + return CLICKHOUSE; + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/DorisConnectorTableFunction.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/DorisConnectorTableFunction.java new file mode 100644 index 00000000000..2cb68caf2dc --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/DorisConnectorTableFunction.java @@ -0,0 +1,22 @@ +/* + * 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.library.relational.tablefunction.connector; + +public class DorisConnectorTableFunction {} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/GaussDBConnectorTableFunction.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/GaussDBConnectorTableFunction.java new file mode 100644 index 00000000000..adccba555b6 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/GaussDBConnectorTableFunction.java @@ -0,0 +1,47 @@ +/* + * 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.library.relational.tablefunction.connector; + +public class GaussDBConnectorTableFunction extends BaseJDBCConnectorTableFunction { + @Override + String getDefaultUrl() { + return "jdbc:gaussdb://localhost:5432/postgres"; + } + + @Override + String getDefaultUser() { + return "gaussdb"; + } + + @Override + String getDefaultPassword() { + return ""; + } + + @Override + String getDriverClassName() { + return "com.huawei.gaussdb.jdbc.Driver"; + } + + @Override + String getDBName() { + return "GaussDB"; + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/JDBCConnectionPool.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/JDBCConnectionPool.java new file mode 100644 index 00000000000..f7a4df4ac67 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/JDBCConnectionPool.java @@ -0,0 +1,163 @@ +/* + * 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.library.relational.tablefunction.connector; + +import org.apache.iotdb.library.i18n.LibraryUdfMessages; +import org.apache.iotdb.library.relational.tablefunction.connector.converter.BinaryConverter; +import org.apache.iotdb.library.relational.tablefunction.connector.converter.BlobConverter; +import org.apache.iotdb.library.relational.tablefunction.connector.converter.BooleanConverter; +import org.apache.iotdb.library.relational.tablefunction.connector.converter.DateConverter; +import org.apache.iotdb.library.relational.tablefunction.connector.converter.DoubleConverter; +import org.apache.iotdb.library.relational.tablefunction.connector.converter.FloatConverter; +import org.apache.iotdb.library.relational.tablefunction.connector.converter.Int32Converter; +import org.apache.iotdb.library.relational.tablefunction.connector.converter.Int64Converter; +import org.apache.iotdb.library.relational.tablefunction.connector.converter.ResultSetConverter; +import org.apache.iotdb.library.relational.tablefunction.connector.converter.StringConverter; +import org.apache.iotdb.library.relational.tablefunction.connector.converter.TimeConverter; +import org.apache.iotdb.library.relational.tablefunction.connector.converter.TimestampConverter; +import org.apache.iotdb.udf.api.exception.UDFException; +import org.apache.iotdb.udf.api.exception.UDFTypeMismatchException; +import org.apache.iotdb.udf.api.type.Type; + +import java.sql.Connection; +import java.sql.Driver; +import java.sql.SQLException; +import java.util.Properties; + +public class JDBCConnectionPool { + + private JDBCConnectionPool() {} + + public static Connection getConnection( + String driverClassName, String url, String userName, String password) throws SQLException { + final Driver driver; + try { + driver = + (Driver) + Class.forName(driverClassName, true, JDBCConnectionPool.class.getClassLoader()) + .getDeclaredConstructor() + .newInstance(); + } catch (ReflectiveOperationException | ClassCastException | LinkageError e) { + throw new UDFException( + String.format( + LibraryUdfMessages + .EXCEPTION_FAILED_TO_LOAD_JDBC_DRIVER_ARG_INSTALL_ITS_JAR_IN_THE_UDF_LIBRARY_DIRECTORY_8B1BB953, + driverClassName), + e); + } + Properties properties = new Properties(); + properties.setProperty("user", userName); + properties.setProperty("password", password); + Connection connection = driver.connect(url, properties); + if (connection == null) { + throw new UDFException( + String.format( + LibraryUdfMessages.EXCEPTION_JDBC_URL_IS_NOT_ACCEPTED_BY_DRIVER_ARG_78687D88, + driverClassName)); + } + return connection; + } + + public static Type translateJDBCTypeToUDFType(int type) { + switch (type) { + case java.sql.Types.TINYINT: + case java.sql.Types.SMALLINT: + case java.sql.Types.INTEGER: + return Type.INT32; + case java.sql.Types.BIGINT: + return Type.INT64; + case java.sql.Types.FLOAT: + return Type.FLOAT; + case java.sql.Types.DOUBLE: + case java.sql.Types.REAL: + case java.sql.Types.NUMERIC: + case java.sql.Types.DECIMAL: + return Type.DOUBLE; + case java.sql.Types.CHAR: + case java.sql.Types.VARCHAR: + case java.sql.Types.LONGVARCHAR: + case java.sql.Types.NCHAR: + case java.sql.Types.NVARCHAR: + case java.sql.Types.LONGNVARCHAR: + return Type.STRING; + case java.sql.Types.DATE: + return Type.DATE; + case java.sql.Types.TIME: + case java.sql.Types.TIMESTAMP: + case java.sql.Types.TIMESTAMP_WITH_TIMEZONE: + return Type.TIMESTAMP; + case java.sql.Types.BINARY: + case java.sql.Types.VARBINARY: + case java.sql.Types.LONGVARBINARY: + case java.sql.Types.BLOB: + return Type.BLOB; + case java.sql.Types.BIT: + case java.sql.Types.BOOLEAN: + return Type.BOOLEAN; + default: + throw new UDFTypeMismatchException( + String.format(LibraryUdfMessages.EXCEPTION_UNSUPPORTED_JDBC_TYPE_ARG_D8792616, type)); + } + } + + public static ResultSetConverter getResultSetConverter(int type) { + switch (type) { + case java.sql.Types.TINYINT: + case java.sql.Types.SMALLINT: + case java.sql.Types.INTEGER: + return new Int32Converter(); + case java.sql.Types.BIGINT: + return new Int64Converter(); + case java.sql.Types.FLOAT: + return new FloatConverter(); + case java.sql.Types.DOUBLE: + case java.sql.Types.REAL: + case java.sql.Types.NUMERIC: + case java.sql.Types.DECIMAL: + return new DoubleConverter(); + case java.sql.Types.CHAR: + case java.sql.Types.VARCHAR: + case java.sql.Types.LONGVARCHAR: + case java.sql.Types.NCHAR: + case java.sql.Types.NVARCHAR: + case java.sql.Types.LONGNVARCHAR: + return new StringConverter(); + case java.sql.Types.DATE: + return new DateConverter(); + case java.sql.Types.TIME: + return new TimeConverter(); + case java.sql.Types.TIMESTAMP: + case java.sql.Types.TIMESTAMP_WITH_TIMEZONE: + return new TimestampConverter(); + case java.sql.Types.BINARY: + case java.sql.Types.VARBINARY: + case java.sql.Types.LONGVARBINARY: + return new BinaryConverter(); + case java.sql.Types.BLOB: + return new BlobConverter(); + case java.sql.Types.BIT: + case java.sql.Types.BOOLEAN: + return new BooleanConverter(); + default: + throw new UDFTypeMismatchException( + String.format(LibraryUdfMessages.EXCEPTION_UNSUPPORTED_JDBC_TYPE_ARG_D8792616, type)); + } + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/MongoDBConnectorTableFunction.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/MongoDBConnectorTableFunction.java new file mode 100644 index 00000000000..2ad4bfaa9b0 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/MongoDBConnectorTableFunction.java @@ -0,0 +1,22 @@ +/* + * 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.library.relational.tablefunction.connector; + +public class MongoDBConnectorTableFunction {} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/MySqlConnectorTableFunction.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/MySqlConnectorTableFunction.java new file mode 100644 index 00000000000..e7a625e0754 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/MySqlConnectorTableFunction.java @@ -0,0 +1,47 @@ +/* + * 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.library.relational.tablefunction.connector; + +public class MySqlConnectorTableFunction extends BaseJDBCConnectorTableFunction { + @Override + String getDefaultUrl() { + return "jdbc:mysql://localhost:3306/"; + } + + @Override + String getDefaultUser() { + return "root"; + } + + @Override + String getDefaultPassword() { + return ""; + } + + @Override + String getDriverClassName() { + return "com.mysql.cj.jdbc.Driver"; + } + + @Override + String getDBName() { + return "MySQL"; + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/OpenGaussConnectorTableFunction.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/OpenGaussConnectorTableFunction.java new file mode 100644 index 00000000000..4cdd4f2804c --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/OpenGaussConnectorTableFunction.java @@ -0,0 +1,47 @@ +/* + * 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.library.relational.tablefunction.connector; + +public class OpenGaussConnectorTableFunction extends BaseJDBCConnectorTableFunction { + @Override + String getDefaultUrl() { + return "jdbc:opengauss://localhost:5432/postgres"; + } + + @Override + String getDefaultUser() { + return "gaussdb"; + } + + @Override + String getDefaultPassword() { + return ""; + } + + @Override + String getDriverClassName() { + return "org.opengauss.Driver"; + } + + @Override + String getDBName() { + return "openGauss"; + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/PostgreSqlConnectorTableFunction.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/PostgreSqlConnectorTableFunction.java new file mode 100644 index 00000000000..fdc3928fa2a --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/PostgreSqlConnectorTableFunction.java @@ -0,0 +1,47 @@ +/* + * 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.library.relational.tablefunction.connector; + +public class PostgreSqlConnectorTableFunction extends BaseJDBCConnectorTableFunction { + @Override + String getDefaultUrl() { + return "jdbc:postgresql://localhost:5432/postgres"; + } + + @Override + String getDefaultUser() { + return "postgres"; + } + + @Override + String getDefaultPassword() { + return ""; + } + + @Override + String getDriverClassName() { + return "org.postgresql.Driver"; + } + + @Override + String getDBName() { + return "PostgreSQL"; + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/RedisConnectorTableFunction.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/RedisConnectorTableFunction.java new file mode 100644 index 00000000000..d8b9df06924 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/RedisConnectorTableFunction.java @@ -0,0 +1,22 @@ +/* + * 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.library.relational.tablefunction.connector; + +public class RedisConnectorTableFunction {} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/SnowflakeConnectorTableFunction.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/SnowflakeConnectorTableFunction.java new file mode 100644 index 00000000000..e9d0d972fc1 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/SnowflakeConnectorTableFunction.java @@ -0,0 +1,22 @@ +/* + * 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.library.relational.tablefunction.connector; + +public class SnowflakeConnectorTableFunction {} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/BinaryConverter.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/BinaryConverter.java new file mode 100644 index 00000000000..b1caf760aaa --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/BinaryConverter.java @@ -0,0 +1,39 @@ +/* + * 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.library.relational.tablefunction.connector.converter; + +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.utils.Binary; + +import java.sql.ResultSet; +import java.sql.SQLException; + +public class BinaryConverter implements ResultSetConverter { + @Override + public void append(ResultSet row, int columnIndex, ColumnBuilder properColumnBuilder) + throws SQLException { + byte[] value = row.getBytes(columnIndex); + if (value == null || row.wasNull()) { + properColumnBuilder.appendNull(); + } else { + properColumnBuilder.writeBinary(new Binary(value)); + } + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/BlobConverter.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/BlobConverter.java new file mode 100644 index 00000000000..557fa6b805d --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/BlobConverter.java @@ -0,0 +1,45 @@ +/* + * 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.library.relational.tablefunction.connector.converter; + +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.utils.Binary; + +import java.sql.Blob; +import java.sql.ResultSet; +import java.sql.SQLException; + +public class BlobConverter implements ResultSetConverter { + @Override + public void append(ResultSet row, int columnIndex, ColumnBuilder properColumnBuilder) + throws SQLException { + Blob blob = row.getBlob(columnIndex); + if (blob == null || row.wasNull()) { + properColumnBuilder.appendNull(); + } else { + try { + properColumnBuilder.writeBinary( + new Binary(blob.getBytes(1, Math.toIntExact(blob.length())))); + } finally { + blob.free(); + } + } + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/BooleanConverter.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/BooleanConverter.java new file mode 100644 index 00000000000..6a6a2739b7e --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/BooleanConverter.java @@ -0,0 +1,38 @@ +/* + * 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.library.relational.tablefunction.connector.converter; + +import org.apache.tsfile.block.column.ColumnBuilder; + +import java.sql.ResultSet; +import java.sql.SQLException; + +public class BooleanConverter implements ResultSetConverter { + @Override + public void append(ResultSet row, int columnIndex, ColumnBuilder properColumnBuilder) + throws SQLException { + boolean value = row.getBoolean(columnIndex); + if (row.wasNull()) { + properColumnBuilder.appendNull(); + } else { + properColumnBuilder.writeBoolean(value); + } + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/DateConverter.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/DateConverter.java new file mode 100644 index 00000000000..7ad9a879d4d --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/DateConverter.java @@ -0,0 +1,40 @@ +/* + * 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.library.relational.tablefunction.connector.converter; + +import org.apache.tsfile.block.column.ColumnBuilder; + +import java.sql.ResultSet; +import java.sql.SQLException; + +public class DateConverter implements ResultSetConverter { + @Override + public void append(ResultSet row, int columnIndex, ColumnBuilder properColumnBuilder) + throws SQLException { + java.sql.Date value = row.getDate(columnIndex); + if (value == null || row.wasNull()) { + properColumnBuilder.appendNull(); + } else { + java.time.LocalDate date = value.toLocalDate(); + properColumnBuilder.writeInt( + date.getYear() * 10000 + date.getMonthValue() * 100 + date.getDayOfMonth()); + } + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/DoubleConverter.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/DoubleConverter.java new file mode 100644 index 00000000000..fd9f849f488 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/DoubleConverter.java @@ -0,0 +1,38 @@ +/* + * 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.library.relational.tablefunction.connector.converter; + +import org.apache.tsfile.block.column.ColumnBuilder; + +import java.sql.ResultSet; +import java.sql.SQLException; + +public class DoubleConverter implements ResultSetConverter { + @Override + public void append(ResultSet row, int columnIndex, ColumnBuilder properColumnBuilder) + throws SQLException { + double value = row.getDouble(columnIndex); + if (row.wasNull()) { + properColumnBuilder.appendNull(); + } else { + properColumnBuilder.writeDouble(value); + } + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/FloatConverter.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/FloatConverter.java new file mode 100644 index 00000000000..72b5fdc2700 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/FloatConverter.java @@ -0,0 +1,38 @@ +/* + * 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.library.relational.tablefunction.connector.converter; + +import org.apache.tsfile.block.column.ColumnBuilder; + +import java.sql.ResultSet; +import java.sql.SQLException; + +public class FloatConverter implements ResultSetConverter { + @Override + public void append(ResultSet row, int columnIndex, ColumnBuilder properColumnBuilder) + throws SQLException { + float value = row.getFloat(columnIndex); + if (row.wasNull()) { + properColumnBuilder.appendNull(); + } else { + properColumnBuilder.writeFloat(value); + } + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/Int32Converter.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/Int32Converter.java new file mode 100644 index 00000000000..6b7fef75c6f --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/Int32Converter.java @@ -0,0 +1,38 @@ +/* + * 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.library.relational.tablefunction.connector.converter; + +import org.apache.tsfile.block.column.ColumnBuilder; + +import java.sql.ResultSet; +import java.sql.SQLException; + +public class Int32Converter implements ResultSetConverter { + @Override + public void append(ResultSet row, int columnIndex, ColumnBuilder properColumnBuilder) + throws SQLException { + int value = row.getInt(columnIndex); + if (row.wasNull()) { + properColumnBuilder.appendNull(); + } else { + properColumnBuilder.writeInt(value); + } + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/Int64Converter.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/Int64Converter.java new file mode 100644 index 00000000000..66138be99a7 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/Int64Converter.java @@ -0,0 +1,38 @@ +/* + * 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.library.relational.tablefunction.connector.converter; + +import org.apache.tsfile.block.column.ColumnBuilder; + +import java.sql.ResultSet; +import java.sql.SQLException; + +public class Int64Converter implements ResultSetConverter { + @Override + public void append(ResultSet row, int columnIndex, ColumnBuilder properColumnBuilder) + throws SQLException { + long value = row.getLong(columnIndex); + if (row.wasNull()) { + properColumnBuilder.appendNull(); + } else { + properColumnBuilder.writeLong(value); + } + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/ResultSetConverter.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/ResultSetConverter.java new file mode 100644 index 00000000000..516a3315ebb --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/ResultSetConverter.java @@ -0,0 +1,30 @@ +/* + * 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.library.relational.tablefunction.connector.converter; + +import org.apache.tsfile.block.column.ColumnBuilder; + +import java.sql.ResultSet; +import java.sql.SQLException; + +public interface ResultSetConverter { + void append(ResultSet row, int columnIndex, ColumnBuilder properColumnBuilder) + throws SQLException; +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/StringConverter.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/StringConverter.java new file mode 100644 index 00000000000..66b21a278dc --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/StringConverter.java @@ -0,0 +1,41 @@ +/* + * 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.library.relational.tablefunction.connector.converter; + +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.utils.Binary; + +import java.sql.ResultSet; +import java.sql.SQLException; + +public class StringConverter implements ResultSetConverter { + + @Override + public void append(ResultSet row, int columnIndex, ColumnBuilder properColumnBuilder) + throws SQLException { + String value = row.getString(columnIndex); + if (row.wasNull()) { + properColumnBuilder.appendNull(); + } else { + properColumnBuilder.writeBinary(new Binary(value, TSFileConfig.STRING_CHARSET)); + } + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/TimeConverter.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/TimeConverter.java new file mode 100644 index 00000000000..3e5b4c3c643 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/TimeConverter.java @@ -0,0 +1,38 @@ +/* + * 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.library.relational.tablefunction.connector.converter; + +import org.apache.tsfile.block.column.ColumnBuilder; + +import java.sql.ResultSet; +import java.sql.SQLException; + +public class TimeConverter implements ResultSetConverter { + @Override + public void append(ResultSet row, int columnIndex, ColumnBuilder properColumnBuilder) + throws SQLException { + java.sql.Time value = row.getTime(columnIndex); + if (value == null || row.wasNull()) { + properColumnBuilder.appendNull(); + } else { + properColumnBuilder.writeLong(value.getTime()); + } + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/TimestampConverter.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/TimestampConverter.java new file mode 100644 index 00000000000..7dce175703d --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/converter/TimestampConverter.java @@ -0,0 +1,38 @@ +/* + * 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.library.relational.tablefunction.connector.converter; + +import org.apache.tsfile.block.column.ColumnBuilder; + +import java.sql.ResultSet; +import java.sql.SQLException; + +public class TimestampConverter implements ResultSetConverter { + @Override + public void append(ResultSet row, int columnIndex, ColumnBuilder properColumnBuilder) + throws SQLException { + java.sql.Timestamp value = row.getTimestamp(columnIndex); + if (row.wasNull()) { + properColumnBuilder.appendNull(); + } else { + properColumnBuilder.writeLong(value.getTime()); + } + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/exception/CloseFailedInExternalDB.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/exception/CloseFailedInExternalDB.java new file mode 100644 index 00000000000..9bd987634b9 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/exception/CloseFailedInExternalDB.java @@ -0,0 +1,32 @@ +/* + * 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.library.relational.tablefunction.connector.exception; + +import org.apache.iotdb.library.i18n.LibraryUdfMessages; +import org.apache.iotdb.udf.api.exception.UDFException; + +public class CloseFailedInExternalDB extends UDFException { + public CloseFailedInExternalDB(String externalDB, Throwable throwable) { + super( + String.format( + LibraryUdfMessages.EXCEPTION_CLOSING_JDBC_RESOURCES_FAILED_IN_ARG_A300EDC2, externalDB), + throwable); + } +} diff --git a/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/exception/ExecutionFailedInExternalDB.java b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/exception/ExecutionFailedInExternalDB.java new file mode 100644 index 00000000000..ab586498457 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/relational/tablefunction/connector/exception/ExecutionFailedInExternalDB.java @@ -0,0 +1,32 @@ +/* + * 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.library.relational.tablefunction.connector.exception; + +import org.apache.iotdb.library.i18n.LibraryUdfMessages; +import org.apache.iotdb.udf.api.exception.UDFException; + +public class ExecutionFailedInExternalDB extends UDFException { + public ExecutionFailedInExternalDB(String externalDB, Throwable throwable) { + super( + String.format( + LibraryUdfMessages.EXCEPTION_QUERY_EXECUTION_FAILED_IN_ARG_E4F52371, externalDB), + throwable); + } +} diff --git a/library-udf/src/test/java/org/apache/iotdb/library/relational/tablefunction/connector/JDBCConnectorTableFunctionTest.java b/library-udf/src/test/java/org/apache/iotdb/library/relational/tablefunction/connector/JDBCConnectorTableFunctionTest.java new file mode 100644 index 00000000000..fb964ee46f2 --- /dev/null +++ b/library-udf/src/test/java/org/apache/iotdb/library/relational/tablefunction/connector/JDBCConnectorTableFunctionTest.java @@ -0,0 +1,325 @@ +/* + * 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.library.relational.tablefunction.connector; + +import org.apache.iotdb.library.relational.tablefunction.connector.exception.CloseFailedInExternalDB; +import org.apache.iotdb.library.relational.tablefunction.connector.exception.ExecutionFailedInExternalDB; +import org.apache.iotdb.udf.api.exception.UDFException; +import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis; +import org.apache.iotdb.udf.api.relational.table.TableFunctionHandle; +import org.apache.iotdb.udf.api.relational.table.argument.Argument; +import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument; +import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionLeafProcessor; +import org.apache.iotdb.udf.api.type.Type; + +import org.apache.tsfile.common.conf.TSFileDescriptor; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.common.block.TsBlockBuilder; +import org.junit.Test; + +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Proxy; +import java.sql.Connection; +import java.sql.Driver; +import java.sql.DriverPropertyInfo; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.sql.Statement; +import java.sql.Types; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.logging.Logger; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +public class JDBCConnectorTableFunctionTest { + @Test + public void preservesAliasesAndSerializedHandle() { + AtomicInteger closes = new AtomicInteger(); + ResultSetMetaData metadata = + proxy( + ResultSetMetaData.class, + (p, m, a) -> { + switch (m.getName()) { + case "getColumnCount": + return 1; + case "getColumnType": + return Types.INTEGER; + case "getColumnLabel": + return "renamed_id"; + case "getColumnName": + return "original_id"; + default: + throw new AssertionError(m.getName()); + } + }); + PreparedStatement statement = + proxy( + PreparedStatement.class, + (p, m, a) -> { + if (m.getName().equals("getMetaData")) return metadata; + if (m.getName().equals("close")) { + closes.incrementAndGet(); + return null; + } + throw new AssertionError(m.getName()); + }); + StubDriver.connection = + proxy( + Connection.class, + (p, m, a) -> { + if (m.getName().equals("prepareStatement")) return statement; + if (m.getName().equals("close")) { + closes.incrementAndGet(); + return null; + } + throw new AssertionError(m.getName()); + }); + BaseJDBCConnectorTableFunction function = function(); + TableFunctionAnalysis analysis = + function.analyze(arguments("SELECT original_id AS renamed_id")); + assertEquals( + "renamed_id", analysis.getProperColumnSchema().get().getFields().get(0).getName().get()); + assertEquals(2, closes.get()); + TableFunctionHandle copy = function.createTableFunctionHandle(); + copy.deserialize(analysis.getTableFunctionHandle().serialize()); + BaseJDBCConnectorTableFunction.BaseJDBCConnectorTableFunctionHandle restored = + (BaseJDBCConnectorTableFunction.BaseJDBCConnectorTableFunctionHandle) copy; + assertEquals("SELECT original_id AS renamed_id", restored.sql); + assertEquals("jdbc:stub:test", restored.url); + assertArrayEquals(new int[] {Types.INTEGER}, restored.types); + } + + @Test + public void closesConnectionAndStatementWhenExecutionFails() { + AtomicInteger closes = new AtomicInteger(); + Statement statement = + proxy( + Statement.class, + (p, m, a) -> { + if (m.getName().equals("executeQuery")) throw new SQLException("query failed"); + if (m.getName().equals("close")) { + closes.incrementAndGet(); + return null; + } + throw new AssertionError(m.getName()); + }); + StubDriver.connection = connection(statement, closes); + TableFunctionLeafProcessor processor = processor(); + assertThrows(ExecutionFailedInExternalDB.class, processor::beforeStart); + assertEquals(2, closes.get()); + processor.beforeDestroy(); + assertEquals(2, closes.get()); + } + + @Test + public void closesRemainingResourcesWhenResultSetCloseFails() { + AtomicInteger closes = new AtomicInteger(); + ResultSet result = + proxy( + ResultSet.class, + (p, m, a) -> { + if (m.getName().equals("close")) { + closes.incrementAndGet(); + throw new SQLException("close failed"); + } + throw new AssertionError(m.getName()); + }); + Statement statement = + proxy( + Statement.class, + (p, m, a) -> { + if (m.getName().equals("executeQuery")) return result; + if (m.getName().equals("close")) { + closes.incrementAndGet(); + return null; + } + throw new AssertionError(m.getName()); + }); + StubDriver.connection = connection(statement, closes); + TableFunctionLeafProcessor processor = processor(); + processor.beforeStart(); + assertThrows(CloseFailedInExternalDB.class, processor::beforeDestroy); + assertEquals(3, closes.get()); + processor.beforeDestroy(); + assertEquals(3, closes.get()); + } + + @Test + public void doesNotLoseRowsBetweenBatches() { + int rows = TSFileDescriptor.getInstance().getConfig().getMaxTsBlockLineNumber() * 2 + 1; + AtomicInteger row = new AtomicInteger(); + ResultSet result = + proxy( + ResultSet.class, + (p, m, a) -> { + switch (m.getName()) { + case "next": + return row.incrementAndGet() <= rows; + case "getInt": + return row.get(); + case "wasNull": + return false; + case "close": + return null; + default: + throw new AssertionError(m.getName()); + } + }); + Statement statement = + proxy( + Statement.class, + (p, m, a) -> { + if (m.getName().equals("executeQuery")) return result; + if (m.getName().equals("close")) return null; + throw new AssertionError(m.getName()); + }); + StubDriver.connection = connection(statement, new AtomicInteger()); + TableFunctionLeafProcessor processor = processor(); + processor.beforeStart(); + int total = 0; + int batches = 0; + while (!processor.isFinish()) { + assertTrue(++batches < 100); + TsBlockBuilder builder = new TsBlockBuilder(Collections.singletonList(TSDataType.INT32)); + processor.process(Arrays.asList(builder.getValueColumnBuilders())); + org.apache.tsfile.block.column.Column column = builder.getColumnBuilder(0).build(); + for (int i = 0; i < column.getPositionCount(); i++) assertEquals(++total, column.getInt(i)); + } + processor.beforeDestroy(); + assertEquals(rows, total); + assertTrue(batches >= 3); + } + + @Test + public void rejectsInvalidArgumentsBeforeConnecting() { + assertThrows(UDFException.class, () -> function().analyze(arguments(" "))); + } + + @Test + public void vendorDriversUseDistinctNamespacesAndRejectOtherUrls() throws Exception { + for (BaseJDBCConnectorTableFunction function : + Arrays.asList( + new PostgreSqlConnectorTableFunction(), + new OpenGaussConnectorTableFunction(), + new GaussDBConnectorTableFunction())) { + Driver driver = + (Driver) + Class.forName(function.getDriverClassName()).getDeclaredConstructor().newInstance(); + assertTrue(driver.acceptsURL(function.getDefaultUrl())); + assertThrows( + UDFException.class, + () -> + JDBCConnectionPool.getConnection( + function.getDriverClassName(), "jdbc:unsupported:test", "u", "p")); + } + } + + private static BaseJDBCConnectorTableFunction function() { + return new PostgreSqlConnectorTableFunction() { + @Override + String getDriverClassName() { + return StubDriver.class.getName(); + } + }; + } + + private static TableFunctionLeafProcessor processor() { + return function() + .getProcessorProvider( + new BaseJDBCConnectorTableFunction.BaseJDBCConnectorTableFunctionHandle( + "SELECT id", "jdbc:stub:test", "u", "p", new int[] {Types.INTEGER})) + .getSplitProcessor(); + } + + private static Map<String, Argument> arguments(String sql) { + Map<String, Argument> values = new HashMap<>(); + values.put("SQL", new ScalarArgument(Type.STRING, sql)); + values.put("URL", new ScalarArgument(Type.STRING, "jdbc:stub:test")); + values.put("USERNAME", new ScalarArgument(Type.STRING, "u")); + values.put("PASSWORD", new ScalarArgument(Type.STRING, "p")); + return values; + } + + private static Connection connection(Statement statement, AtomicInteger closes) { + return proxy( + Connection.class, + (p, m, a) -> { + if (m.getName().equals("createStatement")) return statement; + if (m.getName().equals("close")) { + closes.incrementAndGet(); + return null; + } + throw new AssertionError(m.getName()); + }); + } + + private static <T> T proxy(Class<T> type, InvocationHandler handler) { + return type.cast(Proxy.newProxyInstance(type.getClassLoader(), new Class<?>[] {type}, handler)); + } + + public static class StubDriver implements Driver { + static Connection connection; + + @Override + public Connection connect(String url, Properties info) { + return connection; + } + + @Override + public boolean acceptsURL(String url) { + return url.startsWith("jdbc:stub:"); + } + + @Override + public DriverPropertyInfo[] getPropertyInfo(String url, Properties info) { + return new DriverPropertyInfo[0]; + } + + @Override + public int getMajorVersion() { + return 1; + } + + @Override + public int getMinorVersion() { + return 0; + } + + @Override + public boolean jdbcCompliant() { + return false; + } + + @Override + public Logger getParentLogger() { + return Logger.getGlobal(); + } + } +}
