This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new de1a622 [IOTDB-1524] Support SELECT ... INTO ... clause (#3614)
de1a622 is described below
commit de1a622d90a5e453083af2e01a9d5fc4ca5d702a
Author: Steve Yurong Su (宇荣) <[email protected]>
AuthorDate: Tue Jul 27 06:24:09 2021 -0500
[IOTDB-1524] Support SELECT ... INTO ... clause (#3614)
---
.../antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4 | 46 +-
docs/UserGuide/Advanced-Features/Select-Into.md | 235 +++++++++
docs/zh/UserGuide/Advanced-Features/Select-Into.md | 234 +++++++++
.../main/java/org/apache/iotdb/SessionExample.java | 14 +
.../resources/conf/iotdb-engine.properties | 9 +
.../org/apache/iotdb/db/auth/AuthorityChecker.java | 1 +
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 14 +
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 15 +-
.../selectinto/InsertTabletPlanGenerator.java | 229 +++++++++
.../selectinto/InsertTabletPlansIterator.java | 139 +++++
.../main/java/org/apache/iotdb/db/qp/Planner.java | 22 +-
.../apache/iotdb/db/qp/constant/SQLConstant.java | 4 +
.../org/apache/iotdb/db/qp/logical/Operator.java | 2 +
.../db/qp/logical/crud/SelectIntoOperator.java | 110 ++++
.../apache/iotdb/db/qp/physical/PhysicalPlan.java | 17 +-
.../iotdb/db/qp/physical/crud/SelectIntoPlan.java | 113 +++++
.../apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java | 125 +++--
.../iotdb/db/qp/strategy/LogicalChecker.java | 5 +
.../iotdb/db/query/control/QueryTimeManager.java | 14 +
.../org/apache/iotdb/db/service/TSServiceImpl.java | 246 +++++----
.../iotdb/db/integration/IoTDBSelectIntoIT.java | 564 +++++++++++++++++++++
.../iotdb/db/qp/physical/PhysicalPlanTest.java | 33 +-
site/src/main/.vuepress/config.js | 6 +-
.../test/java/org/apache/iotdb/db/sql/Cases.java | 25 +
.../java/org/apache/iotdb/tsfile/utils/BitMap.java | 5 +
25 files changed, 2040 insertions(+), 187 deletions(-)
diff --git a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4
b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4
index 2e157d8..f3d1c5a 100644
--- a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4
+++ b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlBase.g4
@@ -105,7 +105,7 @@ statement
| START TRIGGER triggerName=ID #startTrigger
| STOP TRIGGER triggerName=ID #stopTrigger
| SHOW TRIGGERS #showTriggers
- | selectClause fromClause whereClause? specialClause? #selectStatement
+ | selectClause intoClause? fromClause whereClause? specialClause?
#selectStatement
| CREATE (CONTINUOUS QUERY | CQ) continuousQueryName=ID
resampleClause?
cqSelectIntoClause #createContinuousQueryStatement
@@ -114,26 +114,35 @@ statement
;
selectClause
- : SELECT (LAST | topClause)? resultColumn (COMMA resultColumn)*
- ;
+ : SELECT (LAST | topClause)? resultColumn (COMMA resultColumn)*
+ ;
resultColumn
- : expression (AS ID)?
- ;
+ : expression (AS ID)?
+ ;
expression
- : LR_BRACKET unary=expression RR_BRACKET
- | (PLUS | MINUS) unary=expression
- | leftExpression=expression (STAR | DIV | MOD) rightExpression=expression
- | leftExpression=expression (PLUS | MINUS) rightExpression=expression
- | functionName=suffixPath LR_BRACKET expression (COMMA expression)*
functionAttribute* RR_BRACKET
- | suffixPath
- | literal=SINGLE_QUOTE_STRING_LITERAL
- ;
+ : LR_BRACKET unary=expression RR_BRACKET
+ | (PLUS | MINUS) unary=expression
+ | leftExpression=expression (STAR | DIV | MOD) rightExpression=expression
+ | leftExpression=expression (PLUS | MINUS) rightExpression=expression
+ | functionName=suffixPath LR_BRACKET expression (COMMA expression)*
functionAttribute* RR_BRACKET
+ | suffixPath
+ | literal=SINGLE_QUOTE_STRING_LITERAL
+ ;
functionAttribute
- : COMMA functionAttributeKey=stringLiteral OPERATOR_EQ
functionAttributeValue=stringLiteral
- ;
+ : COMMA functionAttributeKey=stringLiteral OPERATOR_EQ
functionAttributeValue=stringLiteral
+ ;
+
+intoClause
+ : INTO intoPath (COMMA intoPath)*
+ ;
+
+intoPath
+ : fullPath
+ | nodeNameWithoutStar (DOT nodeNameWithoutStar)*
+ ;
alias
: LR_BRACKET ID RR_BRACKET
@@ -339,12 +348,7 @@ resampleClause
: RESAMPLE (EVERY DURATION)? (FOR DURATION)?;
cqSelectIntoClause
- : BEGIN
- selectClause
- INTO (fullPath | nodeNameWithoutStar)
- fromClause
- cqGroupByTimeClause
- END
+ : BEGIN selectClause INTO intoPath fromClause cqGroupByTimeClause END
;
cqGroupByTimeClause
diff --git a/docs/UserGuide/Advanced-Features/Select-Into.md
b/docs/UserGuide/Advanced-Features/Select-Into.md
new file mode 100644
index 0000000..8a914a1
--- /dev/null
+++ b/docs/UserGuide/Advanced-Features/Select-Into.md
@@ -0,0 +1,235 @@
+<!--
+
+ 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.
+
+-->
+
+
+
+# Query Write-back (SELECT INTO)
+
+The `SELECT ... INTO ...` statement copies data from query result set into
target time series.
+
+
+
+## SQL
+
+### Syntax
+
+**The following is the syntax definition of the `select` statement:**
+
+```sql
+selectClause
+intoClause?
+fromClause
+whereClause?
+specialClause?
+```
+
+If the `intoClause` is removed, then the `select` statement is a simple query
statement.
+
+The `intoClause` is the mark clause for query write-back.
+
+
+
+**The following is the definition of the `intoClause`:**
+
+```sql
+intoClause
+ : INTO intoPath (COMMA intoPath)*
+ ;
+
+intoPath
+ : fullPath
+ | nodeNameWithoutStar (DOT nodeNameWithoutStar)*
+ ;
+```
+
+There are 2 ways to specify target paths (`intoPath`).
+
+* Full target series name starting with `root`.
+
+ * Example：
+
+ ```sql
+ select s1, s1
+ into root.sg.d1.t1, root.sg.d1.t2
+ from root.sg.d1
+ ```
+
+* Suffix path does not start with `root`. In this case, the target series name
equals to the series prefix path in the `from` clause + the suffix path.
+
+ * Example:
+
+ ```sql
+ select s1, s1
+ into t1, t2
+ from root.sg.d1
+ ```
+
+ which equals to:
+
+ ```sql
+ select s1, s1
+ into root.sg.d1.t1, root.sg.d1.t2
+ from root.sg.d1
+ ```
+
+
+
+**In `intoPath`, you can also use `${i}` to represent part of the prefix path
in the `from` clause. **
+
+For example, for the path `root.sg1.d1.v1`, `${1}` means `sg1`, `${2}` means
`d1`, and `${3}` means `v1`.
+
+
+ * Example:
+
+ ```sql
+ select s1, s1, s1
+ into ${1}_t1, ${2}, root.${2}.${1}.t2
+ from root.sg.d1
+ ```
+
+ which equals to:
+
+ ```sql
+ select s1, s1, s1
+ into root.sg.d1.sg_t1, root.sg.d1.d1, root.d1.sg.t2
+ from root.sg.d1
+ ```
+
+
+
+### Supported Query Types
+
+**Note that except for the following types of queries, other types of queries
(such as `LAST` queries) are not supported. **
+
+* Raw time series query
+
+ ```sql
+ select s1, s1
+ into t1, t2
+ from root.sg.d1
+ ```
+
+* Time series generating function query（including UDF query）
+
+ ```sql
+ select s1, sin(s2)
+ into t1, t2
+ from root.sg.d1
+ ```
+
+* Arithmetic query
+
+ ```sql
+ select s1, sin(s2), s1 + s3
+ into t1, t2, t3
+ from root.sg.d1
+ ```
+
+* Fill query
+
+ ```sql
+ select s1
+ into fill_s1
+ from root.sg.d1
+ where time = 10
+ fill(float [linear, 1ms, 1ms])
+ ```
+
+* Group-by query
+
+ ```sql
+ select count(s1)
+ into group_by_s1
+ from root.sg.d1
+ group by ([1, 5), 1ms)
+ ```
+
+* Group-by-fill query
+
+ ```sql
+ select last_value(s1)
+ into group_by_fill_s1
+ from root.sg.d1
+ group by ([1, 10),1ms)
+ fill (float[PREVIOUS])
+ ```
+
+
+
+### Special Cluases Supported in Queries
+
+**Note that except for the following clauses, other query clauses (such as
`DESC`, `SOFFSET`, etc.) are not supported. **
+
+* Value filter
+
+ ```sql
+ select s1, s1
+ into t1, t2
+ from root.sg.d1
+ where s1 > 0 and s2 < 0
+ ```
+
+* Time filter
+
+ ```sql
+ select s1, s1
+ into t1, t2
+ from root.sg.d1
+ where time > 0
+ ```
+
+* LIMIT / OFFSET
+
+ ```sql
+ select s1, s1
+ into t1, t2
+ from root.sg.d1
+ limit 5 offset 1000
+ ```
+
+
+
+### Other Restrictions
+
+* The number of source series in the `select` clause and the number of target
series in the `into` clause must be the same.
+* The `select *` clause is not supported.
+* The target series in the `into` clause do not need to be created in advance.
+* When the target series in the `into` clause already exist, you need to
ensure that the source series in the `select` clause and the target series in
the `into` clause have the same data types.
+* The target series in the `into` clause must be different from each other.
+* Only one prefix path of a series is allowed in the `from` clause.
+
+
+
+## User Permission Management
+
+The user must have the following permissions to execute a query write-back
statement:
+
+* All `READ_TIMESERIES` permissions for the source series in the `select`
clause
+* All `INSERT_TIMESERIES` permissions for the target series in the `into`
clause
+
+For more user permissions related content, please refer to [Account Management
Statements](../Administration-Management/Administration.md).
+
+
+
+## Configurable Properties
+
+* `select_into_insert_tablet_plan_row_limit`: The maximum number of rows can
be processed in one insert-tablet-plan when executing select-into statements.
10000 by default.
+
diff --git a/docs/zh/UserGuide/Advanced-Features/Select-Into.md
b/docs/zh/UserGuide/Advanced-Features/Select-Into.md
new file mode 100644
index 0000000..f3f9460
--- /dev/null
+++ b/docs/zh/UserGuide/Advanced-Features/Select-Into.md
@@ -0,0 +1,234 @@
+<!--
+
+ 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.
+
+-->
+
+
+
+# 查询写回（SELECT ... INTO ...）
+
+`SELECT ... INTO ...` 语句允许您将查询结果集写回到指定序列上。
+
+
+
+## SQL
+
+### 语法
+
+**下面是 `select` 语句的语法定义：**
+
+```sql
+selectClause
+intoClause?
+fromClause
+whereClause?
+specialClause?
+```
+
+如果去除 `intoClause` 子句，那么 `select` 语句即是单纯的查询语句。
+
+`intoClause` 子句是写回功能的标记语句。
+
+
+
+**下面是 `intoClause` 子句的定义：**
+
+```sql
+intoClause
+ : INTO intoPath (COMMA intoPath)*
+ ;
+
+intoPath
+ : fullPath
+ | nodeNameWithoutStar (DOT nodeNameWithoutStar)*
+ ;
+```
+
+`intoPath`（目标序列）支持两种方式指定：
+
+* 以 `root` 开头的完整序列名指定
+
+ * 例子：
+
+ ```sql
+ select s1, s1
+ into root.sg.d1.t1, root.sg.d1.t2
+ from root.sg.d1
+ ```
+
+* 不以 `root` 开头的部分序列名指定，此时目标序列由 `from` 子句中的序列前缀和`intoPath`拼接而成
+
+ * 例子：
+
+ ```sql
+ select s1, s1
+ into t1, t2
+ from root.sg.d1
+ ```
+
+ 这等价于
+
+ ```sql
+ select s1, s1
+ into root.sg.d1.t1, root.sg.d1.t2
+ from root.sg.d1
+ ```
+
+
+
+**在`intoPath` 中，您还可以使用 `${i}`风格的路径匹配符来表示`from`子句中的部分路径。**
+
+比如，对于路径`root.sg1.d1.v1`而言，`${1}`表示`sg1`，`${2}`表示`d1`，`${3}`表示`v1`。
+
+
+ * 例子：
+
+ ```sql
+ select s1, s1, s1
+ into ${1}_t1, ${2}, root.${2}.${1}.t2
+ from root.sg.d1
+ ```
+
+ 这等价于
+
+ ```sql
+ select s1, s1, s1
+ into root.sg.d1.sg_t1, root.sg.d1.d1, root.d1.sg.t2
+ from root.sg.d1
+ ```
+
+
+
+### 支持写回的查询类型
+
+**注意，除了下述类型的查询，其余类型的查询（如`LAST`查询）都不被支持。**
+
+* 原始序列查询
+
+ ```sql
+ select s1, s1
+ into t1, t2
+ from root.sg.d1
+ ```
+
+* 时间序列生成函数查询（UDF查询）
+
+ ```sql
+ select s1, sin(s2)
+ into t1, t2
+ from root.sg.d1
+ ```
+
+* 数学表达式查询
+
+ ```sql
+ select s1, sin(s2), s1 + s3
+ into t1, t2, t3
+ from root.sg.d1
+ ```
+
+* Fill 查询
+
+ ```sql
+ select s1
+ into fill_s1
+ from root.sg.d1
+ where time = 10
+ fill(float [linear, 1ms, 1ms])
+ ```
+
+* Group By 查询
+
+ ```sql
+ select count(s1)
+ into group_by_s1
+ from root.sg.d1
+ group by ([1, 5), 1ms)
+ ```
+
+* Group By Fill 查询
+
+ ```sql
+ select last_value(s1)
+ into group_by_fill_s1
+ from root.sg.d1
+ group by ([1, 10),1ms)
+ fill (float[PREVIOUS])
+ ```
+
+
+
+### 支持写回的查询子句
+
+**注意，除了下述子句，其余查询子句（如 `DESC` / `SOFFSET` 等）都不被支持。**
+
+* 支持值过滤
+
+ ```sql
+ select s1, s1
+ into t1, t2
+ from root.sg.d1
+ where s1 > 0 and s2 < 0
+ ```
+
+* 支持时间过滤
+
+ ```sql
+ select s1, s1
+ into t1, t2
+ from root.sg.d1
+ where time > 0
+ ```
+
+* LIMIT / OFFSET
+
+ ```sql
+ select s1, s1
+ into t1, t2
+ from root.sg.d1
+ limit 5 offset 1000
+ ```
+
+
+
+### 其他限制
+
+* `select`子句中的源序列和`into`子句中的目标序列数量必须相同
+* `select`子句不支持带 `*` 查询
+* `into`子句中的目标序列不必预先创建（可使用自动创建schema功能）
+* 当`into`子句中的目标序列已存在时，您需要保证`select`子句中的源序列和`into`子句中的目标序列的数据类型一致
+* `into`子句中的目标序列必须是互不相同的
+* `from`子句只允许有一列序列前缀
+
+
+
+## 权限
+
+用户必须有下列权限才能正常执行查询写回语句：
+
+* 所有 `select` 子句中源序列的 `READ_TIMESERIES` 权限
+* 所有 `into` 子句中目标序列 `INSERT_TIMESERIES` 权限
+
+更多用户权限相关的内容，请参考[权限管理语句](../Administration-Management/Administration.md)。
+
+
+
+## 配置参数
+
+* `select_into_insert_tablet_plan_row_limit`：执行 select-into 语句时，一个
insert-tablet-plan 中可以处理的最大行数。 默认为 10000。
diff --git a/example/session/src/main/java/org/apache/iotdb/SessionExample.java
b/example/session/src/main/java/org/apache/iotdb/SessionExample.java
index e3f1243..0f1ad8b 100644
--- a/example/session/src/main/java/org/apache/iotdb/SessionExample.java
+++ b/example/session/src/main/java/org/apache/iotdb/SessionExample.java
@@ -74,6 +74,7 @@ public class SessionExample {
insertTablet();
insertTablets();
insertRecords();
+ selectInto();
createAndDropContinuousQueries();
nonQuery();
query();
@@ -487,6 +488,19 @@ public class SessionExample {
}
}
+ private static void selectInto() throws IoTDBConnectionException,
StatementExecutionException {
+ session.executeNonQueryStatement(
+ "select s1, s2, s3 into into_s1, into_s2, into_s3 from root.sg1.d1");
+
+ SessionDataSet dataSet =
+ session.executeQueryStatement("select into_s1, into_s2, into_s3 from
root.sg1.d1");
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ dataSet.closeOperationHandle();
+ }
+
private static void deleteData() throws IoTDBConnectionException,
StatementExecutionException {
String path = ROOT_SG1_D1_S1;
long deleteTime = 99;
diff --git a/server/src/assembly/resources/conf/iotdb-engine.properties
b/server/src/assembly/resources/conf/iotdb-engine.properties
index 63e79d4..c8fb395 100644
--- a/server/src/assembly/resources/conf/iotdb-engine.properties
+++ b/server/src/assembly/resources/conf/iotdb-engine.properties
@@ -837,6 +837,15 @@ timestamp_precision=ms
# continuous_query_min_every_interval=1s
####################
+### Select-Into Configuration
+####################
+
+# The maximum number of rows can be processed in insert-tablet-plan when
executing select-into statements.
+# When <= 0, use 10000.
+# Datatype: int
+# select_into_insert_tablet_plan_row_limit=10000
+
+####################
### Index Configuration
####################
diff --git
a/server/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
b/server/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
index c0f86b3..ac89e4b 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java
@@ -131,6 +131,7 @@ public class AuthorityChecker {
case LAST:
case FILL:
case GROUP_BY_FILL:
+ case SELECT_INTO:
return PrivilegeType.READ_TIMESERIES.ordinal();
case INSERT:
case LOAD_DATA:
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 3979903..1ed7acd 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -563,6 +563,12 @@ public class IoTDBConfig {
*/
private long continuousQueryMinimumEveryInterval = 1000;
+ /**
+ * The maximum number of rows can be processed in insert-tablet-plan when
executing select-into
+ * statements.
+ */
+ private int selectIntoInsertTabletPlanRowLimit = 10000;
+
private MergeFileStrategy mergeFileStrategy =
MergeFileStrategy.MAX_SERIES_NUM;
/** Default system file storage is in local file system (unsupported) */
@@ -1488,6 +1494,14 @@ public class IoTDBConfig {
this.continuousQueryMinimumEveryInterval = minimumEveryInterval;
}
+ public int getSelectIntoInsertTabletPlanRowLimit() {
+ return selectIntoInsertTabletPlanRowLimit;
+ }
+
+ public void setSelectIntoInsertTabletPlanRowLimit(int
selectIntoInsertTabletPlanRowLimit) {
+ this.selectIntoInsertTabletPlanRowLimit =
selectIntoInsertTabletPlanRowLimit;
+ }
+
public int getMergeWriteThroughputMbPerSec() {
return mergeWriteThroughputMbPerSec;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 1b0e4eb..149ae46 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -46,7 +46,7 @@ public class IoTDBDescriptor {
private static final Logger logger =
LoggerFactory.getLogger(IoTDBDescriptor.class);
- private IoTDBConfig conf = new IoTDBConfig();
+ private final IoTDBConfig conf = new IoTDBConfig();
protected IoTDBDescriptor() {
loadProps();
@@ -741,6 +741,12 @@ public class IoTDBDescriptor {
conf.setAdminPassword(properties.getProperty("admin_password",
conf.getAdminPassword()));
+ conf.setSelectIntoInsertTabletPlanRowLimit(
+ Integer.parseInt(
+ properties.getProperty(
+ "select_into_insert_tablet_plan_row_limit",
+
String.valueOf(conf.getSelectIntoInsertTabletPlanRowLimit()))));
+
// At the same time, set TSFileConfig
TSFileDescriptor.getInstance()
.getConfig()
@@ -1088,6 +1094,13 @@ public class IoTDBDescriptor {
properties.getProperty(
"merge_write_throughput_mb_per_sec",
Integer.toString(conf.getMergeWriteThroughputMbPerSec()))));
+
+ // update insert-tablet-plan's row limit for select-into
+ conf.setSelectIntoInsertTabletPlanRowLimit(
+ Integer.parseInt(
+ properties.getProperty(
+ "select_into_insert_tablet_plan_row_limit",
+
String.valueOf(conf.getSelectIntoInsertTabletPlanRowLimit()))));
} catch (Exception e) {
throw new QueryProcessException(String.format("Fail to reload
configuration because %s", e));
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/selectinto/InsertTabletPlanGenerator.java
b/server/src/main/java/org/apache/iotdb/db/engine/selectinto/InsertTabletPlanGenerator.java
new file mode 100644
index 0000000..4b12b26
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/selectinto/InsertTabletPlanGenerator.java
@@ -0,0 +1,229 @@
+/*
+ * 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.db.engine.selectinto;
+
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
+import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Field;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.utils.BitMap;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/** internallyConstructNewPlan -> collectRowRecord * N ->
generateInsertTabletPlan */
+public class InsertTabletPlanGenerator {
+
+ private final String targetDevice;
+ // the index of column in InsertTabletPlan -> the index of output column of
query data set
+ private final List<Integer> queryDataSetIndexes;
+ // the index of column in InsertTabletPlan -> the measurement id of the
column
+ private final List<String> targetMeasurementIds;
+
+ private final int tabletRowLimit;
+
+ // the following fields are used to construct plan
+ private int rowCount;
+ private long[] times;
+ private Object[] columns;
+ private BitMap[] bitMaps;
+ private TSDataType[] dataTypes;
+
+ private int numberOfInitializedColumns;
+
+ public InsertTabletPlanGenerator(String targetDevice, int tabletRowLimit) {
+ this.targetDevice = targetDevice;
+ queryDataSetIndexes = new ArrayList<>();
+ targetMeasurementIds = new ArrayList<>();
+
+ this.tabletRowLimit = tabletRowLimit;
+ }
+
+ public void collectTargetPathInformation(String targetMeasurementId, int
queryDataSetIndex) {
+ targetMeasurementIds.add(targetMeasurementId);
+ queryDataSetIndexes.add(queryDataSetIndex);
+ }
+
+ public void internallyConstructNewPlan() {
+ rowCount = 0;
+ times = new long[tabletRowLimit];
+ columns = new Object[targetMeasurementIds.size()];
+ bitMaps = new BitMap[targetMeasurementIds.size()];
+ for (int i = 0; i < bitMaps.length; ++i) {
+ bitMaps[i] = new BitMap(tabletRowLimit);
+ bitMaps[i].markAll();
+ }
+ dataTypes = new TSDataType[targetMeasurementIds.size()];
+
+ numberOfInitializedColumns = 0;
+ }
+
+ public void collectRowRecord(RowRecord rowRecord) {
+ if (numberOfInitializedColumns != columns.length) {
+ List<Integer> initializedDataTypeIndexes = trySetDataTypes(rowRecord);
+ tryInitColumns(initializedDataTypeIndexes);
+ numberOfInitializedColumns += initializedDataTypeIndexes.size();
+ }
+
+ times[rowCount] = rowRecord.getTimestamp();
+
+ for (int i = 0; i < columns.length; ++i) {
+ Field field = rowRecord.getFields().get(queryDataSetIndexes.get(i));
+
+ // if the field is NULL
+ if (field == null || field.getDataType() == null) {
+ // bit in bitMaps are marked as 1 (NULL) by default
+ continue;
+ }
+
+ bitMaps[i].unmark(rowCount);
+ switch (field.getDataType()) {
+ case INT32:
+ ((int[]) columns[i])[rowCount] = field.getIntV();
+ break;
+ case INT64:
+ ((long[]) columns[i])[rowCount] = field.getLongV();
+ break;
+ case FLOAT:
+ ((float[]) columns[i])[rowCount] = field.getFloatV();
+ break;
+ case DOUBLE:
+ ((double[]) columns[i])[rowCount] = field.getDoubleV();
+ break;
+ case BOOLEAN:
+ ((boolean[]) columns[i])[rowCount] = field.getBoolV();
+ break;
+ case TEXT:
+ ((Binary[]) columns[i])[rowCount] = field.getBinaryV();
+ break;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format(
+ "data type %s is not supported when convert data at client",
+ field.getDataType()));
+ }
+ }
+
+ ++rowCount;
+ }
+
+ private List<Integer> trySetDataTypes(RowRecord rowRecord) {
+ List<Integer> initializedDataTypeIndexes = new ArrayList<>();
+ List<Field> fields = rowRecord.getFields();
+
+ for (int i = 0; i < dataTypes.length; ++i) {
+ // if the data type is already set
+ if (dataTypes[i] != null) {
+ continue;
+ }
+
+ // get the field index of the row record
+ int queryDataSetIndex = queryDataSetIndexes.get(i);
+ // if the field is not null
+ if (fields.get(queryDataSetIndex) != null
+ && fields.get(queryDataSetIndex).getDataType() != null) {
+ // set the data type to the field type
+ dataTypes[i] = fields.get(queryDataSetIndex).getDataType();
+ initializedDataTypeIndexes.add(i);
+ }
+ }
+
+ for (int i = 0; i < dataTypes.length; ++i) {
+ if (dataTypes[i] == null && fields.get(i) != null &&
fields.get(i).getDataType() != null) {
+ dataTypes[i] = fields.get(i).getDataType();
+ initializedDataTypeIndexes.add(i);
+ }
+ }
+ return initializedDataTypeIndexes;
+ }
+
+ private void tryInitColumns(List<Integer> initializedDataTypeIndexes) {
+ for (int i : initializedDataTypeIndexes) {
+ switch (dataTypes[i]) {
+ case BOOLEAN:
+ columns[i] = new boolean[tabletRowLimit];
+ break;
+ case INT32:
+ columns[i] = new int[tabletRowLimit];
+ break;
+ case INT64:
+ columns[i] = new long[tabletRowLimit];
+ break;
+ case FLOAT:
+ columns[i] = new float[tabletRowLimit];
+ break;
+ case DOUBLE:
+ columns[i] = new double[tabletRowLimit];
+ break;
+ case TEXT:
+ columns[i] = new Binary[tabletRowLimit];
+ break;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format(
+ "data type %s is not supported when convert data at client",
dataTypes[i]));
+ }
+ }
+ }
+
+ public InsertTabletPlan generateInsertTabletPlan() throws
IllegalPathException {
+ List<String> nonEmptyColumnNames = new ArrayList<>();
+
+ int countOfNonEmptyColumns = 0;
+ for (int i = 0; i < columns.length; ++i) {
+ if (columns[i] == null) {
+ continue;
+ }
+
+ nonEmptyColumnNames.add(targetMeasurementIds.get(i));
+ times[countOfNonEmptyColumns] = times[i];
+ columns[countOfNonEmptyColumns] = columns[i];
+ bitMaps[countOfNonEmptyColumns] = bitMaps[i];
+ dataTypes[countOfNonEmptyColumns] = dataTypes[i];
+
+ ++countOfNonEmptyColumns;
+ }
+
+ InsertTabletPlan insertTabletPlan =
+ new InsertTabletPlan(new PartialPath(targetDevice),
nonEmptyColumnNames);
+
+ insertTabletPlan.setAligned(false);
+ insertTabletPlan.setRowCount(rowCount);
+
+ if (countOfNonEmptyColumns == columns.length) {
+ insertTabletPlan.setTimes(times);
+ insertTabletPlan.setColumns(columns);
+ insertTabletPlan.setBitMaps(bitMaps);
+ insertTabletPlan.setDataTypes(dataTypes);
+ } else {
+ insertTabletPlan.setTimes(Arrays.copyOf(times, countOfNonEmptyColumns));
+ insertTabletPlan.setColumns(Arrays.copyOf(columns,
countOfNonEmptyColumns));
+ insertTabletPlan.setBitMaps(Arrays.copyOf(bitMaps,
countOfNonEmptyColumns));
+ insertTabletPlan.setDataTypes(Arrays.copyOf(dataTypes,
countOfNonEmptyColumns));
+ }
+
+ return insertTabletPlan;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/selectinto/InsertTabletPlansIterator.java
b/server/src/main/java/org/apache/iotdb/db/engine/selectinto/InsertTabletPlansIterator.java
new file mode 100644
index 0000000..95051ff
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/selectinto/InsertTabletPlansIterator.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.engine.selectinto;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
+import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
+import org.apache.iotdb.db.query.expression.ResultColumn;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class InsertTabletPlansIterator {
+
+ private static final Pattern leveledPathNodePattern =
Pattern.compile("\\$\\{\\w+}");
+
+ private final QueryPlan queryPlan;
+ private final QueryDataSet queryDataSet;
+
+ private final PartialPath fromPath;
+ private final List<PartialPath> intoPaths;
+
+ private final int tabletRowLimit;
+
+ private InsertTabletPlanGenerator[] insertTabletPlanGenerators;
+
+ public InsertTabletPlansIterator(
+ QueryPlan queryPlan,
+ QueryDataSet queryDataSet,
+ PartialPath fromPath,
+ List<PartialPath> intoPaths)
+ throws IllegalPathException {
+ this.queryPlan = queryPlan;
+ this.queryDataSet = queryDataSet;
+ this.fromPath = fromPath;
+ this.intoPaths = intoPaths;
+
+ tabletRowLimit =
+
IoTDBDescriptor.getInstance().getConfig().getSelectIntoInsertTabletPlanRowLimit();
+
+ generateActualIntoPaths();
+ constructInsertTabletPlanGenerators();
+ }
+
+ private void generateActualIntoPaths() throws IllegalPathException {
+ for (int i = 0; i < intoPaths.size(); ++i) {
+ intoPaths.set(i, generateActualIntoPath(i));
+ }
+ }
+
+ private PartialPath generateActualIntoPath(int index) throws
IllegalPathException {
+ String[] nodes = fromPath.getNodes();
+ StringBuffer sb = new StringBuffer();
+ Matcher m =
leveledPathNodePattern.matcher(intoPaths.get(index).getFullPath());
+ while (m.find()) {
+ String param = m.group();
+ String value = nodes[Integer.parseInt(param.substring(2, param.length()
- 1).trim())];
+ m.appendReplacement(sb, value == null ? "" : value);
+ }
+ m.appendTail(sb);
+ return new PartialPath(sb.toString());
+ }
+
+ private void constructInsertTabletPlanGenerators() {
+ final Map<String, Integer> sourcePathToQueryDataSetIndex =
queryPlan.getPathToIndex();
+ final List<ResultColumn> resultColumns = queryPlan.getResultColumns();
+
+ Map<String, InsertTabletPlanGenerator> deviceToPlanGeneratorMap = new
HashMap<>();
+ for (int i = 0, intoPathsSize = intoPaths.size(); i < intoPathsSize; i++) {
+ String device = intoPaths.get(i).getDevice();
+ if (!deviceToPlanGeneratorMap.containsKey(device)) {
+ deviceToPlanGeneratorMap.put(device, new
InsertTabletPlanGenerator(device, tabletRowLimit));
+ }
+ deviceToPlanGeneratorMap
+ .get(device)
+ .collectTargetPathInformation(
+ intoPaths.get(i).getMeasurement(),
+
sourcePathToQueryDataSetIndex.get(resultColumns.get(i).getResultColumnName()));
+ }
+
+ insertTabletPlanGenerators =
+ deviceToPlanGeneratorMap.values().toArray(new
InsertTabletPlanGenerator[0]);
+ }
+
+ public boolean hasNext() throws IOException {
+ return queryDataSet.hasNext();
+ }
+
+ public List<InsertTabletPlan> next() throws IOException,
IllegalPathException {
+ for (InsertTabletPlanGenerator insertTabletPlanGenerator :
insertTabletPlanGenerators) {
+ insertTabletPlanGenerator.internallyConstructNewPlan();
+ }
+
+ collectRowRecordIntoInsertTabletPlanGenerators();
+
+ List<InsertTabletPlan> insertTabletPlans = new ArrayList<>();
+ for (InsertTabletPlanGenerator insertTabletPlanGenerator :
insertTabletPlanGenerators) {
+
insertTabletPlans.add(insertTabletPlanGenerator.generateInsertTabletPlan());
+ }
+ return insertTabletPlans;
+ }
+
+ private void collectRowRecordIntoInsertTabletPlanGenerators() throws
IOException {
+ int count = 0;
+ while (queryDataSet.hasNext() && count < tabletRowLimit) {
+ RowRecord rowRecord = queryDataSet.next();
+ for (InsertTabletPlanGenerator insertTabletPlanGenerator :
insertTabletPlanGenerators) {
+ insertTabletPlanGenerator.collectRowRecord(rowRecord);
+ }
+ ++count;
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/Planner.java
b/server/src/main/java/org/apache/iotdb/db/qp/Planner.java
index 94b43ff..7d469a4 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/Planner.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/Planner.java
@@ -26,6 +26,7 @@ import
org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.qp.logical.Operator;
import org.apache.iotdb.db.qp.logical.crud.FilterOperator;
import org.apache.iotdb.db.qp.logical.crud.QueryOperator;
+import org.apache.iotdb.db.qp.logical.crud.SelectIntoOperator;
import org.apache.iotdb.db.qp.logical.crud.WhereComponent;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan;
@@ -42,9 +43,6 @@ import org.apache.iotdb.service.rpc.thrift.TSRawDataQueryReq;
import java.time.ZoneId;
-import static org.apache.iotdb.db.qp.logical.Operator.OperatorType.QUERY;
-import static org.apache.iotdb.db.qp.logical.Operator.OperatorType.QUERY_INDEX;
-
/** provide a integration method for other user. */
public class Planner {
@@ -109,9 +107,15 @@ public class Planner {
*/
protected Operator logicalOptimize(Operator operator)
throws LogicalOperatorException, PathNumOverLimitException {
- return operator.getType().equals(QUERY) ||
operator.getType().equals(QUERY_INDEX)
- ? optimizeQueryOperator((QueryOperator) operator)
- : operator;
+ switch (operator.getType()) {
+ case QUERY:
+ case QUERY_INDEX:
+ return optimizeQueryOperator((QueryOperator) operator);
+ case SELECT_INTO:
+ return optimizeSelectIntoOperator((SelectIntoOperator) operator);
+ default:
+ return operator;
+ }
}
/**
@@ -138,6 +142,12 @@ public class Planner {
return root;
}
+ private Operator optimizeSelectIntoOperator(SelectIntoOperator operator)
+ throws PathNumOverLimitException, LogicalOperatorException {
+
operator.setQueryOperator(optimizeQueryOperator(operator.getQueryOperator()));
+ return operator;
+ }
+
@TestOnly
public PhysicalPlan parseSQLToPhysicalPlan(String sqlStr) throws
QueryProcessException {
return parseSQLToPhysicalPlan(sqlStr, ZoneId.systemDefault());
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
b/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
index 7509bf0..bf2e9c2 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
@@ -179,6 +179,8 @@ public class SQLConstant {
public static final int TOK_CONTINUOUS_QUERY_DROP = 107;
public static final int TOK_SHOW_CONTINUOUS_QUERIES = 108;
+ public static final int TOK_SELECT_INTO = 109;
+
public static final Map<Integer, String> tokenNames = new HashMap<>();
public static String[] getSingleRootArray() {
@@ -247,6 +249,8 @@ public class SQLConstant {
tokenNames.put(TOK_CONTINUOUS_QUERY_CREATE, "TOK_CONTINUOUS_QUERY_CREATE");
tokenNames.put(TOK_CONTINUOUS_QUERY_DROP, "TOK_CONTINUOUS_QUERY_DROP");
tokenNames.put(TOK_SHOW_CONTINUOUS_QUERIES, "TOK_SHOW_CONTINUOUS_QUERIES");
+
+ tokenNames.put(TOK_SELECT_INTO, "TOK_SELECT_INTO");
}
public static boolean isReservedPath(PartialPath pathStr) {
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
b/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
index 56eba91..53234cb 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
@@ -126,6 +126,8 @@ public abstract class Operator {
UDAF,
UDTF,
+ SELECT_INTO,
+
CREATE_FUNCTION,
DROP_FUNCTION,
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectIntoOperator.java
b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectIntoOperator.java
new file mode 100644
index 0000000..a91fd37
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/SelectIntoOperator.java
@@ -0,0 +1,110 @@
+/*
+ * 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.db.qp.logical.crud;
+
+import org.apache.iotdb.db.exception.query.LogicalOperatorException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.qp.constant.SQLConstant;
+import org.apache.iotdb.db.qp.logical.Operator;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
+import org.apache.iotdb.db.qp.physical.crud.SelectIntoPlan;
+import org.apache.iotdb.db.qp.strategy.PhysicalGenerator;
+
+import java.util.HashSet;
+import java.util.List;
+
+public class SelectIntoOperator extends Operator {
+
+ private QueryOperator queryOperator;
+
+ private List<PartialPath> intoPaths;
+
+ public SelectIntoOperator() {
+ super(SQLConstant.TOK_SELECT_INTO);
+ operatorType = OperatorType.SELECT_INTO;
+ }
+
+ @Override
+ public PhysicalPlan generatePhysicalPlan(PhysicalGenerator generator)
+ throws QueryProcessException {
+ QueryPlan queryPlan = (QueryPlan)
queryOperator.generatePhysicalPlan(generator);
+ if (intoPaths.size() != queryPlan.getPaths().size()) {
+ throw new QueryProcessException(
+ "select into: the number of source paths and the number of target
paths should be the same.");
+ }
+ return new SelectIntoPlan(
+ queryPlan, queryOperator.getFromComponent().getPrefixPaths().get(0),
intoPaths);
+ }
+
+ public void check() throws LogicalOperatorException {
+ queryOperator.check();
+
+ if (intoPaths.size() > new HashSet<>(intoPaths).size()) {
+ throw new LogicalOperatorException(
+ "select into: target paths in into clause should be different.");
+ }
+
+ if (queryOperator.isAlignByDevice()) {
+ throw new LogicalOperatorException("select into: align by device clauses
are not supported.");
+ }
+
+ // disable align
+ if (!queryOperator.isAlignByTime()) {
+ throw new LogicalOperatorException("select into: disable align clauses
are not supported.");
+ }
+
+ if (queryOperator instanceof LastQueryOperator) {
+ throw new LogicalOperatorException("select into: last clauses are not
supported.");
+ }
+
+ if (queryOperator instanceof AggregationQueryOperator
+ && !(queryOperator instanceof GroupByQueryOperator)) {
+ throw new LogicalOperatorException("select into: aggregation queries are
not supported.");
+ }
+
+ if (queryOperator.getSpecialClauseComponent() != null) {
+ SpecialClauseComponent specialClauseComponent =
queryOperator.getSpecialClauseComponent();
+ if (specialClauseComponent.hasSlimit()) {
+ throw new LogicalOperatorException("select into: slimit clauses are
not supported.");
+ }
+ if (specialClauseComponent.getSeriesOffset() > 0) {
+ throw new LogicalOperatorException("select into: soffset clauses are
not supported.");
+ }
+ if (!specialClauseComponent.isAscending()) {
+ throw new LogicalOperatorException(
+ "select into: order by time desc clauses are not supported.");
+ }
+ }
+ }
+
+ public void setQueryOperator(QueryOperator queryOperator) {
+ this.queryOperator = queryOperator;
+ }
+
+ public QueryOperator getQueryOperator() {
+ return queryOperator;
+ }
+
+ public void setIntoPaths(List<PartialPath> intoPaths) {
+ this.intoPaths = intoPaths;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
index 2c87217..a018f1c 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
@@ -29,6 +29,7 @@ import
org.apache.iotdb.db.qp.physical.crud.InsertMultiTabletPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowsPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
+import org.apache.iotdb.db.qp.physical.crud.SelectIntoPlan;
import org.apache.iotdb.db.qp.physical.crud.SetDeviceTemplatePlan;
import org.apache.iotdb.db.qp.physical.sys.AlterTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
@@ -120,6 +121,10 @@ public abstract class PhysicalPlan {
return isQuery;
}
+ public boolean isSelectInto() {
+ return false;
+ }
+
public Operator.OperatorType getOperatorType() {
return operatorType;
}
@@ -170,7 +175,7 @@ public abstract class PhysicalPlan {
*
* @param buffer
*/
- public void deserialize(ByteBuffer buffer) throws IllegalPathException {
+ public void deserialize(ByteBuffer buffer) throws IllegalPathException,
IOException {
throw new UnsupportedOperationException(SERIALIZATION_UNIMPLEMENTED);
}
@@ -223,7 +228,9 @@ public abstract class PhysicalPlan {
}
public void setLoginUserName(String loginUserName) {
- this.loginUserName = loginUserName;
+ if (this instanceof AuthorPlan) {
+ this.loginUserName = loginUserName;
+ }
}
public static class Factory {
@@ -400,6 +407,9 @@ public abstract class PhysicalPlan {
case DROP_FUNCTION:
plan = new DropFunctionPlan();
break;
+ case SELECT_INTO:
+ plan = new SelectIntoPlan();
+ break;
default:
throw new IOException("unrecognized log type " + type);
}
@@ -464,7 +474,8 @@ public abstract class PhysicalPlan {
CREATE_SNAPSHOT,
CLEARCACHE,
CREATE_FUNCTION,
- DROP_FUNCTION
+ DROP_FUNCTION,
+ SELECT_INTO
}
public long getIndex() {
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/SelectIntoPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/SelectIntoPlan.java
new file mode 100644
index 0000000..b2c2a06
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/SelectIntoPlan.java
@@ -0,0 +1,113 @@
+/*
+ * 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.db.qp.physical.crud;
+
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+public class SelectIntoPlan extends PhysicalPlan {
+
+ private QueryPlan queryPlan;
+ private PartialPath fromPath;
+ private List<PartialPath> intoPaths;
+
+ public SelectIntoPlan() {
+ super(false, OperatorType.SELECT_INTO);
+ }
+
+ public SelectIntoPlan(QueryPlan queryPlan, PartialPath fromPath,
List<PartialPath> intoPaths) {
+ super(false, OperatorType.SELECT_INTO);
+ this.queryPlan = queryPlan;
+ this.fromPath = fromPath;
+ this.intoPaths = intoPaths;
+ }
+
+ @Override
+ public boolean isSelectInto() {
+ return true;
+ }
+
+ @Override
+ public void serialize(DataOutputStream outputStream) throws IOException {
+ outputStream.writeByte((byte) PhysicalPlanType.SELECT_INTO.ordinal());
+
+ queryPlan.serialize(outputStream);
+
+ putString(outputStream, fromPath.getFullPath());
+
+ outputStream.write(intoPaths.size());
+ for (PartialPath intoPath : intoPaths) {
+ putString(outputStream, intoPath.getFullPath());
+ }
+ }
+
+ @Override
+ public void serialize(ByteBuffer buffer) {
+ buffer.put((byte) PhysicalPlanType.SELECT_INTO.ordinal());
+
+ queryPlan.serialize(buffer);
+
+ putString(buffer, fromPath.getFullPath());
+
+ buffer.putInt(intoPaths.size());
+ for (PartialPath intoPath : intoPaths) {
+ putString(buffer, intoPath.getFullPath());
+ }
+ }
+
+ @Override
+ public void deserialize(ByteBuffer buffer) throws IllegalPathException,
IOException {
+ queryPlan = (QueryPlan) Factory.create(buffer);
+
+ fromPath = new PartialPath(readString(buffer));
+
+ int intoPathsSize = buffer.getInt();
+ intoPaths = new ArrayList<>(intoPathsSize);
+ for (int i = 0; i < intoPathsSize; ++i) {
+ intoPaths.add(new PartialPath(readString(buffer)));
+ }
+ }
+
+ /** mainly for query auth. */
+ @Override
+ public List<PartialPath> getPaths() {
+ return queryPlan.getPaths();
+ }
+
+ public QueryPlan getQueryPlan() {
+ return queryPlan;
+ }
+
+ public PartialPath getFromPath() {
+ return fromPath;
+ }
+
+ public List<PartialPath> getIntoPaths() {
+ return intoPaths;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
b/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
index 5d6989c..bcd6a06 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
@@ -45,6 +45,7 @@ import org.apache.iotdb.db.qp.logical.crud.InsertOperator;
import org.apache.iotdb.db.qp.logical.crud.LastQueryOperator;
import org.apache.iotdb.db.qp.logical.crud.QueryOperator;
import org.apache.iotdb.db.qp.logical.crud.SelectComponent;
+import org.apache.iotdb.db.qp.logical.crud.SelectIntoOperator;
import org.apache.iotdb.db.qp.logical.crud.SpecialClauseComponent;
import org.apache.iotdb.db.qp.logical.crud.UDFQueryOperator;
import org.apache.iotdb.db.qp.logical.crud.WhereComponent;
@@ -157,6 +158,7 @@ import
org.apache.iotdb.db.qp.sql.SqlBaseParser.InsertColumnsSpecContext;
import org.apache.iotdb.db.qp.sql.SqlBaseParser.InsertMultiValueContext;
import org.apache.iotdb.db.qp.sql.SqlBaseParser.InsertStatementContext;
import org.apache.iotdb.db.qp.sql.SqlBaseParser.InsertValuesSpecContext;
+import org.apache.iotdb.db.qp.sql.SqlBaseParser.IntoPathContext;
import org.apache.iotdb.db.qp.sql.SqlBaseParser.KillQueryContext;
import org.apache.iotdb.db.qp.sql.SqlBaseParser.LimitClauseContext;
import org.apache.iotdb.db.qp.sql.SqlBaseParser.LimitStatementContext;
@@ -285,7 +287,9 @@ public class IoTDBSqlVisitor extends
SqlBaseBaseVisitor<Operator> {
"For delete statement, where clause can only contain atomic expressions
like : "
+ "time > XXX, time <= XXX, or two atomic expressions connected by
'AND'";
- private static final Pattern cqLevelNodePattern =
Pattern.compile("\\$\\{\\w+}");
+ // used to match "{x}", where x is a integer.
+ // for create-cq clause and select-into clause.
+ private static final Pattern leveledPathNodePattern =
Pattern.compile("\\$\\{\\w+}");
private ZoneId zoneId;
private QueryOperator queryOp;
@@ -1046,8 +1050,8 @@ public class IoTDBSqlVisitor extends
SqlBaseBaseVisitor<Operator> {
queryOp.setWhereComponent(whereComponent);
}
}
-
- return queryOp;
+ // 4. Check whether it's a select-into clause
+ return ctx.intoClause() == null ? queryOp :
parseAndConstructSelectIntoOperator(ctx);
}
public void parseSelectClause(SelectClauseContext ctx) {
@@ -1101,6 +1105,75 @@ public class IoTDBSqlVisitor extends
SqlBaseBaseVisitor<Operator> {
resultColumnContext.AS() == null ? null :
resultColumnContext.ID().getText());
}
+ private SelectIntoOperator
parseAndConstructSelectIntoOperator(SelectStatementContext ctx) {
+ if (queryOp.getFromComponent().getPrefixPaths().size() != 1) {
+ throw new SQLParserException(
+ "select into: the number of prefix paths in the from clause should
be 1.");
+ }
+
+ int sourcePathsCount =
queryOp.getSelectComponent().getResultColumns().size();
+ if (sourcePathsCount != ctx.intoClause().intoPath().size()) {
+ throw new SQLParserException(
+ "select into: the number of source paths and the number of target
paths should be the same.");
+ }
+
+ SelectIntoOperator selectIntoOperator = new SelectIntoOperator();
+ selectIntoOperator.setQueryOperator(queryOp);
+ List<PartialPath> intoPaths = new ArrayList<>();
+ for (int i = 0; i < sourcePathsCount; ++i) {
+ intoPaths.add(parseIntoPath(ctx.intoClause().intoPath(i)));
+ }
+ selectIntoOperator.setIntoPaths(intoPaths);
+ return selectIntoOperator;
+ }
+
+ private PartialPath parseIntoPath(IntoPathContext intoPathContext) {
+ int levelLimitOfSourcePrefixPath =
+ queryOp.getSpecialClauseComponent() != null
+ ? queryOp.getSpecialClauseComponent().getLevel()
+ : -1;
+ if (levelLimitOfSourcePrefixPath == -1) {
+ levelLimitOfSourcePrefixPath =
+ queryOp.getFromComponent().getPrefixPaths().get(0).getNodeLength() -
1;
+ }
+
+ PartialPath intoPath = null;
+ if (intoPathContext.fullPath() != null) {
+ intoPath = parseFullPath(intoPathContext.fullPath());
+
+ Matcher m = leveledPathNodePattern.matcher(intoPath.getFullPath());
+ while (m.find()) {
+ String param = m.group();
+ int nodeIndex = 0;
+ try {
+ nodeIndex = Integer.parseInt(param.substring(2, param.length() -
1).trim());
+ } catch (NumberFormatException e) {
+ throw new SQLParserException("the x of ${x} should be an integer.");
+ }
+ if (nodeIndex < 1 || levelLimitOfSourcePrefixPath < nodeIndex) {
+ throw new SQLParserException(
+ "the x of ${x} should be greater than 0 and equal to or less
than <level> or the length of queried path prefix.");
+ }
+ }
+ } else if (intoPathContext.nodeNameWithoutStar() != null) {
+ List<NodeNameWithoutStarContext> nodeNameWithoutStars =
intoPathContext.nodeNameWithoutStar();
+ String[] intoPathNodes =
+ new String[1 + levelLimitOfSourcePrefixPath +
nodeNameWithoutStars.size()];
+
+ intoPathNodes[0] = "root";
+ for (int i = 1; i <= levelLimitOfSourcePrefixPath; ++i) {
+ intoPathNodes[i] = "${" + i + "}";
+ }
+ for (int i = 1; i <= nodeNameWithoutStars.size(); ++i) {
+ intoPathNodes[levelLimitOfSourcePrefixPath + i] =
nodeNameWithoutStars.get(i - 1).getText();
+ }
+
+ intoPath = new PartialPath(intoPathNodes);
+ }
+
+ return intoPath;
+ }
+
@Override
public Operator
visitDropContinuousQueryStatement(DropContinuousQueryStatementContext ctx) {
DropContinuousQueryOperator dropContinuousQueryOperator =
@@ -1209,54 +1282,18 @@ public class IoTDBSqlVisitor extends
SqlBaseBaseVisitor<Operator> {
}
if (queryOp.getFromComponent().getPrefixPaths().size() > 1) {
- throw new SQLParserException("CQ: CQ currently does not support multiple
series .");
+ throw new SQLParserException("CQ: CQ currently does not support multiple
series.");
}
parseCqGroupByTimeClause(ctx.cqGroupByTimeClause());
- int fromLen =
queryOp.getFromComponent().getPrefixPaths().get(0).getNodeLength();
- int queryLevel = queryOp.getSpecialClauseComponent().getLevel();
- if (queryLevel >= fromLen) {
+ int groupByQueryLevel = queryOp.getSpecialClauseComponent().getLevel();
+ int fromPrefixLevelLimit =
queryOp.getFromComponent().getPrefixPaths().get(0).getNodeLength();
+ if (groupByQueryLevel >= fromPrefixLevelLimit) {
throw new SQLParserException("CQ: Level should not exceed the
<from_prefix> length.");
}
- PartialPath targetPath = null;
-
- int trueLevel = queryLevel;
- if (trueLevel == -1) {
- trueLevel = fromLen - 1;
- }
-
- if (ctx.fullPath() != null) {
- targetPath = parseFullPath(ctx.fullPath());
- Matcher m = cqLevelNodePattern.matcher(targetPath.getFullPath());
- while (m.find()) {
- String param = m.group();
- int nodeIndex = 0;
- try {
- nodeIndex = Integer.parseInt(param.substring(2, param.length() -
1).trim());
- } catch (NumberFormatException e) {
- throw new SQLParserException("CQ: x of ${x} should be an integer.");
- }
- if (nodeIndex < 1 || nodeIndex > trueLevel) {
- throw new SQLParserException(
- "CQ: x of ${x} should be greater than 0 and equal to or less
than <level> or the length of queried path prefix.");
- }
- }
- } else if (ctx.nodeNameWithoutStar() != null) {
-
- List<String> targetNodes = new ArrayList<>();
-
- targetNodes.add("root");
-
- for (int i = 1; i <= trueLevel; i++) {
- targetNodes.add("${" + i + "}");
- }
- targetNodes.add(ctx.nodeNameWithoutStar().getText());
- targetPath = new PartialPath(targetNodes.toArray(new String[0]));
- }
-
- createContinuousQueryOperator.setTargetPath(targetPath);
+ createContinuousQueryOperator.setTargetPath(parseIntoPath(ctx.intoPath()));
createContinuousQueryOperator.setQueryOperator(queryOp);
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalChecker.java
b/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalChecker.java
index 23f6a6f..14234d5 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalChecker.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalChecker.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.qp.strategy;
import org.apache.iotdb.db.exception.query.LogicalOperatorException;
import org.apache.iotdb.db.qp.logical.Operator;
import org.apache.iotdb.db.qp.logical.crud.QueryOperator;
+import org.apache.iotdb.db.qp.logical.crud.SelectIntoOperator;
public class LogicalChecker {
@@ -32,5 +33,9 @@ public class LogicalChecker {
if (operator instanceof QueryOperator) {
((QueryOperator) operator).check();
}
+
+ if (operator instanceof SelectIntoOperator) {
+ ((SelectIntoOperator) operator).check();
+ }
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/control/QueryTimeManager.java
b/server/src/main/java/org/apache/iotdb/db/query/control/QueryTimeManager.java
index 3d78a92..48bb732 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/control/QueryTimeManager.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/control/QueryTimeManager.java
@@ -22,6 +22,8 @@ import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.query.QueryTimeoutRuntimeException;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.sys.ShowQueryProcesslistPlan;
import org.apache.iotdb.db.service.IService;
import org.apache.iotdb.db.service.ServiceType;
@@ -76,6 +78,14 @@ public class QueryTimeManager implements IService {
queryScheduledTaskMap.put(queryId, scheduledFuture);
}
+ public void registerQuery(
+ long queryId, long startTime, String sql, long timeout, PhysicalPlan
plan) {
+ if (plan instanceof ShowQueryProcesslistPlan) {
+ return;
+ }
+ registerQuery(queryId, startTime, sql, timeout);
+ }
+
public void killQuery(long queryId) {
if (queryInfoMap.get(queryId) == null) {
return;
@@ -101,6 +111,10 @@ public class QueryTimeManager implements IService {
return successRemoved;
}
+ public AtomicBoolean unRegisterQuery(long queryId, PhysicalPlan plan) {
+ return plan instanceof ShowQueryProcesslistPlan ? null :
unRegisterQuery(queryId);
+ }
+
public static void checkQueryAlive(long queryId) {
QueryInfo queryInfo = getInstance().queryInfoMap.get(queryId);
if (queryInfo != null && queryInfo.isInterrupted()) {
diff --git
a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
index cc78805..79311b6 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.cost.statistic.Measurement;
import org.apache.iotdb.db.cost.statistic.Operation;
+import org.apache.iotdb.db.engine.selectinto.InsertTabletPlansIterator;
import org.apache.iotdb.db.exception.BatchProcessException;
import org.apache.iotdb.db.exception.IoTDBException;
import org.apache.iotdb.db.exception.QueryInBatchStatementException;
@@ -50,7 +51,6 @@ import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan;
import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan.MeasurementType;
import org.apache.iotdb.db.qp.physical.crud.CreateTemplatePlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
-import org.apache.iotdb.db.qp.physical.crud.GroupByTimePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertMultiTabletPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowsOfOneDevicePlan;
@@ -58,6 +58,7 @@ import org.apache.iotdb.db.qp.physical.crud.InsertRowsPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
+import org.apache.iotdb.db.qp.physical.crud.SelectIntoPlan;
import org.apache.iotdb.db.qp.physical.crud.SetDeviceTemplatePlan;
import org.apache.iotdb.db.qp.physical.crud.UDFPlan;
import org.apache.iotdb.db.qp.physical.crud.UDTFPlan;
@@ -142,7 +143,6 @@ import java.sql.SQLException;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Calendar;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@@ -172,30 +172,25 @@ public class TSServiceImpl implements TSIService.Iface {
private static final String INFO_QUERY_PROCESS_ERROR = "Error occurred in
query process: ";
private static final String INFO_NOT_ALLOWED_IN_BATCH_ERROR =
"The query statement is not allowed in batch: ";
-
private static final String INFO_INTERRUPT_ERROR =
"Current Thread interrupted when dealing with request {}";
+ public static final TSProtocolVersion CURRENT_RPC_VERSION =
+ TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V3;
+
private static final int MAX_SIZE =
IoTDBDescriptor.getInstance().getConfig().getQueryCacheSizeInMetric();
private static final int DELETE_SIZE = 20;
- private static final int DEFAULT_FETCH_SIZE = 10000;
- private static final long MS_TO_MONTH = 30 * 86400_000L;
private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
- private final boolean enableMetric = config.isEnableMetricService();
private static final List<SqlArgument> sqlArgumentList = new
ArrayList<>(MAX_SIZE);
- protected Planner processor;
- protected IPlanExecutor executor;
-
- private final SessionManager sessionManager = SessionManager.getInstance();
+ private static final AtomicInteger queryCount = new AtomicInteger(0);
private final QueryTimeManager queryTimeManager =
QueryTimeManager.getInstance();
+ private final SessionManager sessionManager = SessionManager.getInstance();
- public static final TSProtocolVersion CURRENT_RPC_VERSION =
- TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V3;
-
- private static final AtomicInteger queryCount = new AtomicInteger(0);
+ protected Planner processor;
+ protected IPlanExecutor executor;
public TSServiceImpl() throws QueryProcessException {
processor = new Planner();
@@ -486,7 +481,7 @@ public class TSServiceImpl implements TSIService.Iface {
InsertRowsPlan insertRowsPlan;
int index = 0;
- List executeList = new ArrayList();
+ List<Object> executeList = new ArrayList<>();
OperatorType lastOperatorType = null;
CreateMultiTimeSeriesPlan multiPlan;
for (int i = 0; i < req.getStatements().size(); i++) {
@@ -494,7 +489,7 @@ public class TSServiceImpl implements TSIService.Iface {
try {
PhysicalPlan physicalPlan =
processor.parseSQLToPhysicalPlan(statement,
sessionManager.getZoneId(req.sessionId));
- if (physicalPlan.isQuery()) {
+ if (physicalPlan.isQuery() || physicalPlan.isSelectInto()) {
throw new QueryInBatchStatementException(statement);
}
@@ -548,14 +543,14 @@ public class TSServiceImpl implements TSIService.Iface {
}
} else {
lastOperatorType = physicalPlan.getOperatorType();
- if (executeList.size() > 0) {
+ if (!executeList.isEmpty()) {
if (!executeBatchList(executeList, result)) {
isAllSuccessful = false;
}
executeList.clear();
}
long t2 = System.currentTimeMillis();
- TSExecuteStatementResp resp = executeUpdateStatement(physicalPlan,
req.getSessionId());
+ TSExecuteStatementResp resp = executeNonQueryStatement(physicalPlan,
req.getSessionId());
Measurement.INSTANCE.addOperationLatency(Operation.EXECUTE_ONE_SQL_IN_BATCH,
t2);
result.add(resp.status);
if (resp.getStatus().code !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
@@ -599,9 +594,15 @@ public class TSServiceImpl implements TSIService.Iface {
physicalPlan,
req.fetchSize,
req.timeout,
- sessionManager.getUsername(req.getSessionId()),
+ req.getSessionId(),
req.isEnableRedirectQuery())
- : executeUpdateStatement(physicalPlan, req.getSessionId());
+ : executeUpdateStatement(
+ statement,
+ req.statementId,
+ physicalPlan,
+ req.fetchSize,
+ req.timeout,
+ req.getSessionId());
} catch (InterruptedException e) {
LOGGER.error(INFO_INTERRUPT_ERROR, req, e);
Thread.currentThread().interrupt();
@@ -629,7 +630,7 @@ public class TSServiceImpl implements TSIService.Iface {
physicalPlan,
req.fetchSize,
req.timeout,
- sessionManager.getUsername(req.getSessionId()),
+ req.getSessionId(),
req.isEnableRedirectQuery())
: RpcUtils.getTSExecuteStatementResp(
TSStatusCode.EXECUTE_STATEMENT_ERROR, "Statement is not a query
statement.");
@@ -660,7 +661,7 @@ public class TSServiceImpl implements TSIService.Iface {
physicalPlan,
req.fetchSize,
config.getQueryTimeoutThreshold(),
- sessionManager.getUsername(req.sessionId),
+ req.sessionId,
req.isEnableRedirectQuery())
: RpcUtils.getTSExecuteStatementResp(
TSStatusCode.EXECUTE_STATEMENT_ERROR, "Statement is not a query
statement.");
@@ -676,7 +677,7 @@ public class TSServiceImpl implements TSIService.Iface {
}
@Override
- public TSExecuteStatementResp executeLastDataQuery(TSLastDataQueryReq req)
throws TException {
+ public TSExecuteStatementResp executeLastDataQuery(TSLastDataQueryReq req) {
try {
if (!checkLogin(req.getSessionId())) {
return RpcUtils.getTSExecuteStatementResp(getNotLoggedInStatus());
@@ -691,7 +692,7 @@ public class TSServiceImpl implements TSIService.Iface {
physicalPlan,
req.fetchSize,
config.getQueryTimeoutThreshold(),
- sessionManager.getUsername(req.sessionId),
+ req.sessionId,
req.isEnableRedirectQuery())
: RpcUtils.getTSExecuteStatementResp(
TSStatusCode.EXECUTE_STATEMENT_ERROR, "Statement is not a query
statement.");
@@ -717,7 +718,7 @@ public class TSServiceImpl implements TSIService.Iface {
PhysicalPlan plan,
int fetchSize,
long timeout,
- String username,
+ long sessionId,
boolean enableRedirect)
throws QueryProcessException, SQLException, StorageEngineException,
QueryFilterOptimizationException, MetadataException, IOException,
InterruptedException,
@@ -725,16 +726,12 @@ public class TSServiceImpl implements TSIService.Iface {
queryCount.incrementAndGet();
AUDIT_LOGGER.debug(
"Session {} execute Query: {}", sessionManager.getCurrSessionId(),
statement);
- long startTime = System.currentTimeMillis();
- long queryId = -1;
- try {
- // generate the queryId for the operation
- queryId = sessionManager.requestQueryId(statementId, true);
- // register query info to queryTimeManager
- if (!(plan instanceof ShowQueryProcesslistPlan)) {
- queryTimeManager.registerQuery(queryId, startTime, statement, timeout);
- }
+ final long startTime = System.currentTimeMillis();
+ final long queryId = sessionManager.requestQueryId(statementId, true);
+
+ try {
+ queryTimeManager.registerQuery(queryId, startTime, statement, timeout,
plan);
if (plan instanceof QueryPlan && config.isEnablePerformanceTracing()) {
TracingManager tracingManager = TracingManager.getInstance();
if (!(plan instanceof AlignByDevicePlan)) {
@@ -744,9 +741,8 @@ public class TSServiceImpl implements TSIService.Iface {
}
}
- if (plan instanceof AuthorPlan) {
- plan.setLoginUserName(username);
- }
+ String username = sessionManager.getUsername(sessionId);
+ plan.setLoginUserName(username);
TSExecuteStatementResp resp = null;
// execute it before createDataSet since it may change the content of
query plan
@@ -810,14 +806,14 @@ public class TSServiceImpl implements TSIService.Iface {
}
}
}
+
resp.setQueryId(queryId);
if (plan instanceof AlignByDevicePlan &&
config.isEnablePerformanceTracing()) {
TracingManager.getInstance()
.writePathsNum(queryId, ((AlignByDeviceDataSet)
newDataSet).getPathsNum());
}
-
- if (enableMetric) {
+ if (config.isEnableMetricService()) {
long endTime = System.currentTimeMillis();
SqlArgument sqlArgument = new SqlArgument(resp, plan, statement,
startTime, endTime);
synchronized (sqlArgumentList) {
@@ -827,11 +823,8 @@ public class TSServiceImpl implements TSIService.Iface {
}
}
}
+ queryTimeManager.unRegisterQuery(queryId, plan);
- // remove query info in QueryTimeManager
- if (!(plan instanceof ShowQueryProcesslistPlan)) {
- queryTimeManager.unRegisterQuery(queryId);
- }
return resp;
} catch (Exception e) {
sessionManager.releaseQueryResourceNoExceptions(queryId);
@@ -845,21 +838,6 @@ public class TSServiceImpl implements TSIService.Iface {
}
}
- /*
- calculate fetch size for group by time plan
- */
- private int getFetchSizeForGroupByTimePlan(GroupByTimePlan plan) {
- int rows = (int) ((plan.getEndTime() - plan.getStartTime()) /
plan.getInterval());
- // rows gets 0 is caused by: the end time - the start time < the time
interval.
- if (rows == 0 && plan.isIntervalByMonth()) {
- Calendar calendar = Calendar.getInstance();
- calendar.setTimeInMillis(plan.getStartTime());
- calendar.add(Calendar.MONTH, (int) (plan.getInterval() / MS_TO_MONTH));
- rows = calendar.getTimeInMillis() <= plan.getEndTime() ? 1 : 0;
- }
- return rows;
- }
-
private TSExecuteStatementResp getListDataSetHeaders(QueryDataSet dataSet) {
return StaticResps.getNoTimeExecuteResp(
dataSet.getPaths().stream().map(Path::getFullPath).collect(Collectors.toList()),
@@ -998,6 +976,81 @@ public class TSServiceImpl implements TSIService.Iface {
plan.setPaths(null);
}
+ private TSExecuteStatementResp executeSelectIntoStatement(
+ String statement,
+ long statementId,
+ PhysicalPlan physicalPlan,
+ int fetchSize,
+ long timeout,
+ long sessionId)
+ throws IoTDBException, TException, SQLException, IOException,
InterruptedException,
+ QueryFilterOptimizationException {
+ TSStatus status = checkAuthority(physicalPlan, sessionId);
+ if (status != null) {
+ return new TSExecuteStatementResp(status);
+ }
+
+ final long startTime = System.currentTimeMillis();
+ final long queryId = sessionManager.requestQueryId(statementId, true);
+ final SelectIntoPlan selectIntoPlan = (SelectIntoPlan) physicalPlan;
+ final QueryPlan queryPlan = selectIntoPlan.getQueryPlan();
+
+ queryCount.incrementAndGet();
+ AUDIT_LOGGER.debug(
+ "Session {} execute select into: {}",
sessionManager.getCurrSessionId(), statement);
+ if (config.isEnablePerformanceTracing()) {
+ TracingManager.getInstance()
+ .writeQueryInfo(queryId, statement, startTime,
queryPlan.getPaths().size());
+ }
+
+ try {
+ queryTimeManager.registerQuery(queryId, startTime, statement, timeout,
queryPlan);
+
+ InsertTabletPlansIterator insertTabletPlansIterator =
+ new InsertTabletPlansIterator(
+ queryPlan,
+ createQueryDataSet(queryId, queryPlan, fetchSize),
+ selectIntoPlan.getFromPath(),
+ selectIntoPlan.getIntoPaths());
+ while (insertTabletPlansIterator.hasNext()) {
+ TSStatus executionStatus =
+ insertTabletsInternally(insertTabletPlansIterator.next(),
sessionId);
+ if (executionStatus.getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ && executionStatus.getCode() !=
TSStatusCode.NEED_REDIRECTION.getStatusCode()) {
+ return
RpcUtils.getTSExecuteStatementResp(executionStatus).setQueryId(queryId);
+ }
+ }
+
+ return
RpcUtils.getTSExecuteStatementResp(TSStatusCode.SUCCESS_STATUS).setQueryId(queryId);
+ } catch (Exception e) {
+ sessionManager.releaseQueryResourceNoExceptions(queryId);
+ throw e;
+ } finally {
+ queryTimeManager.unRegisterQuery(queryId, queryPlan);
+ Measurement.INSTANCE.addOperationLatency(Operation.EXECUTE_QUERY,
startTime);
+ long costTime = System.currentTimeMillis() - startTime;
+ if (costTime >= config.getSlowQueryThreshold()) {
+ SLOW_SQL_LOGGER.info("Cost: {} ms, sql is {}", costTime, statement);
+ }
+ }
+ }
+
+ private TSStatus insertTabletsInternally(
+ List<InsertTabletPlan> insertTabletPlans, long sessionId) {
+ InsertMultiTabletPlan insertMultiTabletPlan = new InsertMultiTabletPlan();
+ for (int i = 0; i < insertTabletPlans.size(); i++) {
+ InsertTabletPlan insertTabletPlan = insertTabletPlans.get(i);
+ TSStatus status = checkAuthority(insertTabletPlan, sessionId);
+ if (status != null) {
+ // not authorized
+ insertMultiTabletPlan.getResults().put(i, status);
+ }
+ }
+ insertMultiTabletPlan.setInsertTabletPlanList(insertTabletPlans);
+
+ return executeNonQueryPlan(insertMultiTabletPlan);
+ }
+
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity
warning
@Override
public TSFetchResultsResp fetchResults(TSFetchResultsReq req) {
@@ -1120,6 +1173,7 @@ public class TSServiceImpl implements TSIService.Iface {
return new QueryContext(queryId, debug);
}
+ /** update statement can be: 1. select-into statement 2. non-query statement
*/
@Override
public TSExecuteStatementResp executeUpdateStatement(TSExecuteStatementReq
req) {
if (!checkLogin(req.getSessionId())) {
@@ -1127,41 +1181,48 @@ public class TSServiceImpl implements TSIService.Iface {
}
try {
- return executeUpdateStatement(req.getStatement(), req.getSessionId());
+ PhysicalPlan physicalPlan =
+ processor.parseSQLToPhysicalPlan(req.statement,
sessionManager.getZoneId(req.sessionId));
+ return physicalPlan.isQuery()
+ ? RpcUtils.getTSExecuteStatementResp(
+ TSStatusCode.EXECUTE_STATEMENT_ERROR, "Statement is a query
statement.")
+ : executeUpdateStatement(
+ req.statement,
+ req.statementId,
+ physicalPlan,
+ req.fetchSize,
+ req.timeout,
+ req.getSessionId());
+ } catch (InterruptedException e) {
+ LOGGER.error(INFO_INTERRUPT_ERROR, req, e);
+ Thread.currentThread().interrupt();
+ return RpcUtils.getTSExecuteStatementResp(onQueryException(e, "executing
update statement"));
} catch (Exception e) {
return RpcUtils.getTSExecuteStatementResp(onQueryException(e, "executing
update statement"));
}
}
- private TSExecuteStatementResp executeUpdateStatement(PhysicalPlan plan,
long sessionId) {
- TSStatus status = checkAuthority(plan, sessionId);
- if (status != null) {
- return new TSExecuteStatementResp(status);
- }
-
- status = executeNonQueryPlan(plan);
- TSExecuteStatementResp resp = RpcUtils.getTSExecuteStatementResp(status);
- long queryId = sessionManager.requestQueryId(false);
- return resp.setQueryId(queryId);
- }
-
- private boolean executeNonQuery(PhysicalPlan plan)
- throws QueryProcessException, StorageGroupNotSetException,
StorageEngineException {
- if (IoTDBDescriptor.getInstance().getConfig().isReadOnly()) {
- throw new QueryProcessException(
- "Current system mode is read-only, does not support non-query
operation");
- }
- return executor.processNonQuery(plan);
+ /** update statement can be: 1. select-into statement 2. non-query statement
*/
+ private TSExecuteStatementResp executeUpdateStatement(
+ String statement,
+ long statementId,
+ PhysicalPlan plan,
+ int fetchSize,
+ long timeout,
+ long sessionId)
+ throws TException, SQLException, IoTDBException, IOException,
InterruptedException,
+ QueryFilterOptimizationException {
+ return plan.isSelectInto()
+ ? executeSelectIntoStatement(statement, statementId, plan, fetchSize,
timeout, sessionId)
+ : executeNonQueryStatement(plan, sessionId);
}
- private TSExecuteStatementResp executeUpdateStatement(String statement, long
sessionId)
- throws QueryProcessException {
- PhysicalPlan physicalPlan =
- processor.parseSQLToPhysicalPlan(statement,
sessionManager.getZoneId(sessionId));
- return physicalPlan.isQuery()
- ? RpcUtils.getTSExecuteStatementResp(
- TSStatusCode.EXECUTE_STATEMENT_ERROR, "Statement is a query
statement.")
- : executeUpdateStatement(physicalPlan, sessionId);
+ private TSExecuteStatementResp executeNonQueryStatement(PhysicalPlan plan,
long sessionId) {
+ TSStatus status = checkAuthority(plan, sessionId);
+ return status != null
+ ? new TSExecuteStatementResp(status)
+ : RpcUtils.getTSExecuteStatementResp(executeNonQueryPlan(plan))
+ .setQueryId(sessionManager.requestQueryId(false));
}
/**
@@ -1567,7 +1628,7 @@ public class TSServiceImpl implements TSIService.Iface {
return getNotLoggedInStatus();
}
- return insertTabletsInternal(req);
+ return insertTabletsInternally(req);
} catch (NullPointerException e) {
LOGGER.error("{}: error occurs when insertTablets",
IoTDBConstant.GLOBAL_DB_NAME, e);
return RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR);
@@ -1600,7 +1661,7 @@ public class TSServiceImpl implements TSIService.Iface {
}
/** construct one InsertMultiTabletPlan and process it */
- public TSStatus insertTabletsInternal(TSInsertTabletsReq req) throws
IllegalPathException {
+ public TSStatus insertTabletsInternally(TSInsertTabletsReq req) throws
IllegalPathException {
List<InsertTabletPlan> insertTabletPlanList = new ArrayList<>();
InsertMultiTabletPlan insertMultiTabletPlan = new InsertMultiTabletPlan();
for (int i = 0; i < req.deviceIds.size(); i++) {
@@ -1962,6 +2023,15 @@ public class TSServiceImpl implements TSIService.Iface {
: RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR);
}
+ private boolean executeNonQuery(PhysicalPlan plan)
+ throws QueryProcessException, StorageGroupNotSetException,
StorageEngineException {
+ if (IoTDBDescriptor.getInstance().getConfig().isReadOnly()) {
+ throw new QueryProcessException(
+ "Current system mode is read-only, does not support non-query
operation");
+ }
+ return executor.processNonQuery(plan);
+ }
+
protected List<TSDataType> getSeriesTypesByPaths(
List<PartialPath> paths, List<String> aggregations) throws
MetadataException {
return SchemaUtils.getSeriesTypesByPaths(paths, aggregations);
diff --git
a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBSelectIntoIT.java
b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBSelectIntoIT.java
new file mode 100644
index 0000000..be5a23b
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBSelectIntoIT.java
@@ -0,0 +1,564 @@
+/*
+ * 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.db.integration;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.service.IoTDB;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.jdbc.Config;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+// TODO: check null values
+public class IoTDBSelectIntoIT {
+
+ private static final int ROW_LIMIT =
+
IoTDBDescriptor.getInstance().getConfig().getSelectIntoInsertTabletPlanRowLimit();
+
+ private static final String[] INSERTION_SQLS = {
+ "insert into root.sg.d1(time, s2, s3, s4, s5, s6) values (0, 0, 0, 0,
true, '0')",
+ "insert into root.sg.d1(time, s1, s3, s4, s5, s6) values (1, 1, 1, 1,
false, '1')",
+ "insert into root.sg.d1(time, s1, s2, s4, s5, s6) values (2, 2, 2, 2,
true, '2')",
+ "insert into root.sg.d1(time, s1, s2, s3, s5, s6) values (3, 3, 3, 3,
false, '3')",
+ "insert into root.sg.d1(time, s1, s2, s3, s4, s5, s6) values (4, 4, 4, 4,
4, true, '4')",
+ };
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvironmentUtils.envSetUp();
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ createTimeSeries();
+ generateData();
+ }
+
+ private static void createTimeSeries() throws MetadataException {
+ IoTDB.metaManager.setStorageGroup(new PartialPath("root.sg"));
+
+ IoTDB.metaManager.createTimeseries(
+ new PartialPath("root.sg.d1.s1"),
+ TSDataType.INT32,
+ TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED,
+ null);
+ IoTDB.metaManager.createTimeseries(
+ new PartialPath("root.sg.d1.s2"),
+ TSDataType.INT64,
+ TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED,
+ null);
+ IoTDB.metaManager.createTimeseries(
+ new PartialPath("root.sg.d1.s3"),
+ TSDataType.FLOAT,
+ TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED,
+ null);
+ IoTDB.metaManager.createTimeseries(
+ new PartialPath("root.sg.d1.s4"),
+ TSDataType.DOUBLE,
+ TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED,
+ null);
+ IoTDB.metaManager.createTimeseries(
+ new PartialPath("root.sg.d1.s5"),
+ TSDataType.BOOLEAN,
+ TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED,
+ null);
+ IoTDB.metaManager.createTimeseries(
+ new PartialPath("root.sg.d1.s6"),
+ TSDataType.TEXT,
+ TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED,
+ null);
+
+ IoTDB.metaManager.createTimeseries(
+ new PartialPath("root.sg.d1.empty"),
+ TSDataType.TEXT,
+ TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED,
+ null);
+
+ IoTDB.metaManager.createTimeseries(
+ new PartialPath("root.sg.d2.s1"),
+ TSDataType.INT32,
+ TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED,
+ null);
+ }
+
+ private static void generateData() {
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ for (String dataGenerationSql : INSERTION_SQLS) {
+ statement.execute(dataGenerationSql);
+ }
+
+ statement.execute("insert into root.sg.d2(time, s1) values (0, 0)");
+
+ final int size = ROW_LIMIT + 1;
+ for (int i = 0; i < size; ++i) {
+ statement.execute(String.format("insert into root.sg.d3(time, s1)
values (%d, %d)", i, i));
+ }
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvironmentUtils.cleanEnv();
+ }
+
+ @Test // TODO: check values
+ public void selectIntoSameDevice() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute(
+ "select s1, s2, s3, s4, s5, s6 into s7, s8, s9, s10, s11, s12 from
root.sg.d1");
+
+ try (ResultSet resultSet =
+ statement.executeQuery("select s7, s8, s9, s10, s11, s12 from
root.sg.d1")) {
+ assertEquals(1 + 6, resultSet.getMetaData().getColumnCount());
+
+ for (int i = 0; i < INSERTION_SQLS.length; ++i) {
+ assertTrue(resultSet.next());
+ StringBuilder stringBuilder = new StringBuilder();
+ for (int j = 0; j < 6 + 1; ++j) {
+ stringBuilder.append(resultSet.getString(j + 1)).append(',');
+ }
+ System.out.println(stringBuilder.toString());
+ }
+
+ assertFalse(resultSet.next());
+ }
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+ }
+
+ @Test // TODO: check values
+ public void selectIntoDifferentDevices() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute(
+ "select s1, s2, s3, s4, s5, s6 into pre_${2}_suf.s1,
pre_${2}_suf.s2, pre_${2}_suf.s3, pre_${2}_suf.s4, pre_${2}_suf.s5,
pre_${2}_suf.s6 from root.sg.d1");
+
+ try (ResultSet resultSet =
+ statement.executeQuery(
+ "select pre_d1_suf.s1, pre_d1_suf.s2, pre_d1_suf.s3,
pre_d1_suf.s4, pre_d1_suf.s5, pre_d1_suf.s6 from root.sg.d1")) {
+ assertEquals(1 + 6, resultSet.getMetaData().getColumnCount());
+
+ for (int i = 0; i < INSERTION_SQLS.length; ++i) {
+ assertTrue(resultSet.next());
+ StringBuilder stringBuilder = new StringBuilder();
+ for (int j = 0; j < 6 + 1; ++j) {
+ stringBuilder.append(resultSet.getString(j + 1)).append(',');
+ }
+ System.out.println(stringBuilder.toString());
+ }
+
+ assertFalse(resultSet.next());
+ }
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+ }
+
+ @Test
+ public void selectFromEmptySourcePath() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select empty into target from root.sg.d1");
+
+ try (ResultSet resultSet = statement.executeQuery("select target from
root.sg.d1")) {
+ assertEquals(1, resultSet.getMetaData().getColumnCount());
+ assertFalse(resultSet.next());
+ }
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+ }
+
+ @Test
+ public void selectIntoFullTargetPath() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select s1 into root.${2}.${1}.s1 from root.sg.d1
where time>0");
+
+ try (ResultSet resultSet = statement.executeQuery("select sg.d1.s1,
d1.sg.s1 from root")) {
+ assertEquals(1 + 2, resultSet.getMetaData().getColumnCount());
+
+ for (int i = 0; i < INSERTION_SQLS.length - 1; ++i) {
+ assertTrue(resultSet.next());
+ assertEquals(resultSet.getString(1), String.valueOf(i + 1));
+ assertEquals(resultSet.getString(2), resultSet.getString(3));
+ }
+
+ assertFalse(resultSet.next());
+ }
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+ }
+
+ @Test
+ public void selectSameTimeSeries() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select s1, s1 into s1s2, s1s3 from root.sg.d1");
+
+ try (ResultSet resultSet = statement.executeQuery("select s1s2, s1s3
from root.sg.d1")) {
+ assertEquals(1 + 2, resultSet.getMetaData().getColumnCount());
+
+ for (int i = 1; i < INSERTION_SQLS.length; ++i) {
+ assertTrue(resultSet.next());
+ for (int j = 0; j < 2 + 1; ++j) {
+ assertEquals(resultSet.getString(2), resultSet.getString(3));
+ }
+ }
+
+ assertFalse(resultSet.next());
+ }
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+ }
+
+ @Test
+ public void testLargeData() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select s1 into large_s1 from root.sg.d3");
+
+ try (ResultSet resultSet = statement.executeQuery("select large_s1 from
root.sg.d3")) {
+ assertEquals(1 + 1, resultSet.getMetaData().getColumnCount());
+
+ final int size = ROW_LIMIT + 1;
+ for (int i = 0; i < size; ++i) {
+ assertTrue(resultSet.next());
+ assertEquals(
+ Double.parseDouble(resultSet.getString(1)),
+ Double.parseDouble(resultSet.getString(2)),
+ 0);
+ }
+ assertFalse(resultSet.next());
+ }
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+ }
+
+ @Test
+ public void testUDFQuery() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute(
+ "select s1, sin(s1), s1 + s1 into ${2}.s2, ${2}.s3, ${2}.s4 from
root.sg.d1");
+
+ try (ResultSet resultSet = statement.executeQuery("select s2, s3, s4
from root.sg.d1.d1")) {
+ assertEquals(1 + 3, resultSet.getMetaData().getColumnCount());
+
+ for (int i = 1; i < INSERTION_SQLS.length; ++i) {
+ assertTrue(resultSet.next());
+ for (int j = 0; j < 2 + 1; ++j) {
+ double s2 = Double.parseDouble(resultSet.getString(2));
+ double s3 = Double.parseDouble(resultSet.getString(3));
+ double s4 = Double.parseDouble(resultSet.getString(4));
+ assertEquals(i, s2, 0);
+ assertEquals(Math.sin(i), s3, 0);
+ assertEquals((double) i + (double) i, s4, 0);
+ }
+ }
+
+ assertFalse(resultSet.next());
+ }
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+ }
+
+ @Test
+ public void testGroupByQuery() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select count(s1) into count_s1 from root.sg.d1 group
by ([1, 5),1ms);");
+
+ try (ResultSet resultSet = statement.executeQuery("select count_s1 from
root.sg.d1")) {
+ assertEquals(1 + 1, resultSet.getMetaData().getColumnCount());
+
+ for (int i = 1; i < INSERTION_SQLS.length; ++i) {
+ assertTrue(resultSet.next());
+ for (int j = 0; j < 1 + 1; ++j) {
+ assertEquals(String.valueOf(i), resultSet.getString(1));
+ assertEquals("1", resultSet.getString(2));
+ }
+ }
+
+ assertFalse(resultSet.next());
+ }
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+ }
+
+ @Test
+ public void testGroupByFillQuery() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute(
+ "select last_value(s1) into gbf_s1 from root.sg.d1 group by ([1,
10),1ms) fill (float[PREVIOUS]);");
+
+ try (ResultSet resultSet = statement.executeQuery("select gbf_s1 from
root.sg.d1")) {
+ assertEquals(1 + 1, resultSet.getMetaData().getColumnCount());
+
+ for (int i = 1; i < 10; ++i) {
+ assertTrue(resultSet.next());
+ for (int j = 0; j < 1 + 1; ++j) {
+ assertEquals(String.valueOf(i), resultSet.getString(1));
+ assertEquals(i < 5 ? String.valueOf(i) : "0",
resultSet.getString(2));
+ }
+ }
+
+ assertFalse(resultSet.next());
+ }
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+ }
+
+ @Test
+ public void testFillQuery() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute(
+ "select s1 into fill_s1 from root.sg.d1 where time = 10 fill(float
[linear, 1ms, 1ms])");
+
+ try (ResultSet resultSet = statement.executeQuery("select fill_s1 from
root.sg.d1")) {
+ assertEquals(1 + 1, resultSet.getMetaData().getColumnCount());
+
+ assertTrue(resultSet.next());
+ assertEquals("10", resultSet.getString(1));
+ assertEquals("4", resultSet.getString(2));
+
+ assertFalse(resultSet.next());
+ }
+ } catch (SQLException throwable) {
+ fail(throwable.getMessage());
+ }
+ }
+
+ @Test
+ public void testDifferentNumbersOfSourcePathsAndTargetPaths() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select s1, s2 into target from root.sg.d1");
+ fail();
+ } catch (SQLException throwable) {
+ assertTrue(
+ throwable
+ .getMessage()
+ .contains(
+ "the number of source paths and the number of target paths
should be the same"));
+ }
+
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select s1 into target from root.sg.*");
+ fail();
+ } catch (SQLException throwable) {
+ assertTrue(
+ throwable
+ .getMessage()
+ .contains(
+ "the number of source paths and the number of target paths
should be the same"));
+ }
+
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select * into target from root.sg.d1");
+ fail();
+ } catch (SQLException throwable) {
+ assertTrue(
+ throwable
+ .getMessage()
+ .contains(
+ "the number of source paths and the number of target paths
should be the same"));
+ }
+ }
+
+ @Test
+ public void testMultiPrefixPathsInFromClause() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select s1 into target from root.sg.d1, root.sg.d2");
+ fail();
+ } catch (SQLException throwable) {
+ assertTrue(
+ throwable
+ .getMessage()
+ .contains("the number of prefix paths in the from clause should
be 1"));
+ }
+ }
+
+ @Test
+ public void testLeveledPathNodePatternLimit() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select s1 into root.${100}.s1 from root.sg.d1");
+ fail();
+ } catch (SQLException throwable) {
+ assertTrue(
+ throwable
+ .getMessage()
+ .contains(
+ "the x of ${x} should be greater than 0 and equal to or less
than <level> or the length of queried path prefix."));
+ }
+
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select s1 into root.${0}.s1 from root.sg.d1");
+ fail();
+ } catch (SQLException throwable) {
+ assertTrue(
+ throwable
+ .getMessage()
+ .contains(
+ "the x of ${x} should be greater than 0 and equal to or less
than <level> or the length of queried path prefix."));
+ }
+
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select s1 into root.${wrong}.s1 from root.sg.d1");
+ fail();
+ } catch (SQLException throwable) {
+ assertTrue(throwable.getMessage().contains("the x of ${x} should be an
integer."));
+ }
+ }
+
+ @Test
+ public void testAlignByDevice() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select s1 into root.${1}.s1 from root.sg.d1 align by
device");
+ fail();
+ } catch (SQLException throwable) {
+ assertTrue(throwable.getMessage().contains("align by device clauses are
not supported."));
+ }
+ }
+
+ @Test
+ public void testDisableDevice() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select s1 into root.${1}.s1 from root.sg.d1 disable
align");
+ fail();
+ } catch (SQLException throwable) {
+ assertTrue(throwable.getMessage().contains("disable align clauses are
not supported."));
+ }
+ }
+
+ @Test
+ public void testLastQuery() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select last s1 into root.${1}.s1 from root.sg.d1");
+ fail();
+ } catch (SQLException throwable) {
+ assertTrue(throwable.getMessage().contains("last clauses are not
supported."));
+ }
+ }
+
+ @Test
+ public void testSlimit() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select s1, s2 into ${1}.s1, ${2}.s1 from root.sg.d1
slimit 1");
+ fail();
+ } catch (SQLException throwable) {
+ assertTrue(throwable.getMessage().contains("slimit clauses are not
supported."));
+ }
+ }
+
+ @Test
+ public void testDescending() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select s1, s2 into ${1}.s1, ${2}.s1 from root.sg.d1
order by time desc");
+ fail();
+ } catch (SQLException throwable) {
+ assertTrue(throwable.getMessage().contains("desc clauses are not
supported."));
+ }
+ }
+
+ @Test
+ public void testSameTargetPaths() {
+ try (Statement statement =
+ DriverManager.getConnection(Config.IOTDB_URL_PREFIX +
"127.0.0.1:6667/", "root", "root")
+ .createStatement()) {
+ statement.execute("select s1, s2 into ${1}.s1, ${1}.s1 from root.sg.d1");
+ fail();
+ } catch (SQLException throwable) {
+ assertTrue(
+ throwable.getMessage().contains("target paths in into clause should
be different."));
+ }
+ }
+}
diff --git
a/server/src/test/java/org/apache/iotdb/db/qp/physical/PhysicalPlanTest.java
b/server/src/test/java/org/apache/iotdb/db/qp/physical/PhysicalPlanTest.java
index 2712308..c6c0437 100644
--- a/server/src/test/java/org/apache/iotdb/db/qp/physical/PhysicalPlanTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/qp/physical/PhysicalPlanTest.java
@@ -1333,7 +1333,7 @@ public class PhysicalPlanTest {
(CreateContinuousQueryPlan) processor.parseSQLToPhysicalPlan(sql);
fail();
} catch (SQLParserException e) {
- assertEquals("CQ: x of ${x} should be an integer.", e.getMessage());
+ assertTrue(e.getMessage().contains("the x of ${x} should be an
integer."));
}
}
@@ -1347,9 +1347,10 @@ public class PhysicalPlanTest {
(CreateContinuousQueryPlan) processor.parseSQLToPhysicalPlan(sql);
fail();
} catch (SQLParserException e) {
- assertEquals(
- "CQ: x of ${x} should be greater than 0 and equal to or less than
<level> or the length of queried path prefix.",
- e.getMessage());
+ assertTrue(
+ e.getMessage()
+ .contains(
+ "the x of ${x} should be greater than 0 and equal to or less
than <level> or the length of queried path prefix."));
}
}
@@ -1373,19 +1374,6 @@ public class PhysicalPlanTest {
@Test
public void testCreateCQ9() throws QueryProcessException {
- String sql =
- "CREATE CONTINUOUS QUERY cq1 BEGIN SELECT max_value(temperature) INTO
${1}.${0}.${2}.${3}.temperature_max FROM root.ln.*.*.* GROUP BY time(10s),
level = 3 END";
- try {
- CreateContinuousQueryPlan plan =
- (CreateContinuousQueryPlan) processor.parseSQLToPhysicalPlan(sql);
- fail();
- } catch (ParseCancellationException e) {
- assertTrue(e.getMessage().contains("mismatched input '.' expecting
FROM"));
- }
- }
-
- @Test
- public void testCreateCQ10() throws QueryProcessException {
String sql =
"CREATE CQ cq1 RESAMPLE FOR 20s BEGIN SELECT max_value(temperature)
INTO root.${0}.${3}.temperature_max FROM root.ln.*.*.* GROUP BY time(10s),
level = 3 END";
@@ -1394,14 +1382,15 @@ public class PhysicalPlanTest {
(CreateContinuousQueryPlan) processor.parseSQLToPhysicalPlan(sql);
fail();
} catch (SQLParserException e) {
- assertEquals(
- "CQ: x of ${x} should be greater than 0 and equal to or less than
<level> or the length of queried path prefix.",
- e.getMessage());
+ assertTrue(
+ e.getMessage()
+ .contains(
+ "the x of ${x} should be greater than 0 and equal to or less
than <level> or the length of queried path prefix."));
}
}
@Test
- public void testCreateCQ11() throws QueryProcessException {
+ public void testCreateCQ10() throws QueryProcessException {
String sql =
"CREATE CQ cq1 RESAMPLE FOR 20s BEGIN SELECT max_value(temperature),
avg(temperature) INTO root.${0}.${3}.temperature_max FROM root.ln.*.*.* GROUP
BY time(10s), level = 3 END";
@@ -1415,7 +1404,7 @@ public class PhysicalPlanTest {
}
@Test
- public void testCreateCQ12() throws QueryProcessException {
+ public void testCreateCQ11() throws QueryProcessException {
long minEveryInterval =
IoTDBDescriptor.getInstance().getConfig().getContinuousQueryMinimumEveryInterval();
long everyInterval = minEveryInterval / 2;
diff --git a/site/src/main/.vuepress/config.js
b/site/src/main/.vuepress/config.js
index a2cb26c..a8f70eb 100644
--- a/site/src/main/.vuepress/config.js
+++ b/site/src/main/.vuepress/config.js
@@ -712,7 +712,8 @@ var config = {
// The trigger module
has not been implemented yet,
// so the website
should not show users how to use it to avoid misleading.
//
['Advanced-Features/Triggers','Trigger'],
-
['Advanced-Features/Continuous-Query','CQ (Continuous Query)'],
+ ['Advanced-Features/Select-Into','Query Write-back (SELECT
INTO)'],
+ ['Advanced-Features/Continuous-Query','CQ (Continuous Query)'],
]
},
{
@@ -1504,7 +1505,8 @@ var config = {
// The trigger module has not
been implemented yet,
// so the website should not
show users how to use it to avoid misleading.
//
['Advanced-Features/Triggers','触发器'],
-
['Advanced-Features/Continuous-Query','连续查询(CQ)'],
+ ['Advanced-Features/Select-Into','查询写回(SELECT INTO)'],
+ ['Advanced-Features/Continuous-Query','连续查询(CQ)'],
]
},
{
diff --git a/testcontainer/src/test/java/org/apache/iotdb/db/sql/Cases.java
b/testcontainer/src/test/java/org/apache/iotdb/db/sql/Cases.java
index c4c8178..fbbc3fa 100644
--- a/testcontainer/src/test/java/org/apache/iotdb/db/sql/Cases.java
+++ b/testcontainer/src/test/java/org/apache/iotdb/db/sql/Cases.java
@@ -461,4 +461,29 @@ public abstract class Cases {
resultSet.close();
}
}
+
+ @Test
+ public void testSelectInto() throws SQLException {
+ for (int i = 0; i < 10; i++) {
+ writeStatement.execute(
+ String.format("INSERT INTO root.sg.d%s(timestamp,s) VALUES(%s,%s)",
i, i, i));
+ }
+
+ writeStatement.execute(
+ "SELECT d0.s, d1.s, d2.s, d3.s, d4.s into d0.t, d1.t, d2.t, d3.t, d4.t
from root.sg;");
+ for (int i = 5; i < 10; ++i) {
+ writeStatement.execute(String.format("SELECT d%s.s into d%s.t from
root.sg;", i, i));
+ }
+
+ for (Statement readStatement : readStatements) {
+ for (int i = 0; i < 10; ++i) {
+ try (ResultSet resultSet =
+ readStatement.executeQuery(String.format("SELECT s, t FROM
root.sg.d%s", i))) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals(resultSet.getDouble(2), resultSet.getDouble(3),
0);
+ Assert.assertFalse(resultSet.next());
+ }
+ }
+ }
+ }
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/BitMap.java
b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/BitMap.java
index 91affdb..78fa5b1 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/BitMap.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/BitMap.java
@@ -64,6 +64,11 @@ public class BitMap {
return (bits[position / Byte.SIZE] & BIT_UTIL[position % Byte.SIZE]) != 0;
}
+ /** mark as 1 at all positions. */
+ public void markAll() {
+ Arrays.fill(bits, (byte) 0XFF);
+ }
+
/** mark as 1 at the given bit position. */
public void mark(int position) {
bits[position / Byte.SIZE] |= BIT_UTIL[position % Byte.SIZE];
