You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2021/09/30 09:49:50 UTC
[iotdb] branch influxdb-protocal updated: [To influxdb-protocal]
Compatible influxdb protocal (#4063)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch influxdb-protocal
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/influxdb-protocal by this push:
new 08c6371 [To influxdb-protocal] Compatible influxdb protocal (#4063)
08c6371 is described below
commit 08c6371156404a1be26ab3c671c01953e3c784a9
Author: Xieqijun <44...@users.noreply.github.com>
AuthorDate: Thu Sep 30 17:49:25 2021 +0800
[To influxdb-protocal] Compatible influxdb protocal (#4063)
* Compatible influxdb protocol
* Add apache license
* Autogen antlr4 code
* Format code and Adjust dependent version
* [IOTDB-1689] revise the Chinese notes into English
* [IOTDB-1689] spotless apply format code
* [IOTDB-1689] implements influxdb interface
* [IOTDB-1689] modify readme support function
* [IOTDB-1689] use iotdb built-in func
* [IOTDB-1689] add integration test and fix bugs
* [IOTDB-1689] add workflow it
* [IOTDB-1689] add workflow file
* [IOTDB-1689] add licenses
* [IOTDB-1689] build distribution zip
* [IOTDB-1689] modify bash
* [IOTDB-1689] avoid test influxdb
* [IOTDB-1689] modify maven version
* [IOTDB-1689] add factory
* [IOTDB-1689] spotless apply
* [IOTDB-1689] add license
* [IOTDB-1689] spotless apply
* [IOTDB-1689] fix bugs
* [IOTDB-1689] modify IoTDB name
* [IOTDB-1689] fix bugs
* [IOTDB-1689] remove useless g4
* [IOTDB-1689] rename file name and remove unless g4
* [IOTDB-1689] revise the Chinese note into English
* Update docs/zh/UserGuide/API/IoTDB-InfluxDB.md
* [IOTDB-1689] modify image src
Co-authored-by: Steve Yurong Su <ro...@apache.org>
---
.github/workflows/influxdb-iotdb.yml | 67 ++
.github/workflows/sonar-coveralls.yml | 2 +-
docs/zh/UserGuide/API/IoTDB-InfluxDB.md | 855 +++++++++++++++
iotdb-influxdb/pom.xml | 127 +++
.../org/apache/iotdb/influxdb/qp/sql/InfluxDB.g4 | 487 +++++++++
.../java/org/apache/iotdb/influxdb/Constant.java | 26 +
.../org/apache/iotdb/influxdb/IoTDBInfluxDB.java | 1125 ++++++++++++++++++++
.../iotdb/influxdb/IoTDBInfluxDBFactory.java | 40 +
.../apache/iotdb/influxdb/IoTDBInfluxDBUtils.java | 469 ++++++++
.../iotdb/influxdb/example/InfluxDBExample.java | 113 ++
.../iotdb/influxdb/qp/constant/FilterConstant.java | 95 ++
.../iotdb/influxdb/qp/constant/SQLConstant.java | 241 +++++
.../apache/iotdb/influxdb/qp/logical/Operator.java | 160 +++
.../qp/logical/crud/BasicFunctionOperator.java | 40 +
.../iotdb/influxdb/qp/logical/crud/Condition.java | 63 ++
.../influxdb/qp/logical/crud/FilterOperator.java | 78 ++
.../influxdb/qp/logical/crud/FromComponent.java | 38 +
.../influxdb/qp/logical/crud/FunctionOperator.java | 29 +
.../influxdb/qp/logical/crud/QueryOperator.java | 59 +
.../influxdb/qp/logical/crud/SelectComponent.java | 101 ++
.../influxdb/qp/logical/crud/WhereComponent.java | 39 +
.../influxdb/qp/logical/function/Aggregate.java | 39 +
.../qp/logical/function/CountFunction.java | 74 ++
.../qp/logical/function/FirstFunction.java | 108 ++
.../influxdb/qp/logical/function/Function.java | 69 ++
.../qp/logical/function/FunctionFactory.java | 81 ++
.../qp/logical/function/FunctionValue.java | 46 +
.../influxdb/qp/logical/function/LastFunction.java | 105 ++
.../influxdb/qp/logical/function/MaxFunction.java | 113 ++
.../influxdb/qp/logical/function/MeanFunction.java | 95 ++
.../qp/logical/function/MedianFunction.java | 63 ++
.../influxdb/qp/logical/function/MinFunction.java | 116 ++
.../influxdb/qp/logical/function/ModeFunction.java | 78 ++
.../influxdb/qp/logical/function/Selector.java | 61 ++
.../qp/logical/function/SpreadFunction.java | 110 ++
.../qp/logical/function/StddevFunction.java | 57 +
.../influxdb/qp/logical/function/SumFunction.java | 83 ++
.../iotdb/influxdb/qp/sql/InfluxDBSqlVisitor.java | 289 +++++
.../influxdb/qp/strategy/LogicalGenerator.java | 66 ++
.../iotdb/influxdb/qp/strategy/SQLParseError.java | 40 +
.../iotdb/influxdb/qp/utils/DatetimeUtils.java | 138 +++
.../apache/iotdb/influxdb/qp/utils/MathUtil.java | 75 ++
.../apache/iotdb/influxdb/qp/utils/TypeUtil.java | 26 +
.../influxdb/query/expression/Expression.java | 22 +
.../influxdb/query/expression/ResultColumn.java | 52 +
.../expression/binary/AdditionExpression.java | 34 +
.../query/expression/binary/BinaryExpression.java | 49 +
.../expression/binary/DivisionExpression.java | 34 +
.../query/expression/binary/ModuloExpression.java | 34 +
.../binary/MultiplicationExpression.java | 34 +
.../expression/binary/SubtractionExpression.java | 34 +
.../query/expression/unary/FunctionExpression.java | 119 +++
.../query/expression/unary/NegationExpression.java | 38 +
.../query/expression/unary/NodeExpression.java | 38 +
.../influxdb/DesignSchemePerformanceTest.java | 158 +++
.../iotdb/influxdb/IoTDBInfluxDBUtilsTest.java | 39 +
.../test/java/org/apache/iotdb/influxdb/Util.java | 34 +
.../influxdb/integration/IoTDBInfluxDBIT.java | 133 +++
site/src/main/.vuepress/config.js | 3 +-
59 files changed, 7039 insertions(+), 2 deletions(-)
diff --git a/.github/workflows/influxdb-iotdb.yml b/.github/workflows/influxdb-iotdb.yml
new file mode 100644
index 0000000..d161c91
--- /dev/null
+++ b/.github/workflows/influxdb-iotdb.yml
@@ -0,0 +1,67 @@
+# Licensed 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.
+
+# This workflow will build a Java project with Maven
+# For more information see: https://help.github.com/actions/language-and-framework-guides/building-and-testing-java-with-maven
+
+name: InfluxDB-IoTDB Test
+
+on:
+ push:
+ branches:
+ - master
+ - 'rel/*'
+ paths-ignore:
+ - 'docs/**'
+ pull_request:
+ branches:
+ - master
+ - 'rel/*'
+ paths-ignore:
+ - 'docs/**'
+ # allow manually run the action:
+ workflow_dispatch:
+
+env:
+ MAVEN_OPTS: -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3
+
+jobs:
+ ubuntu:
+ runs-on: ubuntu-latest
+
+ steps:
+ - uses: actions/checkout@v2
+
+ - name: Set up JDK 11
+ uses: actions/setup-java@v1
+ with:
+ java-version: 11
+
+ - name: Cache Maven packages
+ uses: actions/cache@v2
+ with:
+ path: ~/.m2
+ key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }}
+ restore-keys: ${{ runner.os }}-m2-
+
+ - name: Build Distribution Zip
+ run: ./mvnw.sh -B -DskipTests clean install
+
+ - name: Build Docker Image
+ run: |
+ docker build . -f docker/src/main/Dockerfile-single -t "iotdb:$GITHUB_SHA"
+ docker images
+
+ - name: IT Test
+ shell: bash
+ run: |
+ cd iotdb-influxdb && mvn -B clean compile post-integration-test -Dtest.port.closed=true
diff --git a/.github/workflows/sonar-coveralls.yml b/.github/workflows/sonar-coveralls.yml
index 30ec9aa..8bee245 100644
--- a/.github/workflows/sonar-coveralls.yml
+++ b/.github/workflows/sonar-coveralls.yml
@@ -49,7 +49,7 @@ jobs:
restore-keys: ${{ runner.os }}-m2-
- name: IT/UT Test
# we do not compile client-cpp for saving time, it is tested in client.yml
- run: mvn -B clean compile post-integration-test -Dtest.port.closed=true -Pcode-coverage -P '!testcontainer'
+ run: mvn -B clean compile post-integration-test -Dtest.port.closed=true -Pcode-coverage -P '!testcontainer,!iotdb-influxdb'
- name: Code Coverage (Coveralls)
if: ${{ success() && (github.event_name == 'pull_request_target' || github.event_name == 'push')}}
run: |
diff --git a/docs/zh/UserGuide/API/IoTDB-InfluxDB.md b/docs/zh/UserGuide/API/IoTDB-InfluxDB.md
new file mode 100644
index 0000000..b94d018
--- /dev/null
+++ b/docs/zh/UserGuide/API/IoTDB-InfluxDB.md
@@ -0,0 +1,855 @@
+<!--
+
+ 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.
+
+-->
+
+# IoTDB-InfluxDB适配器
+
+
+## 1.背景
+
+InfluxDB是当前世界排名第一的时序数据库，具有繁荣的生态系统，目前很多用户使用它来作为自己的第一选择。但是，实际上线后，会有高可用，高扩展的需求。如果换成别的数据库，会有比较高的迁移成本。
+
+## 2.目标
+
+开发一套Java版本的适配器可以使IoTDB兼容InfluxDB协议，完善IoTDB的功能。
+
+1. IoTDB-InfluxDB：支持InfluxDB写入；支持InfluxDB部分查询；支持完整的InfluxDB查询。
+2. 对正确性和性能的测试，不仅要适配InfluxDB，也要知道在繁重的负载下是否可以很好的工作，例如：以非常高的频率生成数据
+ 1. 正确性测试：通过适配器以influxdb的协议插入数据，然后查询IoTDB数据库，将我们认为发送的内容与我们希望存储的内容进行比较。进行正确性测试
+ 2. 性能测试：以多线程的方式或者以Fiber多协程方式并发写入和读取，进行性能测试，类似的 demo：https://github.com/Tencent/TencentKona-8/tree/KonaFiber/demo/fiber/iotdb-sync-stress-demo
+
+
+## 3.方案一
+
+### 3.1 IoTDB-InfluxDB适配器
+
+适配器是一个继承至InfluxDB基类的子类，实现了InfluxDB主要的写入和查询方法，用户通过改变代码中InfluxDB的实现类，从而使InfluxDB原有的操作函数没有改变，但是会以IoTDB的协议写入IoTDB数据库中。
+
+
+
+
+### 3.2 将InfluxDB的数据格式转换成IoTDB的数据格式
+
+#### 3.2.1问题
+
+1. 问题：InfluxDB中Tag的顺序不敏感，而在IoTDB中是敏感的。
+
+ a. InfluxDB的时序构成
+
+ i. measurement
+
+ ii.tag key tag value
+
+ iii. field key field value
+
+ b. IoTDB的时序构成
+
+ i. storage group
+
+ ii. path(time series ID)
+
+ iii. measurement
+
+2. 关键点：需要记录每个tag对应的顺序，确保InfluxDB中label顺序不同的同一条时序对应到IoTDB中也是一条时序
+3. 需要解决的事情
+ 1. 怎样映射tag key和它对应的order
+ 2. 在不知道所有的label key的情况下，怎么维护他们之间的顺序
+
+### 3.2.2解决方案
+
+##### 3.2.2.1主要思想
+
+1. 内存中Map <Database_Measurement, Map <Tag Key, Order> > table结构维护Tag之间的顺序
+2. InfluxDB中时序根据label顺序对应到IoTDB
+
+##### 3.2.2.2实例
+
+a. 添加时序
+
+ 1. InfluxDB时序(database=testdabase)：
+
+ (1)student{name=A,phone=B,sex=C}
+
+ (2)student{address=D}
+
+ (3))student{name=A,phone=B,sex=C,address=D}
+
+ 2. 简单对上述InfluxDB的时序进行解释，database是testdatabase；measurement是student；tag分别是name，phone、sex和address
+
+ 3. (1)对应的记录tag顺序的table为
+
+
+
+ | database_measurement | tag_key | order |
+ | -------------------- | ------- | ----- |
+ | testdatabase_student | name | 0 |
+ | testdatabase_student | phone | 1 |
+ | testdatabase_student | sex | 2 |
+
+ (2)对应的记录tag顺序的table为
+
+
+
+ | database_measurement | tag_key | order |
+ | -------------------- | ------- | ----- |
+ | testdatabase_student | name | 0 |
+ | testdatabase_student | phone | 1 |
+ | testdatabase_student | sex | 2 |
+ | testdatabase_student | address | 3 |
+
+ (3)对应的记录tag顺序的table为
+
+
+
+ | database_measurement | tag_key | order |
+ | -------------------- | ------- | ----- |
+ | testdatabase_student | name | 0 |
+ | testdatabase_student | phone | 1 |
+ | testdatabase_student | sex | 2 |
+ | testdatabase_student | address | 3 |
+
+4. (1)对应IoTDB时序为root.testdatabase.student.A.B.C
+
+ (2)对应IoTDB时序为root.testdatabase.student.ph.ph.ph.D(其中ph表示占位符)
+
+ (3)对应IoTDB时序为root.testdatabase.student.A.B.C.D
+
+5. 为了重启时候对table的恢复，在IoTDB中记录数据
+
+ | `root.TAG_INFO.database_name` | `root.TAG_INFO.measurement_name` | `root.TAG_INFO.tag_name` | `root.TAG_INFO.tag_order` |
+ | :------------------------------- | :------------------------------- | :----------------------- | :------------------------ |
+ | testdatabase| student | name | 0 |
+ | testdatabase| student | phone | 1 |
+ | testdatabase| student | sex | 2 |
+ | testdatabase| student | address | 3 |
+
+b. 查询数据
+
+ 1. 查询student中phone=B的数据。在testdatabase_student中phone的顺序为1，order最大值是3，对应到IoTDB的查询为：select * from root.testdatabase.student.*.B
+ 2. 查询student中phone=B且存储的socre>97的数据，对应到IoTDB的查询为：select * from root.testdatabase.student.*.B where socre>98
+ 3. 查询student中phone=B且存储的socre>97且时间在最近七天内的的数据，对应到IoTDB的查询为：select * from root.testdatabase.student.*.B where socre>98 and time > now()-7d
+
+## 4.方案二
+
+### 4.1关系映射
+
+#### 4.11influxdb：
+
+1. database
+2. measurement
+3. tag key tag value. field key field value
+
+#### 4.1.2IoTDB
+
+1. storage group
+2. device
+3. timeseries
+
+我们将上面对应的三点一一对应，举例如下
+
+### 4.2举例
+
+#### 4.2.1写入
+
+influxdb的写入语句分别为（默认database=testdata）
+
+1. insert cpu,host=serverA,region=us value=0.64（tag为host、region）
+2. insert cpu,host=serverA,region=us,sex=n value=0.64（tag为host、region、sex）
+
+最终写入情况为：
+
+
+
+IoTDB的写入数据为
+
+1. insert into root.testdata.cpu(timestamp,host,region,value) values(now(),"serverA","us",0.64)
+2. insert into root.testdata.cpu(timestamp,host,region,sex,value) values(now(),"serverA","us","n",0.64)
+
+最终写入情况为：
+
+
+
+我们可以发现，二者实际存储对表面形式也比较类似。
+
+#### 4.2.2查询
+
+当把influxdb的tag和field都当作timeseries写入数据中时，由于存储的表面形式比较类似，最终查询就比较方便了。
+
+1. 根据值查询
+
+ select * from root.testdata.cpu where value = 0.64
+
+2. 根据tag查询
+
+ select * from. root.testdata.cpu where region = "us"
+
+3. 根据time查询
+
+ select * from root.testdata.cpu where time>now()-7d
+
+4. 混合查询
+
+ 由于把tag和filed都当作timeseries查询，混合查询也同理。
+
+5. group by
+
+ influxdb中group by可以按照tag分组展示
+
+ 
+
+ inflxudb中的group by功能不太类似，因此实现方法是：先获取所有的数据，然后在列表中以hash的方式手动分组，理论上复杂度为O（n），复杂度可以接受。
+
+## 5.方案对比-性能测试
+
+### 5.1存储情况
+
+1. influxdb的tag和filed均当作IoTDB的timeseries
+
+
+
+2. influxdb的tag当作路径
+
+
+
+### 5.2查询情况
+
+1. 查找条件为select * from root.teststress.test1 where RL = 1 and A = 1 and B =1 and C=1
+
+ 其中RL是tag，A，B，C均为field
+
+2. 取中间值情况，即当有十个tag时，取中间第五个累积递增
+
+ ```java
+ SessionDataSet dataSet = session.executeQueryStatement("select * from root.teststress.test2.*.*.*.*.SL where A=1 and B=1 and C=1");
+ dataSet = session.executeQueryStatement("select * from root.teststress.test2.*.*.*.*.SL.* where A=1 and B=1 and C=1");
+ dataSet = session.executeQueryStatement("select * from root.teststress.test2.*.*.*.*.SL.*.*.* where A=1 and B=1 and C=1");
+ dataSet = session.executeQueryStatement("select * from root.teststress.test2.*.*.*.*.SL.*.*.*.* where A=1 and B=1 and C=1");
+ dataSet = session.executeQueryStatement("select * from root.teststress.test2.*.*.*.*.SL.*.*.*.*.* where A=1 and B=1 ");
+ ```
+
+### 5.3特殊说明
+
+1. 为了存储的方便，第二种情况的存储，没有把tag的value存入路径中，即直接把tag的key存入路径中。其表现为
+
+ 前面累积的path都是相同的，这样最后会导致的结果是：会加快根据path过滤的查找，如：
+
+ ```sql
+ select * from root.teststress.test2.*.*.*.*.SL.*.*.*.*.*
+ ```
+
+ 这时只有一条路径，所有速度会变快，即第二种查找的时间会比实际的快一些
+
+### 5.4测试结果
+
+第一种查找的时间平均可以达到1000多ms，第二种查找的时间在300ms附近
+
+同时如果第二种查找的时候，在较靠前的路径使用*，(select * from root.teststress.test2.* where A=1 and B=1 )会导致需要查找的path过多，报错信息如下
+
+```log
+Too many paths in one query! Currently allowed max deduplicated path number is 715, this query contains 1000 deduplicated path. Please use slimit to choose what you real want or adjust max_deduplicated_path_num in iotdb-engine.properties.
+```
+
+综上所述，建议采取第二种方案（把tag放在path中存储）(同时需要解决上述提到的问题:Too many paths in one query)。
+
+## 6.查询优化
+为了尽可能的降低session查询的次数，应将一些可以合并的过滤条件进行合并，从而减少查询的次数。
+
+同时需要注意的是，由于tag过滤是通过path进行过滤，而filed过滤是通过IoTDB的where值过滤，因此二者的**or**过滤无法合并成一次过滤。即where tag1=value1 or field1=value2这种情况，无法合并成一次过滤。因此下文中尝试解决就是把可以合并的情况进行合并，即将**and**过滤进行合并。为了减少查询的次数，该算法使用递归的方法，对子树进行判断并尝试合并。
+
+举例如下：
+
+```sql
+select * from cpu where (host = 'serverA' and regions='us') or (regions = 'us' and value=0.77)
+```
+
+InfluxDB语法解析器生成的语法树如下图所示：
+
+
+此时我们由两种解决思路：
+
+1. 分别进行四次查询，host=serverA ；regions=us;regions=us和value=0.77四次查询，然后分别按照树的token，即or或者and进行合并。
+2. 将and的子树进行合并，即查询host=serverA and regions=us ；regions=us and value=0.77.只需要进行两次查询。也就是本文提到的查询优化算法。
+
+当然，如果根节点是and，同时左右子树分别对应的两个操作也是and，那么最终会合并成**一次**查询。
+
+## 7. Influxdb所有函数列表
+
+### 1. COUNT（）
+
+返回非空字段值的数目。
+
+#### 语法描述
+
+```
+COUNT(field_key)
+```
+
+返回field key对应的field values的数目。
+
+```
+COUNT(/regular_expression/)
+```
+
+返回匹配正则表达式的field key对应的field values的数目。
+
+```
+COUNT(*)
+```
+
+返回measurement中的每个field key对应的field value的数目。
+
+
+### 2. DISTINCE（）
+
+返回field value的不同值列表。
+
+### 语法描述
+
+```
+DISTINCT(field_key)
+```
+
+返回field key对应的不同field values。
+
+```
+DISTINCT(/regular_expression/)
+```
+
+返回匹配正则表达式的field key对应的不同field values。
+
+```
+DISTINCT(*)
+```
+
+返回measurement中的每个field key对应的不同field value。
+
+`DISTINCT()`支持所有数据类型的field value，InfluxQL支持`COUNT()`嵌套`DISTINCT()`。
+
+
+### 3.INTEGRAL()
+
+返回字段曲线下的面积，即是积分。
+
+#### 语法描述
+
+InfluxDB计算字段曲线下的面积，并将这些结果转换为每`unit`的总和面积。`unit`参数是一个整数，后跟一个时间字符串，它是可选的。如果查询未指定单位，则单位默认为1秒（`1s`）。
+
+```
+INTEGRAL(field_key)
+```
+
+返回field key关联的值之下的面积。
+
+```
+INTEGRAL(/regular_expression/)
+```
+
+返回满足正则表达式的每个field key关联的值之下的面积。
+
+```
+INTEGRAL(*)
+```
+
+返回measurement中每个field key关联的值之下的面积。
+
+`INTEGRAL()`不支持`fill()`，`INTEGRAL()`支持int64和float64两个数据类型。
+
+
+### 4.MEAN（）
+
+返回字段的平均值
+
+#### 语法描述
+
+```
+MEAN(field_key)
+```
+
+返回field key关联的值的平均值。
+
+```
+MEAN(/regular_expression/)
+```
+
+返回满足正则表达式的每个field key关联的值的平均值。
+
+```
+MEAN(*)
+```
+
+返回measurement中每个field key关联的值的平均值。
+
+`MEAN()`支持int64和float64两个数据类型。
+
+
+### 5.MEDIAN（）
+
+在已经生成的QueryResult中找到满足条件的column，遍历获取不同值的列表
+
+#### 语法描述
+
+```
+MEDIAN(field_key)
+```
+
+返回field key关联的值的中位数。
+
+```
+MEDIAN(/regular_expression/)
+```
+
+返回满足正则表达式的每个field key关联的值的中位数。
+
+```
+MEDIAN(*)
+```
+
+返回measurement中每个field key关联的值的中位数。
+
+`MEDIAN()`支持int64和float64两个数据类型。
+
+> 注意：`MEDIAN()`近似于`PERCENTILE（field_key，50）`，除了如果该字段包含偶数个值，`MEDIAN()`返回两个中间字段值的平均值之外。
+
+
+### 6. MODE（）
+
+返回字段中出现频率最高的值。
+
+#### 语法描述
+
+```
+MODE(field_key)
+```
+
+返回field key关联的值的出现频率最高的值。
+
+```
+MODE(/regular_expression/)
+```
+
+返回满足正则表达式的每个field key关联的值的出现频率最高的值。
+
+```
+MODE(*)
+```
+
+返回measurement中每个field key关联的值的出现频率最高的值。
+
+`MODE()`支持所有数据类型。
+
+> 注意：`MODE()`如果最多出现次数有两个或多个值，则返回具有最早时间戳的字段值。
+
+### 7.SPREAD()
+
+返回字段中最大和最小值的差值。
+
+#### 语法描述
+
+```
+SPREAD(field_key)
+```
+
+返回field key最大和最小值的差值。
+
+```
+SPREAD(/regular_expression/)
+```
+
+返回满足正则表达式的每个field key最大和最小值的差值。
+
+```
+SPREAD(*)
+```
+
+返回measurement中每个field key最大和最小值的差值。
+
+`SPREAD()`支持所有的数值类型的field。
+
+
+### 8.STDDEV()
+
+返回字段的标准差。
+
+#### 语法描述
+
+```
+STDDEV(field_key)
+```
+
+返回field key的标准差。
+
+```
+STDDEV(/regular_expression/)
+```
+
+返回满足正则表达式的每个field key的标准差。
+
+```
+STDDEV(*)
+```
+
+返回measurement中每个field key的标准差。
+
+`STDDEV()`支持所有的数值类型的field。
+
+
+### 9.SUM（）
+
+返回字段值的和。
+
+#### 语法描述
+
+```
+SUM(field_key)
+```
+
+返回field key的值的和。
+
+```
+SUM(/regular_expression/)
+```
+
+返回满足正则表达式的每个field key的值的和。
+
+```
+SUM(*)
+```
+
+返回measurement中每个field key的值的和。
+
+`SUM()`支持所有的数值类型的field。
+
+
+### 10.BOTTOM()
+
+返回最小的N个field值。
+
+#### 语法描述
+
+```
+BOTTOM(field_key,N)
+```
+
+返回field key的最小的N个field value。
+
+```
+BOTTOM(field_key,tag_key(s),N)
+```
+
+返回某个tag key的N个tag value的最小的field value。
+
+```
+BOTTOM(field_key,N),tag_key(s),field_key(s)
+```
+
+返回括号里的字段的最小N个field value，以及相关的tag或field，或者两者都有。
+
+`BOTTOM()`支持所有的数值类型的field。
+
+> 说明：
+>
+> - 如果一个field有两个或多个相等的field value，`BOTTOM()`返回时间戳最早的那个。
+> - `BOTTOM()`和`INTO`子句一起使用的时候，和其他的函数有些不一样。
+
+
+### 11.FIRST()
+
+返回时间戳最早的值
+
+#### 语法描述
+
+```
+FIRST(field_key)
+```
+
+返回field key时间戳最早的值。
+
+```
+FIRST(/regular_expression/)
+```
+
+返回满足正则表达式的每个field key的时间戳最早的值。
+
+```
+FIRST(*)
+```
+
+返回measurement中每个field key的时间戳最早的值。
+
+```
+FIRST(field_key),tag_key(s),field_key(s)
+```
+
+返回括号里的字段的时间戳最早的值，以及相关联的tag或field，或者两者都有。
+
+`FIRST()`支持所有类型的field。
+
+
+### 12.LAST()
+
+返回时间戳最近的值
+
+#### 语法描述
+
+```
+LAST(field_key)
+```
+
+返回field key时间戳最近的值。
+
+```
+LAST(/regular_expression/)
+```
+
+返回满足正则表达式的每个field key的时间戳最近的值。
+
+```
+LAST(*)
+```
+
+返回measurement中每个field key的时间戳最近的值。
+
+```
+LAST(field_key),tag_key(s),field_key(s)
+```
+
+返回括号里的字段的时间戳最近的值，以及相关联的tag或field，或者两者都有。
+
+`LAST()`支持所有类型的field。
+
+
+### 13.MAX()
+
+返回最大的字段值
+
+#### 语法描述
+
+```
+MAX(field_key)
+```
+
+返回field key的最大值。
+
+```
+MAX(/regular_expression/)
+```
+
+返回满足正则表达式的每个field key的最大值。
+
+```
+MAX(*)
+```
+
+返回measurement中每个field key的最大值。
+
+```
+MAX(field_key),tag_key(s),field_key(s)
+```
+
+返回括号里的字段的最大值，以及相关联的tag或field，或者两者都有。
+
+`MAX()`支持所有数值类型的field。
+
+
+### 14.MIN()
+
+返回最小的字段值
+
+#### 语法描述
+
+```
+MIN(field_key)
+```
+
+返回field key的最小值。
+
+```
+MIN(/regular_expression/)
+```
+
+返回满足正则表达式的每个field key的最小值。
+
+```
+MIN(*)
+```
+
+返回measurement中每个field key的最小值。
+
+```
+MIN(field_key),tag_key(s),field_key(s)
+```
+
+返回括号里的字段的最小值，以及相关联的tag或field，或者两者都有。
+
+`MIN()`支持所有数值类型的field。
+
+
+### 15.PERCENTILE()
+
+返回较大百分之N的字段值
+
+#### 语法描述
+
+```
+PERCENTILE(field_key,N)
+```
+
+返回field key较大的百分之N的值。
+
+```
+PERCENTILE(/regular_expression/,N)
+```
+
+返回满足正则表达式的每个field key较大的百分之N的值。
+
+```
+PERCENTILE(*,N)
+```
+
+返回measurement中每个field key较大的百分之N的值。
+
+```
+PERCENTILE(field_key,N),tag_key(s),field_key(s)
+```
+
+返回括号里的字段较大的百分之N的值，以及相关联的tag或field，或者两者都有。
+
+`N`必须是0到100的整数或者浮点数。
+
+`PERCENTILE()`支持所有数值类型的field。
+
+
+### 16.SAMPLE()
+
+返回`N`个随机抽样的字段值。`SAMPLE()`使用reservoir sampling来生成随机点。
+
+#### 语法描述
+
+```
+SAMPLE(field_key,N)
+```
+
+返回field key的N个随机抽样的字段值。
+
+```
+SAMPLE(/regular_expression/,N)
+```
+
+返回满足正则表达式的每个field key的N个随机抽样的字段值。
+
+```
+SAMPLE(*,N)
+```
+
+返回measurement中每个field key的N个随机抽样的字段值。
+
+```
+SAMPLE(field_key,N),tag_key(s),field_key(s)
+```
+
+返回括号里的字段的N个随机抽样的字段值，以及相关联的tag或field，或者两者都有。
+
+`N`必须是整数。
+
+`SAMPLE()`支持所有类型的field。
+
+
+### 17.TOP()
+
+返回最大的N个field值。
+
+#### 语法描述
+
+```
+TOP(field_key,N)
+```
+
+返回field key的最大的N个field value。
+
+```
+TOP(field_key,tag_key(s),N)
+```
+
+返回某个tag key的N个tag value的最大的field value。
+
+```
+TOP(field_key,N),tag_key(s),field_key(s)
+```
+
+返回括号里的字段的最大N个field value，以及相关的tag或field，或者两者都有。
+
+`TOP()`支持所有的数值类型的field。
+
+> 说明：
+>
+> - 如果一个field有两个或多个相等的field value，`TOP()`返回时间戳最早的那个。
+> - `TOP()`和`INTO`子句一起使用的时候，和其他的函数有些不一样。
+
+
+## 8. 支持的函数
+
+1. first()
+先利用IoTDB中的first_value函数，然后对每一个device中的first值再重回IoTDB中查询，指定path和where限定条件，查询出对应的time
+4. last()
+先利用IoTDB中的last_value函数，然后对每一个device中的last值再重回IoTDB中查询，指定path和where限定条件，查询出对应的time
+5. max()
+通过利用IoTDB中的max_value函数，对多个device对最大值再求最大值，找到最大值后，指定path和where限定条件，再重回IoTDB中查询对应的time
+6. min()
+通过利用IoTDB中的min_value函数，对多个device对最小值再求最小值，找到最小值后，指定path和where限定条件，再重回IoTDB中查询对应的time
+7. count()
+通过利用IoTDB中的count函数,将多个设备的count值求和
+8. sum()
+通过利用IoTDB中的sum函数，对多个device的求和值再进行求和
+9. mean()
+通过利用IoTDB中的avg和count函数，对多个device的数量乘以avg之和再除以总数量
+10. spread()
+通过利用IoTDB中的max_value和min_value函数，对多个device查找最大值和最小值，进行求和
+
+## 9.切换方案
+
+ - 原版本
+```java
+influxDB = InfluxDBFactory.connect(openurl, username, password);
+influxDB.createDatabase(database);
+influxDB.insert(ponit);
+influxDB.query(query);
+```
+
+- 迁移版本
+```java
+influxDB = IotDBInfluxDBFactory.connect(openurl, username, password);
+influxDB.createDatabase(database);
+influxDB.insert(ponit);
+influxDB.query(query);
+```
+
+只需要把
+InfluxDBFactory.connect(openurl, username, password);
+
+改为
+**IotDBInfluxDBFactory**.connect(openurl, username, password);
+
+即可
+
+## 10.参考资料
+
+1. https://summer.iscas.ac.cn/#/org/orgdetail/apacheiotdb
+2. https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=165224300#Prometheus%E8%BF%9E%E6%8E%A5%E5%99%A8-3.4%E5%8F%82%E8%80%83%E6%96%87%E6%A1%A3
+
diff --git a/iotdb-influxdb/pom.xml b/iotdb-influxdb/pom.xml
new file mode 100644
index 0000000..6ce86cd
--- /dev/null
+++ b/iotdb-influxdb/pom.xml
@@ -0,0 +1,127 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-parent</artifactId>
+ <version>0.13.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+ <artifactId>iotdb-influxdb</artifactId>
+ <name>iotdb-influxdb</name>
+ <description>compatible with the protocol of influxdb.</description>
+ <properties>
+ <maven.compiler.source>8</maven.compiler.source>
+ <maven.compiler.target>8</maven.compiler.target>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-session</artifactId>
+ <version>0.12.2</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-thrift</artifactId>
+ <version>0.12.2</version>
+ </dependency>
+ <dependency>
+ <groupId>org.influxdb</groupId>
+ <artifactId>influxdb-java</artifactId>
+ <version>2.21</version>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>RELEASE</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.antlr</groupId>
+ <artifactId>antlr4-runtime</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers</artifactId>
+ <version>LATEST</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.antlr</groupId>
+ <artifactId>antlr4-maven-plugin</artifactId>
+ <version>${antlr4.version}</version>
+ <executions>
+ <execution>
+ <configuration>
+ <listener>false</listener>
+ <visitor>true</visitor>
+ </configuration>
+ <goals>
+ <goal>antlr4</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <version>3.2.0</version>
+ <executions>
+ <execution>
+ <id>add-source</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <configuration>
+ <sources>
+ <source>${project.build.directory}/generated-sources/antlr4</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>run-integration-tests</id>
+ <phase>integration-test</phase>
+ <goals>
+ <goal>integration-test</goal>
+ <goal>verify</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <skipTests>false</skipTests>
+ <skipITs>false</skipITs>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/iotdb-influxdb/src/main/antlr4/org/apache/iotdb/influxdb/qp/sql/InfluxDB.g4 b/iotdb-influxdb/src/main/antlr4/org/apache/iotdb/influxdb/qp/sql/InfluxDB.g4
new file mode 100644
index 0000000..1bf3b5d
--- /dev/null
+++ b/iotdb-influxdb/src/main/antlr4/org/apache/iotdb/influxdb/qp/sql/InfluxDB.g4
@@ -0,0 +1,487 @@
+/*
+ * 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.
+ */
+
+grammar InfluxDB;
+
+singleStatement
+ : statement (';')? EOF
+ ;
+
+/*
+ * According to The Definitive ANTLR 4 Reference, 11. Altering the Parse with Semantic Predicates, Altering the Parse with Semantic Predicates.
+ * "It s a good idea to avoid embedding predicates in the parser when possible for efficiency and clarity reasons."
+ * So if unnecessary, don't use embedding predicates.
+ */
+
+ statement
+ : selectClause fromClause whereClause? #selectStatement
+ ;
+
+selectClause
+ : SELECT resultColumn (COMMA resultColumn)*
+ ;
+
+resultColumn
+ : 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=nodeName LR_BRACKET expression (COMMA expression)* functionAttribute* RR_BRACKET
+ | nodeName
+ | literal=SINGLE_QUOTE_STRING_LITERAL
+ ;
+
+functionAttribute
+ : COMMA functionAttributeKey=stringLiteral OPERATOR_EQ functionAttributeValue=stringLiteral
+ ;
+
+
+whereClause
+ : WHERE orExpression
+ ;
+
+
+orExpression
+ : andExpression (OPERATOR_OR andExpression)*
+ ;
+
+andExpression
+ : predicate (OPERATOR_AND predicate)*
+ ;
+
+predicate
+ : (TIME | TIMESTAMP | nodeName ) comparisonOperator constant
+ | OPERATOR_NOT? LR_BRACKET orExpression RR_BRACKET
+ ;
+
+
+fromClause
+ : FROM nodeName (COMMA nodeName)*
+ ;
+
+comparisonOperator
+ : type = OPERATOR_GT
+ | type = OPERATOR_GTE
+ | type = OPERATOR_LT
+ | type = OPERATOR_LTE
+ | type = OPERATOR_EQ
+ | type = OPERATOR_NEQ
+ ;
+
+nodeName
+ : ID STAR?
+ | STAR
+ | DOUBLE_QUOTE_STRING_LITERAL
+ | DURATION
+ | dataType
+ | dateExpression
+ | MINUS? (EXPONENT | INT)
+ | booleanClause
+ | SELECT
+ | INTO
+ | WHERE
+ | FROM
+ | TO
+ | TIMESERIES
+ | TIMESTAMP
+ | DATATYPE
+ | NOW
+ | TIME
+ | (ID | OPERATOR_IN)? LS_BRACKET INT? ID? RS_BRACKET? ID?
+ ;
+
+dataType
+ : INT32 | INT64 | FLOAT | DOUBLE | BOOLEAN | TEXT
+ ;
+
+dateFormat
+//暂不支持datetime
+// : DATETIME
+ : NOW LR_BRACKET RR_BRACKET
+ ;
+
+constant
+ : dateExpression
+ | NaN
+ | MINUS? realLiteral
+ | MINUS? INT
+ | stringLiteral
+ | booleanClause
+ | NULL
+ ;
+
+booleanClause
+ : TRUE
+ | FALSE
+ ;
+
+dateExpression
+ : dateFormat ((PLUS | MINUS) DURATION)*
+ ;
+
+realLiteral
+ : INT DOT (INT | EXPONENT)?
+ | DOT (INT|EXPONENT)
+ | EXPONENT
+ ;
+
+//============================
+// Start of the keywords list
+//============================
+
+SELECT
+ : S E L E C T
+ ;
+
+INTO
+ : I N T O
+ ;
+
+
+WHERE
+ : W H E R E
+ ;
+
+FROM
+ : F R O M
+ ;
+
+TO
+ : T O
+ ;
+
+NULL
+ : N U L L
+ ;
+
+
+TIMESERIES
+ : T I M E S E R I E S
+ ;
+
+TIMESTAMP
+ : T I M E S T A M P
+ ;
+
+
+
+DATATYPE
+ : D A T A T Y P E
+ ;
+
+INT32
+ : I N T '3' '2'
+ ;
+
+INT64
+ : I N T '6' '4'
+ ;
+
+FLOAT
+ : F L O A T
+ ;
+
+DOUBLE
+ : D O U B L E
+ ;
+
+BOOLEAN
+ : B O O L E A N
+ ;
+
+TEXT
+ : T E X T
+ ;
+
+
+AS
+ : A S
+ ;
+
+
+ALIAS
+ : A L I A S
+ ;
+
+
+NOW
+ : N O W
+ ;
+
+TIME
+ : T I M E
+ ;
+
+TRUE
+ : T R U E
+ ;
+
+FALSE
+ : F A L S E
+ ;
+
+//============================
+// End of the keywords list
+//============================
+COMMA : ',';
+
+STAR : '*';
+
+OPERATOR_EQ : '=' | '==';
+
+OPERATOR_GT : '>';
+
+OPERATOR_GTE : '>=';
+
+OPERATOR_LT : '<';
+
+OPERATOR_LTE : '<=';
+
+OPERATOR_NEQ : '!=' | '<>';
+
+OPERATOR_IN : I N;
+
+OPERATOR_AND
+ : A N D
+ | '&'
+ | '&&'
+ ;
+
+OPERATOR_OR
+ : O R
+ | '|'
+ | '||'
+ ;
+
+OPERATOR_NOT
+ : N O T | '!'
+ ;
+
+OPERATOR_CONTAINS
+ : C O N T A I N S
+ ;
+
+MINUS : '-';
+
+PLUS : '+';
+
+DIV : '/';
+
+MOD : '%';
+
+DOT : '.';
+
+LR_BRACKET : '(';
+
+RR_BRACKET : ')';
+
+LS_BRACKET : '[';
+
+RS_BRACKET : ']';
+
+L_BRACKET : '{';
+
+R_BRACKET : '}';
+
+UNDERLINE : '_';
+
+NaN : 'NaN';
+
+stringLiteral
+ : SINGLE_QUOTE_STRING_LITERAL
+ | DOUBLE_QUOTE_STRING_LITERAL
+ ;
+
+INT : [0-9]+;
+
+EXPONENT : INT ('e'|'E') ('+'|'-')? INT ;
+
+DURATION
+ :
+ (INT+ (Y|M O|W|D|H|M|S|M S|U S|N S))+
+ ;
+
+DATETIME
+ : INT ('-'|'/') INT ('-'|'/') INT
+ ((T | WS)
+ INT ':' INT ':' INT (DOT INT)?
+ (('+' | '-') INT ':' INT)?)?
+ ;
+
+/** Allow unicode rule/token names */
+ID : FIRST_NAME_CHAR NAME_CHAR*;
+
+fragment
+NAME_CHAR
+ : 'A'..'Z'
+ | 'a'..'z'
+ | '0'..'9'
+ | '_'
+ | '-'
+ | ':'
+ | '/'
+ | '@'
+ | '#'
+ | '$'
+ | '%'
+ | '&'
+ | '+'
+ | CN_CHAR
+ ;
+
+fragment
+FIRST_NAME_CHAR
+ : 'A'..'Z'
+ | 'a'..'z'
+ | '0'..'9'
+ | '_'
+ | '/'
+ | '@'
+ | '#'
+ | '$'
+ | '%'
+ | '&'
+ | '+'
+ | CN_CHAR
+ ;
+
+fragment CN_CHAR
+ : '\u2E80'..'\u9FFF'
+ ;
+
+DOUBLE_QUOTE_STRING_LITERAL
+ : '"' ('\\' . | ~'"' )*? '"'
+ ;
+
+SINGLE_QUOTE_STRING_LITERAL
+ : '\'' ('\\' . | ~'\'' )*? '\''
+ ;
+
+//Characters and write it this way for case sensitivity
+fragment A
+ : 'a' | 'A'
+ ;
+
+fragment B
+ : 'b' | 'B'
+ ;
+
+fragment C
+ : 'c' | 'C'
+ ;
+
+fragment D
+ : 'd' | 'D'
+ ;
+
+fragment E
+ : 'e' | 'E'
+ ;
+
+fragment F
+ : 'f' | 'F'
+ ;
+
+fragment G
+ : 'g' | 'G'
+ ;
+
+fragment H
+ : 'h' | 'H'
+ ;
+
+fragment I
+ : 'i' | 'I'
+ ;
+
+fragment J
+ : 'j' | 'J'
+ ;
+
+fragment K
+ : 'k' | 'K'
+ ;
+
+fragment L
+ : 'l' | 'L'
+ ;
+
+fragment M
+ : 'm' | 'M'
+ ;
+
+fragment N
+ : 'n' | 'N'
+ ;
+
+fragment O
+ : 'o' | 'O'
+ ;
+
+fragment P
+ : 'p' | 'P'
+ ;
+
+fragment Q
+ : 'q' | 'Q'
+ ;
+
+fragment R
+ : 'r' | 'R'
+ ;
+
+fragment S
+ : 's' | 'S'
+ ;
+
+fragment T
+ : 't' | 'T'
+ ;
+
+fragment U
+ : 'u' | 'U'
+ ;
+
+fragment V
+ : 'v' | 'V'
+ ;
+
+fragment W
+ : 'w' | 'W'
+ ;
+
+fragment X
+ : 'x' | 'X'
+ ;
+
+fragment Y
+ : 'y' | 'Y'
+ ;
+
+fragment Z
+ : 'z' | 'Z'
+ ;
+
+WS
+ : [ \r\n\t]+ -> channel(HIDDEN)
+ ;
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/Constant.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/Constant.java
new file mode 100644
index 0000000..93791b5
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/Constant.java
@@ -0,0 +1,26 @@
+/*
+ * 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.influxdb;
+
+public class Constant {
+
+ private Constant() {}
+
+ static final String METHOD_NOT_SUPPORTED = "Method not supported";
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/IoTDBInfluxDB.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/IoTDBInfluxDB.java
new file mode 100644
index 0000000..1018e69
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/IoTDBInfluxDB.java
@@ -0,0 +1,1125 @@
+/*
+ * 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.influxdb;
+
+import org.apache.iotdb.influxdb.qp.constant.FilterConstant;
+import org.apache.iotdb.influxdb.qp.constant.SQLConstant;
+import org.apache.iotdb.influxdb.qp.logical.Operator;
+import org.apache.iotdb.influxdb.qp.logical.crud.*;
+import org.apache.iotdb.influxdb.qp.logical.function.*;
+import org.apache.iotdb.influxdb.qp.strategy.LogicalGenerator;
+import org.apache.iotdb.influxdb.query.expression.Expression;
+import org.apache.iotdb.influxdb.query.expression.ResultColumn;
+import org.apache.iotdb.influxdb.query.expression.unary.FunctionExpression;
+import org.apache.iotdb.influxdb.query.expression.unary.NodeExpression;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.SessionDataSet;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+
+import org.influxdb.BatchOptions;
+import org.influxdb.InfluxDB;
+import org.influxdb.dto.*;
+
+import java.lang.reflect.Field;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.*;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+public class IoTDBInfluxDB implements InfluxDB {
+
+ private static Session session;
+ // Database currently selected by influxdb
+ private String database;
+ // Measurement currently selected by influxdb
+ private String measurement;
+ // Tag list and order corresponding to all measurements in the current database
+ // TODO At present, the distributed situation is not considered. It is assumed that all writes are
+ // performed by the instance
+ private Map<String, Map<String, Integer>> measurementTagOrder = new HashMap<>();
+ // Tag list and order under current measurement
+ private Map<String, Integer> tagOrders;
+
+ // The list of fields under the current measurement and the order of the specified rules
+ private Map<String, Integer> fieldOrders;
+ private Map<Integer, String> fieldOrdersReversed;
+
+ private final String placeholder = "PH";
+
+ /**
+ * constructor function
+ *
+ * @param url contain host and port
+ * @param userName username
+ * @param password user password
+ */
+ public IoTDBInfluxDB(String url, String userName, String password) {
+ try {
+ URI uri = new URI(url);
+ new IoTDBInfluxDB(uri.getHost(), uri.getPort(), userName, password);
+ } catch (URISyntaxException e) {
+ e.printStackTrace();
+ throw new IllegalArgumentException(e);
+ }
+ }
+
+ /**
+ * constructor function
+ *
+ * @param host host
+ * @param rpcPort port
+ * @param userName username
+ * @param password user password
+ */
+ public IoTDBInfluxDB(String host, int rpcPort, String userName, String password) {
+ session = new Session(host, rpcPort, userName, password);
+ try {
+ session.open(false);
+ } catch (IoTDBConnectionException e) {
+ e.printStackTrace();
+ throw new IllegalArgumentException(e.getMessage());
+ }
+ session.setFetchSize(10000);
+ }
+
+ /**
+ * write function compatible with influxdb
+ *
+ * @param point Data structure for inserting data
+ */
+ public void write(Point point) {
+ String measurement = null;
+ Map<String, String> tags = new HashMap<>();
+ Map<String, Object> fields = new HashMap<>();
+ Long time = null;
+ Field[] reflectFields = point.getClass().getDeclaredFields();
+ // Get the property of point in influxdb by reflection
+ for (Field reflectField : reflectFields) {
+ reflectField.setAccessible(true);
+ try {
+ if (reflectField.getType().getName().equalsIgnoreCase("java.util.Map")
+ && reflectField.getName().equalsIgnoreCase("fields")) {
+ fields = (Map<String, Object>) reflectField.get(point);
+ }
+ if (reflectField.getType().getName().equalsIgnoreCase("java.util.Map")
+ && reflectField.getName().equalsIgnoreCase("tags")) {
+ tags = (Map<String, String>) reflectField.get(point);
+ }
+ if (reflectField.getType().getName().equalsIgnoreCase("java.lang.String")
+ && reflectField.getName().equalsIgnoreCase("measurement")) {
+ measurement = (String) reflectField.get(point);
+ }
+ if (reflectField.getType().getName().equalsIgnoreCase("java.lang.Number")
+ && reflectField.getName().equalsIgnoreCase("time")) {
+ time = (Long) reflectField.get(point);
+ }
+ } catch (IllegalAccessException e) {
+ throw new IllegalArgumentException(e.getMessage());
+ }
+ }
+ // set current time
+ if (time == null) {
+ time = System.currentTimeMillis();
+ }
+ tagOrders = measurementTagOrder.get(measurement);
+ if (tagOrders == null) {
+ tagOrders = new HashMap<>();
+ }
+ int measurementTagNum = tagOrders.size();
+ // The actual number of tags at the time of current insertion
+ Map<Integer, String> realTagOrders = new HashMap<>();
+ for (Map.Entry<String, String> entry : tags.entrySet()) {
+ if (tagOrders.containsKey(entry.getKey())) {
+ realTagOrders.put(tagOrders.get(entry.getKey()), entry.getKey());
+ } else {
+ measurementTagNum++;
+ try {
+ updateNewTagIntoDB(measurement, entry.getKey(), measurementTagNum);
+ } catch (IoTDBConnectionException | StatementExecutionException e) {
+ e.printStackTrace();
+ }
+ realTagOrders.put(measurementTagNum, entry.getKey());
+ tagOrders.put(entry.getKey(), measurementTagNum);
+ }
+ }
+ // update tagOrder map in memory
+ measurementTagOrder.put(measurement, tagOrders);
+ StringBuilder path = new StringBuilder("root." + database + "." + measurement);
+ for (int i = 1; i <= measurementTagNum; i++) {
+ if (realTagOrders.containsKey(i)) {
+ path.append(".").append(tags.get(realTagOrders.get(i)));
+ } else {
+ path.append("." + placeholder);
+ }
+ }
+
+ List<String> measurements = new ArrayList<>();
+ List<TSDataType> types = new ArrayList<>();
+ List<Object> values = new ArrayList<>();
+ for (Map.Entry<String, Object> entry : fields.entrySet()) {
+ measurements.add(entry.getKey());
+ Object value = entry.getValue();
+ if (value instanceof String) {
+ types.add(TSDataType.TEXT);
+ } else if (value instanceof Integer) {
+ types.add(TSDataType.INT32);
+ } else if (value instanceof Double) {
+ types.add(TSDataType.DOUBLE);
+ } else {
+ System.err.printf("can't solve type:%s", entry.getValue().getClass());
+ }
+ values.add(value);
+ }
+ try {
+ session.insertRecord(String.valueOf(path), time, measurements, types, values);
+ } catch (IoTDBConnectionException | StatementExecutionException e) {
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * query function compatible with influxdb
+ *
+ * @param query query parameters of influxdb, including databasename and SQL statement
+ * @return returns the query result of influxdb
+ */
+ public QueryResult query(Query query) {
+ String sql = query.getCommand();
+ String database = query.getDatabase();
+ if (!this.database.equals(database)) {
+ updateDatabase(database);
+ }
+ Operator operator = LogicalGenerator.generate(sql);
+ IoTDBInfluxDBUtils.checkQueryOperator(operator);
+ QueryOperator queryOperator = (QueryOperator) operator;
+ // update relative data
+ updateMeasurement(queryOperator.getFromComponent().getNodeName().get(0));
+ QueryResult queryResult = null;
+ try {
+ updateFiledOrders();
+ // contain filter condition or don't have function the result of the function is calculated by
+ // traversal
+ if (queryOperator.getWhereComponent() != null
+ || !queryOperator.getSelectComponent().isHasFunction()) {
+ // step1 : generate query results
+ queryResult = queryExpr(queryOperator.getWhereComponent().getFilterOperator());
+ // step2 : select filter
+ ProcessSelectComponent(queryResult, queryOperator.getSelectComponent());
+ }
+ // don't contain filter condition and have function use iotdb function
+ else {
+ queryResult = queryFuncWithoutFilter(queryOperator.getSelectComponent());
+ }
+ } catch (IoTDBConnectionException | StatementExecutionException e) {
+ throw new RuntimeException(e.getMessage());
+ }
+ return queryResult;
+ }
+
+ /**
+ * create database,write to iotdb
+ *
+ * @param name database name
+ */
+ public void createDatabase(String name) {
+ IoTDBInfluxDBUtils.checkNonEmptyString(name, "database name");
+ try {
+ session.setStorageGroup("root." + name);
+ } catch (IoTDBConnectionException | StatementExecutionException e) {
+ if (e instanceof StatementExecutionException
+ && ((StatementExecutionException) e).getStatusCode() == 300) {
+ // current database have been created
+ System.out.println(e.getMessage());
+ } else {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ /**
+ * delete database
+ *
+ * @param name database name
+ */
+ public void deleteDatabase(String name) {
+ try {
+ session.deleteStorageGroup("root." + name);
+ } catch (IoTDBConnectionException | StatementExecutionException e) {
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * set database，and get the list and order of all tags corresponding to the database
+ *
+ * @param database database name
+ */
+ public InfluxDB setDatabase(String database) {
+ if (!database.equals(this.database)) {
+ updateDatabase(database);
+ this.database = database;
+ }
+ return this;
+ }
+
+ @Override
+ public void write(String s) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void write(List<String> list) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void write(String s, String s1, Point point) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void write(int i, Point point) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void write(BatchPoints batchPoints) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void writeWithRetry(BatchPoints batchPoints) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void write(String s, String s1, ConsistencyLevel consistencyLevel, String s2) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void write(
+ String s, String s1, ConsistencyLevel consistencyLevel, TimeUnit timeUnit, String s2) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void write(String s, String s1, ConsistencyLevel consistencyLevel, List<String> list) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void write(
+ String s,
+ String s1,
+ ConsistencyLevel consistencyLevel,
+ TimeUnit timeUnit,
+ List<String> list) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void write(int i, String s) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void write(int i, List<String> list) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void query(Query query, Consumer<QueryResult> consumer, Consumer<Throwable> consumer1) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void query(Query query, int i, Consumer<QueryResult> consumer) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void query(Query query, int i, BiConsumer<Cancellable, QueryResult> biConsumer) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void query(Query query, int i, Consumer<QueryResult> consumer, Runnable runnable) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void query(
+ Query query, int i, BiConsumer<Cancellable, QueryResult> biConsumer, Runnable runnable) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void query(
+ Query query,
+ int i,
+ BiConsumer<Cancellable, QueryResult> biConsumer,
+ Runnable runnable,
+ Consumer<Throwable> consumer) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public QueryResult query(Query query, TimeUnit timeUnit) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public List<String> describeDatabases() {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public boolean databaseExists(String s) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void flush() {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void close() {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public InfluxDB setConsistency(ConsistencyLevel consistencyLevel) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public InfluxDB setRetentionPolicy(String s) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void createRetentionPolicy(String s, String s1, String s2, String s3, int i, boolean b) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void createRetentionPolicy(String s, String s1, String s2, int i, boolean b) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void createRetentionPolicy(String s, String s1, String s2, String s3, int i) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void dropRetentionPolicy(String s, String s1) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public InfluxDB setLogLevel(LogLevel logLevel) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public InfluxDB enableGzip() {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public InfluxDB disableGzip() {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public boolean isGzipEnabled() {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public InfluxDB enableBatch() {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public InfluxDB enableBatch(BatchOptions batchOptions) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public InfluxDB enableBatch(int i, int i1, TimeUnit timeUnit) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public InfluxDB enableBatch(int i, int i1, TimeUnit timeUnit, ThreadFactory threadFactory) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public InfluxDB enableBatch(
+ int i,
+ int i1,
+ TimeUnit timeUnit,
+ ThreadFactory threadFactory,
+ BiConsumer<Iterable<Point>, Throwable> biConsumer,
+ ConsistencyLevel consistencyLevel) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public InfluxDB enableBatch(
+ int i,
+ int i1,
+ TimeUnit timeUnit,
+ ThreadFactory threadFactory,
+ BiConsumer<Iterable<Point>, Throwable> biConsumer) {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public void disableBatch() {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public boolean isBatchEnabled() {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public Pong ping() {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ @Override
+ public String version() {
+ throw new UnsupportedOperationException(Constant.METHOD_NOT_SUPPORTED);
+ }
+
+ /**
+ * when the database changes, update the database related information, that is, obtain the list
+ * and order of all tags corresponding to the database from iotdb
+ *
+ * @param database update database name
+ */
+ private void updateDatabase(String database) {
+ try {
+ SessionDataSet result =
+ session.executeQueryStatement(
+ "select * from root.TAG_INFO where database_name="
+ + String.format("\"%s\"", database));
+ Map<String, Integer> tagOrder = new HashMap<>();
+ String measurementName = null;
+ while (result.hasNext()) {
+ List<org.apache.iotdb.tsfile.read.common.Field> fields = result.next().getFields();
+ String tmpMeasurementName = fields.get(1).getStringValue();
+ if (measurementName == null) {
+ // get measurement name for the first time
+ measurementName = tmpMeasurementName;
+ } else {
+ // if it is not equal, a new measurement is encountered
+ if (!tmpMeasurementName.equals(measurementName)) {
+ // add the tags of the current measurement to it
+ measurementTagOrder.put(measurementName, tagOrder);
+ tagOrder = new HashMap<>();
+ }
+ }
+ tagOrder.put(fields.get(2).getStringValue(), fields.get(3).getIntV());
+ }
+ // the last measurement is to add the tags of the current measurement
+ measurementTagOrder.put(measurementName, tagOrder);
+ } catch (StatementExecutionException e) {
+ // at first execution, tag_ If the info table is not created, intercept the error and print
+ // the log
+ if (e.getStatusCode() == 411) {
+ System.out.println(e.getMessage());
+ }
+ } catch (IoTDBConnectionException e) {
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * get query results in the format of influxdb through conditions
+ *
+ * @param conditions list of conditions, including tag and field condition
+ * @return returns the results of the influxdb query
+ */
+ private QueryResult queryByConditions(List<Condition> conditions)
+ throws IoTDBConnectionException, StatementExecutionException {
+ // used to store the actual order according to the tag
+ Map<Integer, Condition> realTagOrders = new HashMap<>();
+ // stores a list of conditions belonging to the field
+ List<Condition> fieldConditions = new ArrayList<>();
+ // maximum number of tags in the current query criteria
+ int currentQueryMaxTagNum = 0;
+ for (Condition condition : conditions) {
+ // the current condition is in tag
+ if (tagOrders.containsKey(condition.getValue())) {
+ int curOrder = tagOrders.get(condition.getValue());
+ // put it into the map according to the tag
+ realTagOrders.put(curOrder, condition);
+ // update the maximum tag order of the current query criteria
+ currentQueryMaxTagNum = Math.max(currentQueryMaxTagNum, curOrder);
+ } else {
+ fieldConditions.add(condition);
+ }
+ }
+ // construct the actual query path
+ StringBuilder curQueryPath = new StringBuilder("root." + database + "." + measurement);
+ // the maximum number of traversals from 1 to the current query condition
+ for (int i = 1; i <= currentQueryMaxTagNum; i++) {
+ if (realTagOrders.containsKey(i)) {
+ // since it is the value in the path, you need to remove the quotation marks at the
+ // beginning and end
+ curQueryPath
+ .append(".")
+ .append(IoTDBInfluxDBUtils.removeQuotation(realTagOrders.get(i).getLiteral()));
+ } else {
+ curQueryPath.append(".").append("*");
+ }
+ }
+ // construct actual query condition
+ StringBuilder realIotDBCondition = new StringBuilder();
+ for (int i = 0; i < fieldConditions.size(); i++) {
+ Condition condition = fieldConditions.get(i);
+ if (i != 0) {
+ realIotDBCondition.append(" and ");
+ }
+ realIotDBCondition
+ .append(condition.getValue())
+ .append(" ")
+ .append(FilterConstant.filterSymbol.get(condition.getFilterType()))
+ .append(" ")
+ .append(condition.getLiteral());
+ }
+ // actual query SQL statement
+ String realQuerySql;
+
+ realQuerySql = "select * from " + curQueryPath;
+ if (!(realIotDBCondition.length() == 0)) {
+ realQuerySql += " where " + realIotDBCondition;
+ }
+ realQuerySql += " align by device";
+ SessionDataSet sessionDataSet = session.executeQueryStatement(realQuerySql);
+ return iotdbAlignByDeviceResultCvtToInfluxdbResult(sessionDataSet);
+ // The following comments refer to the scheme of non align by device
+ // if (realIotDBCondition.isEmpty()) {
+ // realQuerySql = ("select * from " + curQueryPath);
+ // SessionDataSet sessionDataSet = session.executeQueryStatement(realQuerySql);
+ // queryResult = iotdbResultCvtToInfluxdbResult(sessionDataSet);
+ // System.out.println(sessionDataSet.toString());
+ // } else {
+ // //With filter conditions, we can only traverse multiple times
+ // QueryResult lastQueryResult = null;
+ // for (int i = currentQueryMaxTagNum; i <= measurementTagNum; i++) {
+ // if (i != currentQueryMaxTagNum) {
+ // curQueryPath.append(".*");
+ // }
+ // realQuerySql = ("select * from " + curQueryPath + " where " +
+ // realIotDBCondition + " align by device");
+ // SessionDataSet sessionDataSet = null;
+ // try {
+ // sessionDataSet = session.executeQueryStatement(realQuerySql);
+ // } catch (StatementExecutionException e) {
+ // if (e.getStatusCode() == 411) {
+ // //If the timeseries of where do not match, an error of 411 will be
+ // thrown and blocked for printing
+ // System.out.println(e.getMessage());
+ // } else {
+ // throw e;
+ // }
+ // }
+ // //Temporary conversion results
+ // QueryResult tmpQueryResult = iotdbResultCvtToInfluxdbResult(sessionDataSet);
+ // //If it is the first time, it is assigned directly without or operation
+ // if (i == currentQueryMaxTagNum) {
+ // lastQueryResult = tmpQueryResult;
+ // } else {
+ // //Perform the add operation
+ // lastQueryResult =
+ // IotDBInfluxDBUtils.addQueryResultProcess(lastQueryResult, tmpQueryResult);
+ // }
+ // }
+ // queryResult = lastQueryResult;
+ // }
+ }
+
+ /**
+ * Convert align by device query result of iotdb to the query result of influxdb
+ *
+ * @param sessionDataSet iotdb query results to be converted
+ * @return query results in influxdb format
+ */
+ private QueryResult iotdbAlignByDeviceResultCvtToInfluxdbResult(SessionDataSet sessionDataSet)
+ throws IoTDBConnectionException, StatementExecutionException {
+ if (sessionDataSet == null) {
+ return IoTDBInfluxDBUtils.getNullQueryResult();
+ }
+ // generate series
+ QueryResult.Series series = new QueryResult.Series();
+ series.setName(measurement);
+ // gets the reverse map of the tag
+ Map<Integer, String> tagOrderReversed =
+ tagOrders.entrySet().stream()
+ .collect(Collectors.toMap(Map.Entry::getValue, Map.Entry::getKey));
+ int tagSize = tagOrderReversed.size();
+ ArrayList<String> tagList = new ArrayList<>();
+ for (int i = 1; i <= tagSize; i++) {
+ tagList.add(tagOrderReversed.get(i));
+ }
+
+ ArrayList<String> fieldList = new ArrayList<>();
+ for (int i = 1 + tagSize; i < 1 + tagSize + fieldOrders.size(); i++) {
+ fieldList.add(fieldOrdersReversed.get(i));
+ }
+ ArrayList<String> columns = new ArrayList<>();
+ columns.add("time");
+ columns.addAll(tagList);
+ columns.addAll(fieldList);
+ // insert columns into series
+ series.setColumns(columns);
+
+ List<List<Object>> values = new ArrayList<>();
+
+ List<String> iotdbResultColumn = sessionDataSet.getColumnNames();
+ while (sessionDataSet.hasNext()) {
+ Object[] value = new Object[columns.size()];
+
+ RowRecord record = sessionDataSet.next();
+ List<org.apache.iotdb.tsfile.read.common.Field> fields = record.getFields();
+
+ value[0] = record.getTimestamp();
+
+ String deviceName = fields.get(0).getStringValue();
+ String[] deviceNameList = deviceName.split("\\.");
+ for (int i = 3; i < deviceNameList.length; i++) {
+ if (!deviceNameList[i].equals(placeholder)) {
+ value[i - 2] = deviceNameList[i];
+ }
+ }
+ for (int i = 1; i < fields.size(); i++) {
+ Object o = IoTDBInfluxDBUtils.iotdbFiledCvt(fields.get(i));
+ if (o != null) {
+ // insert the value of filed into it
+ value[fieldOrders.get(iotdbResultColumn.get(i + 1))] = o;
+ }
+ }
+ // insert actual value
+ values.add(Arrays.asList(value));
+ }
+ series.setValues(values);
+
+ QueryResult queryResult = new QueryResult();
+ QueryResult.Result result = new QueryResult.Result();
+ result.setSeries(new ArrayList<>(Arrays.asList(series)));
+ queryResult.setResults(new ArrayList<>(Arrays.asList(result)));
+
+ return queryResult;
+ }
+
+ /**
+ * convert the query result of iotdb to the query result of influxdb
+ *
+ * @param sessionDataSet iotdb query results to be converted
+ * @return query results in influxdb format
+ */
+ private QueryResult iotdbResultCvtToInfluxdbResult(SessionDataSet sessionDataSet)
+ throws IoTDBConnectionException, StatementExecutionException {
+ if (sessionDataSet == null) {
+ return IoTDBInfluxDBUtils.getNullQueryResult();
+ }
+ QueryResult.Series series = new QueryResult.Series();
+ series.setName(measurement);
+ Map<Integer, String> tagOrderReversed =
+ tagOrders.entrySet().stream()
+ .collect(Collectors.toMap(Map.Entry::getValue, Map.Entry::getKey));
+ int tagSize = tagOrderReversed.size();
+ ArrayList<String> tagList = new ArrayList<>();
+ for (int i = 1; i <= tagSize; i++) {
+ tagList.add(tagOrderReversed.get(i));
+ }
+
+ ArrayList<String> fieldList = new ArrayList<>();
+ for (int i = 1 + tagSize; i < 1 + tagSize + fieldOrders.size(); i++) {
+ fieldList.add(fieldOrdersReversed.get(i));
+ }
+ ArrayList<String> columns = new ArrayList<>();
+ columns.add("time");
+ columns.addAll(tagList);
+ columns.addAll(fieldList);
+ series.setColumns(columns);
+
+ List<List<Object>> values = new ArrayList<>();
+
+ List<String> iotdbResultColumn = sessionDataSet.getColumnNames();
+ ArrayList<Integer> samePath =
+ IoTDBInfluxDBUtils.getSamePathForList(
+ iotdbResultColumn.subList(1, iotdbResultColumn.size()));
+ while (sessionDataSet.hasNext()) {
+ Object[] value = new Object[columns.size()];
+
+ RowRecord record = sessionDataSet.next();
+ List<org.apache.iotdb.tsfile.read.common.Field> fields = record.getFields();
+ long timestamp = record.getTimestamp();
+ // determine whether all values of the path are null
+ boolean allNull = true;
+ // record the current index of sameList
+ int sameListIndex = 0;
+ for (int i = 0; i < fields.size(); i++) {
+ Object o = IoTDBInfluxDBUtils.iotdbFiledCvt(fields.get(i));
+ if (o != null) {
+ if (allNull) {
+ allNull = false;
+ }
+ // insert the value of filed into it
+ value[fieldOrders.get(IoTDBInfluxDBUtils.getFiledByPath(iotdbResultColumn.get(i + 1)))] =
+ o;
+ }
+ // the same path has been traversed
+ if (i == samePath.get(sameListIndex)) {
+ // if there is non-null in the data, it will be inserted into the actual data, otherwise
+ // it will be skipped directly
+ if (!allNull) {
+ // insert time into value
+ value[0] = timestamp;
+ // then insert the tag in the path into value
+ // plus 1, the zeroth column is time
+ String tmpPathName = iotdbResultColumn.get(i + 1);
+ String[] tmpTags = tmpPathName.split("\\.");
+ for (int j = 3; i < tmpTags.length - 1; i++) {
+ if (!tmpTags[j].equals(placeholder)) {
+ // put into the specified sequence
+ value[j - 2] = tmpTags[j];
+ }
+ }
+ }
+ // insert actual value
+ values.add(Arrays.asList(value));
+ // init value
+ value = new Object[columns.size()];
+ }
+ }
+ }
+ series.setValues(values);
+
+ QueryResult queryResult = new QueryResult();
+ QueryResult.Result result = new QueryResult.Result();
+ result.setSeries(new ArrayList<>(Arrays.asList(series)));
+ queryResult.setResults(new ArrayList<>(Arrays.asList(result)));
+
+ return queryResult;
+ }
+
+ /**
+ * get query results through the syntax tree of influxdb
+ *
+ * @param operator query syntax tree to be processed
+ * @return query results in influxdb format
+ */
+ private QueryResult queryExpr(FilterOperator operator)
+ throws IoTDBConnectionException, StatementExecutionException {
+ if (operator instanceof BasicFunctionOperator) {
+ List<Condition> conditions = new ArrayList<>();
+ conditions.add(
+ IoTDBInfluxDBUtils.getConditionForBasicFunctionOperator(
+ (BasicFunctionOperator) operator));
+ return queryByConditions(conditions);
+ } else {
+ FilterOperator leftOperator = operator.getChildOperators().get(0);
+ FilterOperator rightOperator = operator.getChildOperators().get(1);
+ if (operator.getFilterType() == FilterConstant.FilterType.KW_OR) {
+ return IoTDBInfluxDBUtils.orQueryResultProcess(
+ queryExpr(leftOperator), queryExpr(rightOperator));
+ } else if (operator.getFilterType() == FilterConstant.FilterType.KW_AND) {
+ if (IoTDBInfluxDBUtils.canMergeOperator(leftOperator)
+ && IoTDBInfluxDBUtils.canMergeOperator(rightOperator)) {
+ List<Condition> conditions1 =
+ IoTDBInfluxDBUtils.getConditionsByFilterOperatorOperator(leftOperator);
+ List<Condition> conditions2 =
+ IoTDBInfluxDBUtils.getConditionsByFilterOperatorOperator(rightOperator);
+ conditions1.addAll(conditions2);
+ return queryByConditions(conditions1);
+ } else {
+ return IoTDBInfluxDBUtils.andQueryResultProcess(
+ queryExpr(leftOperator), queryExpr(rightOperator));
+ }
+ }
+ }
+ throw new IllegalArgumentException("unknown operator " + operator.toString());
+ }
+
+ /**
+ * When a new tag appears, it is inserted into the database
+ *
+ * @param measurement inserted measurement
+ * @param tag tag name
+ * @param order tag order
+ */
+ private void updateNewTagIntoDB(String measurement, String tag, int order)
+ throws IoTDBConnectionException, StatementExecutionException {
+ List<String> measurements = new ArrayList<>();
+ List<TSDataType> types = new ArrayList<>();
+ List<Object> values = new ArrayList<>();
+ measurements.add("database_name");
+ measurements.add("measurement_name");
+ measurements.add("tag_name");
+ measurements.add("tag_order");
+ types.add(TSDataType.TEXT);
+ types.add(TSDataType.TEXT);
+ types.add(TSDataType.TEXT);
+ types.add(TSDataType.INT32);
+ values.add(database);
+ values.add(measurement);
+ values.add(tag);
+ values.add(order);
+ session.insertRecord("root.TAG_INFO", System.currentTimeMillis(), measurements, types, values);
+ }
+
+ /**
+ * Query the select result. By default, there are no filter conditions. The functions to be
+ * queried use the built-in iotdb functions
+ *
+ * @param selectComponent select data to query
+ * @return select query result
+ */
+ private QueryResult queryFuncWithoutFilter(SelectComponent selectComponent) {
+ // columns
+ List<String> columns = new ArrayList<>();
+ columns.add(SQLConstant.RESERVED_TIME);
+
+ List<Function> functions = new ArrayList<>();
+ String path = "root." + this.database + "." + this.measurement;
+ for (ResultColumn resultColumn : selectComponent.getResultColumns()) {
+ Expression expression = resultColumn.getExpression();
+ if (expression instanceof FunctionExpression) {
+ String functionName = ((FunctionExpression) expression).getFunctionName();
+ functions.add(
+ FunctionFactory.generateFunctionBySession(
+ functionName, ((FunctionExpression) expression).getExpressions(), session, path));
+ columns.add(functionName);
+ }
+ }
+
+ List<Object> value = new ArrayList<>();
+ List<List<Object>> values = new ArrayList<>();
+ for (Function function : functions) {
+ FunctionValue functionValue = function.calculateByIoTDBFunc();
+ if (value.size() == 0) {
+ value.add(functionValue.getTimestamp());
+ } else {
+ value.set(0, functionValue.getTimestamp());
+ }
+ value.add(functionValue.getValue());
+ }
+ if (selectComponent.isHasAggregationFunction() || selectComponent.isHasMoreFunction()) {
+ value.set(0, 0);
+ }
+ values.add(value);
+
+ // generate series
+ QueryResult queryResult = new QueryResult();
+ QueryResult.Series series = new QueryResult.Series();
+ series.setColumns(columns);
+ series.setValues(values);
+ series.setName(measurement);
+ QueryResult.Result result = new QueryResult.Result();
+ result.setSeries(new ArrayList<>(Arrays.asList(series)));
+ queryResult.setResults(new ArrayList<>(Arrays.asList(result)));
+ return queryResult;
+ }
+
+ /**
+ * further process the obtained query result through the query criteria of select
+ *
+ * @param queryResult query results to be processed
+ * @param selectComponent select conditions to be filtered
+ */
+ private void ProcessSelectComponent(QueryResult queryResult, SelectComponent selectComponent) {
+ // get the row order map of the current data result first
+ List<String> columns = queryResult.getResults().get(0).getSeries().get(0).getColumns();
+ Map<String, Integer> columnOrders = new HashMap<>();
+ for (int i = 0; i < columns.size(); i++) {
+ columnOrders.put(columns.get(i), i);
+ }
+ // get current values
+ List<List<Object>> values = queryResult.getResults().get(0).getSeries().get(0).getValues();
+ // new columns
+ List<String> newColumns = new ArrayList<>();
+ newColumns.add(SQLConstant.RESERVED_TIME);
+
+ // when have function
+ if (selectComponent.isHasFunction()) {
+ List<Function> functions = new ArrayList<>();
+ for (ResultColumn resultColumn : selectComponent.getResultColumns()) {
+ Expression expression = resultColumn.getExpression();
+ if (expression instanceof FunctionExpression) {
+ String functionName = ((FunctionExpression) expression).getFunctionName();
+ functions.add(
+ FunctionFactory.generateFunction(
+ functionName, ((FunctionExpression) expression).getExpressions()));
+ newColumns.add(functionName);
+ } else if (expression instanceof NodeExpression) {
+ String columnName = ((NodeExpression) expression).getName();
+ if (!columnName.equals(SQLConstant.STAR)) {
+ newColumns.add(columnName);
+ } else {
+ newColumns.addAll(columns.subList(1, columns.size()));
+ }
+ }
+ }
+ for (List<Object> value : values) {
+ for (Function function : functions) {
+ List<Expression> expressions = function.getExpressions();
+ if (expressions == null) {
+ throw new IllegalArgumentException("not support param");
+ }
+ NodeExpression parmaExpression = (NodeExpression) expressions.get(0);
+ String parmaName = parmaExpression.getName();
+ if (columnOrders.containsKey(parmaName)) {
+ Object selectedValue = value.get(columnOrders.get(parmaName));
+ Long selectedTimestamp = (Long) value.get(0);
+ if (selectedValue != null) {
+ // selector function
+ if (function instanceof Selector) {
+ ((Selector) function)
+ .updateValueAndRelate(
+ new FunctionValue(selectedValue, selectedTimestamp), value);
+ } else {
+ // aggregate function
+ ((Aggregate) function)
+ .updateValue(new FunctionValue(selectedValue, selectedTimestamp));
+ }
+ }
+ }
+ }
+ }
+ List<Object> value = new ArrayList<>();
+ values = new ArrayList<>();
+ // after the data is constructed, the final results are generated
+ // First, judge whether there are common queries. If there are, a selector function is allowed
+ // without aggregate functions
+ if (selectComponent.isHasCommonQuery()) {
+ Selector selector = (Selector) functions.get(0);
+ List<Object> relatedValue = selector.getRelatedValues();
+ for (String column : newColumns) {
+ if (SQLConstant.getNativeSelectorFunctionNames().contains(column)) {
+ value.add(selector.calculate().getValue());
+ } else {
+ if (relatedValue != null) {
+ value.add(relatedValue.get(columnOrders.get(column)));
+ }
+ }
+ }
+ } else {
+ // If there are no common queries, they are all function queries
+ for (Function function : functions) {
+ if (value.size() == 0) {
+ value.add(function.calculate().getTimestamp());
+ } else {
+ value.set(0, function.calculate().getTimestamp());
+ }
+ value.add(function.calculate().getValue());
+ }
+ if (selectComponent.isHasAggregationFunction() || selectComponent.isHasMoreFunction()) {
+ value.set(0, 0);
+ }
+ }
+ values.add(value);
+ }
+ // if it is not a function query, it is only a common query
+ else if (selectComponent.isHasCommonQuery()) {
+ // start traversing the scope of the select
+ for (ResultColumn resultColumn : selectComponent.getResultColumns()) {
+ Expression expression = resultColumn.getExpression();
+ if (expression instanceof NodeExpression) {
+ // not star case
+ if (!((NodeExpression) expression).getName().equals(SQLConstant.STAR)) {
+ newColumns.add(((NodeExpression) expression).getName());
+ } else {
+ newColumns.addAll(columns.subList(1, columns.size()));
+ }
+ }
+ }
+ List<List<Object>> newValues = new ArrayList();
+ for (List<Object> value : values) {
+ List<Object> tmpValue = new ArrayList();
+ for (String newColumn : newColumns) {
+ tmpValue.add(value.get(columnOrders.get(newColumn)));
+ }
+ newValues.add(tmpValue);
+ }
+ values = newValues;
+ }
+ IoTDBInfluxDBUtils.updateQueryResultColumnValue(
+ queryResult, IoTDBInfluxDBUtils.removeDuplicate(newColumns), values);
+ }
+
+ /**
+ * before each query, first obtain all the field lists in the measurement, and update all the
+ * field lists of the current measurement and the specified order
+ */
+ private void updateFiledOrders() throws IoTDBConnectionException, StatementExecutionException {
+ // first init
+ fieldOrders = new HashMap<>();
+ fieldOrdersReversed = new HashMap<>();
+ String showTimeseriesSql = "show timeseries root." + database + '.' + measurement;
+ SessionDataSet result = session.executeQueryStatement(showTimeseriesSql);
+ int fieldNums = 0;
+ int tagOrderNums = tagOrders.size();
+ while (result.hasNext()) {
+ List<org.apache.iotdb.tsfile.read.common.Field> fields = result.next().getFields();
+ String filed = IoTDBInfluxDBUtils.getFiledByPath(fields.get(0).getStringValue());
+ if (!fieldOrders.containsKey(filed)) {
+ // The corresponding order of fields is 1 + tagnum (the first is timestamp, then all tags,
+ // and finally all fields)
+ fieldOrders.put(filed, tagOrderNums + fieldNums + 1);
+ fieldOrdersReversed.put(tagOrderNums + fieldNums + 1, filed);
+ fieldNums++;
+ }
+ }
+ }
+
+ /**
+ * update current measurement
+ *
+ * @param measurement measurement to be changed
+ */
+ private void updateMeasurement(String measurement) {
+ if (!measurement.equals(this.measurement)) {
+ this.measurement = measurement;
+ tagOrders = measurementTagOrder.get(measurement);
+ if (tagOrders == null) {
+ tagOrders = new HashMap<>();
+ }
+ }
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/IoTDBInfluxDBFactory.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/IoTDBInfluxDBFactory.java
new file mode 100644
index 0000000..86e5752
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/IoTDBInfluxDBFactory.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.influxdb;
+
+import org.influxdb.InfluxDB;
+
+public enum IoTDBInfluxDBFactory {
+ INSTANCE;
+
+ private IoTDBInfluxDBFactory() {}
+
+ public static InfluxDB connect(String url, String username, String password) {
+ IoTDBInfluxDBUtils.checkNonEmptyString(url, "url");
+ IoTDBInfluxDBUtils.checkNonEmptyString(username, "username");
+ return new IoTDBInfluxDB(url, username, password);
+ }
+
+ public static InfluxDB connect(String host, int rpcPort, String userName, String password) {
+ IoTDBInfluxDBUtils.checkNonEmptyString(host, "host");
+ IoTDBInfluxDBUtils.checkNonEmptyString(userName, "username");
+ return new IoTDBInfluxDB(host, rpcPort, userName, password);
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/IoTDBInfluxDBUtils.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/IoTDBInfluxDBUtils.java
new file mode 100644
index 0000000..77ff72b
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/IoTDBInfluxDBUtils.java
@@ -0,0 +1,469 @@
+/*
+ * 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.influxdb;
+
+import org.apache.iotdb.influxdb.qp.constant.FilterConstant;
+import org.apache.iotdb.influxdb.qp.logical.Operator;
+import org.apache.iotdb.influxdb.qp.logical.crud.*;
+import org.apache.iotdb.tsfile.read.common.Field;
+
+import org.influxdb.dto.QueryResult;
+
+import java.util.*;
+
+public final class IoTDBInfluxDBUtils {
+
+ /**
+ * if the first and last of the current string are quotation marks, they are removed
+ *
+ * @param str string to process
+ * @return string after processing
+ */
+ public static String removeQuotation(String str) {
+ if (str.charAt(0) == '"' && str.charAt(str.length() - 1) == '"') {
+ return str.substring(1, str.length() - 1);
+ }
+ return str;
+ }
+
+ /**
+ * take the intersection of the query results of two influxdb
+ *
+ * @param queryResult1 query result 1
+ * @param queryResult2 query result 2
+ * @return intersection of two query results
+ */
+ public static QueryResult andQueryResultProcess(
+ QueryResult queryResult1, QueryResult queryResult2) {
+ if (checkQueryResultNull(queryResult1) || checkQueryResultNull(queryResult2)) {
+ return getNullQueryResult();
+ }
+ if (!checkSameQueryResult(queryResult1, queryResult2)) {
+ System.out.println("QueryResult1 and QueryResult2 is not same attribute");
+ return queryResult1;
+ }
+ List<List<Object>> values1 = queryResult1.getResults().get(0).getSeries().get(0).getValues();
+ List<List<Object>> values2 = queryResult2.getResults().get(0).getSeries().get(0).getValues();
+ List<List<Object>> sameValues = new ArrayList<>();
+ for (List<Object> value1 : values1) {
+ for (List<Object> value2 : values2) {
+ boolean allEqual = true;
+ for (int t = 0; t < value1.size(); t++) {
+ // if there is any inequality, skip the current J
+ if (!checkEqualsContainNull(value1.get(t), value2.get(t))) {
+ allEqual = false;
+ break;
+ }
+ }
+ // at this time, the matching is completed. If it is completely equal
+ if (allEqual) {
+ sameValues.add(value1);
+ }
+ }
+ }
+ updateQueryResultValue(queryResult1, sameValues);
+ return queryResult1;
+ }
+
+ /**
+ * union the query results of two influxdb
+ *
+ * @param queryResult1 query result 1
+ * @param queryResult2 query result 2
+ * @return union of two query results
+ */
+ public static QueryResult orQueryResultProcess(
+ QueryResult queryResult1, QueryResult queryResult2) {
+ if (checkQueryResultNull(queryResult1)) {
+ return queryResult2;
+ } else if (checkQueryResultNull(queryResult2)) {
+ return queryResult1;
+ }
+ if (!checkSameQueryResult(queryResult1, queryResult2)) {
+ System.out.println("QueryResult1 and QueryResult2 is not same attribute");
+ return queryResult1;
+ }
+ List<List<Object>> values1 = queryResult1.getResults().get(0).getSeries().get(0).getValues();
+ List<List<Object>> values2 = queryResult2.getResults().get(0).getSeries().get(0).getValues();
+ List<List<Object>> notSameValuesInValues1 = new ArrayList<>();
+ for (List<Object> value1 : values1) {
+ boolean allNotEqual = true;
+ for (List<Object> value2 : values2) {
+ boolean notEqual = false;
+ for (int t = 0; t < value1.size(); t++) {
+ // if there is any inequality, skip the current J
+ if (!checkEqualsContainNull(value1.get(t), value2.get(t))) {
+ notEqual = true;
+ break;
+ }
+ }
+ if (!notEqual) {
+ allNotEqual = false;
+ break;
+ }
+ }
+ if (allNotEqual) {
+ notSameValuesInValues1.add(value1);
+ }
+ }
+ // values2 plus a different valueList
+ values2.addAll(notSameValuesInValues1);
+ updateQueryResultValue(queryResult1, values2);
+ return queryResult1;
+ }
+
+ /**
+ * sum the query results of two influxdb (that is, by default, they will not have duplicate data
+ * and will be directly added together)
+ *
+ * @param queryResult1 query result1
+ * @param queryResult2 query result2
+ * @return sum of two query results
+ */
+ public static QueryResult addQueryResultProcess(
+ QueryResult queryResult1, QueryResult queryResult2) {
+ if (checkQueryResultNull(queryResult1)) {
+ return queryResult2;
+ } else if (checkQueryResultNull(queryResult2)) {
+ return queryResult1;
+ }
+ if (!checkSameQueryResult(queryResult1, queryResult2)) {
+ System.out.println("QueryResult1 and QueryResult2 is not same attribute");
+ return queryResult1;
+ }
+ List<List<Object>> values1 = queryResult1.getResults().get(0).getSeries().get(0).getValues();
+ List<List<Object>> values2 = queryResult2.getResults().get(0).getSeries().get(0).getValues();
+ values1.addAll(values2);
+ updateQueryResultValue(queryResult1, values1);
+ return queryResult1;
+ }
+
+ /**
+ * update the new values to the query results of influxdb
+ *
+ * @param queryResult influxdb query results to be updated
+ * @param updateValues values to be updated
+ */
+ private static void updateQueryResultValue(
+ QueryResult queryResult, List<List<Object>> updateValues) {
+ List<QueryResult.Result> results = queryResult.getResults();
+ QueryResult.Result result = results.get(0);
+ List<QueryResult.Series> series = results.get(0).getSeries();
+ QueryResult.Series serie = series.get(0);
+
+ serie.setValues(updateValues);
+ series.set(0, serie);
+ result.setSeries(series);
+ results.set(0, result);
+ }
+
+ /**
+ * update the new values to the query results of influxdb
+ *
+ * @param queryResult influxdb query results to be updated
+ * @param columns columns to be updated
+ * @param updateValues values to be updated
+ */
+ public static void updateQueryResultColumnValue(
+ QueryResult queryResult, List<String> columns, List<List<Object>> updateValues) {
+ List<QueryResult.Result> results = queryResult.getResults();
+ QueryResult.Result result = results.get(0);
+ List<QueryResult.Series> series = results.get(0).getSeries();
+ QueryResult.Series serie = series.get(0);
+
+ serie.setValues(updateValues);
+ serie.setColumns(columns);
+ series.set(0, serie);
+ result.setSeries(series);
+ results.set(0, result);
+ }
+
+ /**
+ * check whether the query results of two influxdb belong to the same query, that is, whether the
+ * measurement and columns are consistent
+ *
+ * @param queryResult1 query result1
+ * @param queryResult2 query result2
+ * @return belong to the same query
+ */
+ private static boolean checkSameQueryResult(QueryResult queryResult1, QueryResult queryResult2) {
+ return queryResult1
+ .getResults()
+ .get(0)
+ .getSeries()
+ .get(0)
+ .getName()
+ .equals(queryResult2.getResults().get(0).getSeries().get(0).getName())
+ && checkSameStringList(
+ queryResult1.getResults().get(0).getSeries().get(0).getColumns(),
+ queryResult2.getResults().get(0).getSeries().get(0).getColumns());
+ }
+
+ /**
+ * determine whether the two string lists are the same
+ *
+ * @param list1 first list to compare
+ * @param list2 second list to compare
+ * @return Is it the same
+ */
+ private static boolean checkSameStringList(List<String> list1, List<String> list2) {
+ if (list1.size() != list2.size()) {
+ return false;
+ } else {
+ for (int i = 0; i < list1.size(); i++) {
+ if (!list1.get(i).equals(list2.get(i))) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ /**
+ * judge whether the subtrees of the syntax tree have or operations. If not, the query can be
+ * merged
+ *
+ * @param operator subtree to judge
+ * @return can merge queries
+ */
+ public static boolean canMergeOperator(FilterOperator operator) {
+ if (operator instanceof BasicFunctionOperator) {
+ return true;
+ } else {
+ if (operator.getFilterType() == FilterConstant.FilterType.KW_OR) {
+ return false;
+ } else {
+ FilterOperator leftOperator = operator.getChildOperators().get(0);
+ FilterOperator rightOperator = operator.getChildOperators().get(1);
+ return canMergeOperator(leftOperator) && canMergeOperator(rightOperator);
+ }
+ }
+ }
+
+ /**
+ * generate query conditions through the syntax tree (if you enter this function, it means that it
+ * must be a syntax tree that can be merged, and there is no or)
+ *
+ * @param filterOperator the syntax tree of query criteria needs to be generated
+ * @return condition list
+ */
+ public static List<Condition> getConditionsByFilterOperatorOperator(
+ FilterOperator filterOperator) {
+ if (filterOperator instanceof BasicFunctionOperator) {
+ // It must be a non-or situation
+ List<Condition> conditions = new ArrayList<>();
+ conditions.add(getConditionForBasicFunctionOperator((BasicFunctionOperator) filterOperator));
+ return conditions;
+ } else {
+ FilterOperator leftOperator = filterOperator.getChildOperators().get(0);
+ FilterOperator rightOperator = filterOperator.getChildOperators().get(1);
+ List<Condition> conditions1 = getConditionsByFilterOperatorOperator(leftOperator);
+ List<Condition> conditions2 = getConditionsByFilterOperatorOperator(rightOperator);
+ conditions1.addAll(conditions2);
+ return conditions1;
+ }
+ }
+
+ /**
+ * conditions are generated from subtrees of unique conditions
+ *
+ * @param basicFunctionOperator subtree to generate condition
+ * @return corresponding conditions
+ */
+ public static Condition getConditionForBasicFunctionOperator(
+ BasicFunctionOperator basicFunctionOperator) {
+ Condition condition = new Condition();
+ condition.setFilterType(basicFunctionOperator.getFilterType());
+ condition.setValue(basicFunctionOperator.getKeyName());
+ condition.setLiteral(basicFunctionOperator.getValue());
+ return condition;
+ }
+
+ /**
+ * get the last node through the path in iotdb
+ *
+ * @param path path to process
+ * @return last node
+ */
+ public static String getFiledByPath(String path) {
+ String[] tmpList = path.split("\\.");
+ return tmpList[tmpList.length - 1];
+ }
+
+ /**
+ * generate a list of the same layer path from a list of rows example: input:{root.d.s.x.s.field1
+ * , root.d.s.x.s.field2 , root.d.s.xx.ss.filed3 , root.d.s.xx.ss.filed1 ,
+ * root.d.s.xxx.sss.field4} output:{1,3,4} corresponding to the subscripts 1, 3 and 4 of the list.
+ * That is, the last subscript of the current same layer path. It is easy to distinguish the same
+ * set of data
+ *
+ * @param columnNames list of row names to be generated
+ * @return the collection of the last subscript in the same layer path
+ */
+ public static ArrayList<Integer> getSamePathForList(List<String> columnNames) {
+ ArrayList<Integer> list = new ArrayList<>();
+ // record the last result to determine whether it is repeated with the current result
+ String lastPath = null;
+ for (int i = 1; i < columnNames.size(); i++) {
+ String columnName = columnNames.get(i);
+ String path = columnName.substring(0, columnName.lastIndexOf("."));
+ if (i == 1) {
+ lastPath = path;
+ } else {
+ if (!lastPath.equals(path)) {
+ list.add(i - 1);
+ lastPath = path;
+ }
+ }
+ }
+ list.add(columnNames.size() - 1);
+ return list;
+ }
+
+ /**
+ * convert the value of field in iotdb to object
+ *
+ * @param field filed to be converted
+ * @return value stored in field
+ */
+ public static Object iotdbFiledCvt(Field field) {
+ if (field.getDataType() == null) {
+ return null;
+ }
+ switch (field.getDataType()) {
+ case TEXT:
+ return field.getStringValue();
+ case INT64:
+ return field.getLongV();
+ case INT32:
+ return field.getIntV();
+ case DOUBLE:
+ return field.getDoubleV();
+ case FLOAT:
+ return field.getFloatV();
+ case BOOLEAN:
+ return field.getBoolV();
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * get a null query result
+ *
+ * @return null queryResult
+ */
+ public static QueryResult getNullQueryResult() {
+ QueryResult queryResult = new QueryResult();
+ QueryResult.Result result = new QueryResult.Result();
+ queryResult.setResults(Arrays.asList(result));
+ return queryResult;
+ }
+
+ /**
+ * checks whether query result is null in the specified way
+ *
+ * @param queryResult query result to be checked
+ * @return is null
+ */
+ public static boolean checkQueryResultNull(QueryResult queryResult) {
+ return queryResult.getResults().get(0).getSeries() == null;
+ }
+
+ /**
+ * judge whether two instances are equal, and allow null check
+ *
+ * @param o1 object1
+ * @param o2 object2
+ * @return are they equal
+ */
+ public static boolean checkEqualsContainNull(Object o1, Object o2) {
+ if (o1 == null && o2 == null) {
+ return true;
+ } else if (o1 == null) {
+ return false;
+ } else if (o2 == null) {
+ return false;
+ } else {
+ return o1.equals(o2);
+ }
+ }
+
+ /**
+ * check whether the field is empty. If it is empty, an error will be thrown
+ *
+ * @param string string to check
+ * @param name prompt information in error throwing
+ */
+ public static void checkNonEmptyString(String string, String name)
+ throws IllegalArgumentException {
+ if (string == null || string.isEmpty()) {
+ throw new IllegalArgumentException("Expecting a non-empty string for " + name);
+ }
+ }
+
+ /**
+ * check whether the operator generated by SQL is a valid query statement. If it is not, an error
+ * will be thrown
+ *
+ * @param operator operator to check
+ */
+ public static void checkQueryOperator(Operator operator) {
+ if (!(operator instanceof QueryOperator)) {
+ throw new IllegalArgumentException("not query sql");
+ }
+ SelectComponent selectComponent = ((QueryOperator) operator).getSelectComponent();
+ if (selectComponent.isHasMoreSelectorFunction() && selectComponent.isHasCommonQuery()) {
+ throw new IllegalArgumentException(
+ "ERR: mixing multiple selector functions with tags or fields is not supported");
+ }
+ if (selectComponent.isHasAggregationFunction() && selectComponent.isHasCommonQuery()) {
+ throw new IllegalArgumentException(
+ "ERR: mixing aggregate and non-aggregate queries is not supported");
+ }
+ }
+
+ /**
+ * remove string list duplicate names
+ *
+ * @param strings the list of strings with duplicate names needs to be removed
+ * @return list of de duplicated strings
+ */
+ public static List<String> removeDuplicate(List<String> strings) {
+ Map<String, Integer> nameNums = new HashMap<>();
+ List<String> result = new ArrayList<>();
+ for (String tmpString : strings) {
+ if (!nameNums.containsKey(tmpString)) {
+ nameNums.put(tmpString, 1);
+ result.add(tmpString);
+ } else {
+ int nums = nameNums.get(tmpString);
+ result.add(tmpString + "_" + nums);
+ nameNums.put(tmpString, nums + 1);
+ }
+ }
+ return result;
+ }
+
+ public static String generateFunctionSql(String functionName, String functionParam, String path) {
+ return String.format(
+ "select %s(%s) from %s align by device", functionName, functionParam, path);
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/example/InfluxDBExample.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/example/InfluxDBExample.java
new file mode 100644
index 0000000..8aa717b
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/example/InfluxDBExample.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.influxdb.example;
+
+import org.apache.iotdb.influxdb.IoTDBInfluxDBFactory;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+
+import org.influxdb.InfluxDB;
+import org.influxdb.dto.Point;
+import org.influxdb.dto.Query;
+import org.influxdb.dto.QueryResult;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class InfluxDBExample {
+ private static InfluxDB influxDB;
+
+ public static void main(String[] args) throws Exception {
+ // init
+ influxDB = IoTDBInfluxDBFactory.connect("http://127.0.0.1:6667", "root", "root");
+ // create database
+ influxDB.createDatabase("database");
+ // set database
+ influxDB.setDatabase("database");
+
+ insertData();
+ queryData();
+ }
+
+ // insert data
+ public static void insertData() throws IoTDBConnectionException, StatementExecutionException {
+
+ // insert the build parameter to construct the influxdb
+ Point.Builder builder = Point.measurement("student");
+ Map<String, String> tags = new HashMap<>();
+ Map<String, Object> fields = new HashMap<>();
+ tags.put("name", "xie");
+ tags.put("sex", "m");
+ fields.put("score", 87);
+ fields.put("tel", "110");
+ fields.put("country", "china");
+ builder.tag(tags);
+ builder.fields(fields);
+ builder.time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
+ Point point = builder.build();
+ // after the build construction is completed, start writing
+ influxDB.write(point);
+
+ builder = Point.measurement("student");
+ tags = new HashMap<>();
+ fields = new HashMap<>();
+ tags.put("name", "xie");
+ tags.put("sex", "m");
+ tags.put("province", "anhui");
+ fields.put("score", 99);
+ fields.put("country", "china");
+ builder.tag(tags);
+ builder.fields(fields);
+ builder.time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
+ point = builder.build();
+ influxDB.write(point);
+ }
+
+ // query data
+ private static void queryData() throws Exception {
+
+ Query query;
+ QueryResult result;
+
+ // the selector query is parallel to the field value
+ query =
+ new Query(
+ "select * from student where (name=\"xie\" and sex=\"m\")or time<now()-7d", "database");
+ result = influxDB.query(query);
+ System.out.println("query1 result:" + result.getResults().get(0).getSeries().get(0).toString());
+
+ // use iotdb built-in func
+ query =
+ new Query(
+ "select max(score),min(score),sum(score),count(score),spread(score),mean(score),first(score),last(score) from student ",
+ "database");
+ result = influxDB.query(query);
+ System.out.println("query2 result:" + result.getResults().get(0).getSeries().get(0).toString());
+
+ // aggregate query and selector query are parallel
+ query =
+ new Query(
+ "select count(score),first(score),last(country),max(score),mean(score),median(score),min(score),mode(score),spread(score),stddev(score),sum(score) from student where (name=\"xie\" and sex=\"m\")or score<99",
+ "database");
+ result = influxDB.query(query);
+ System.out.println("query3 result:" + result.getResults().get(0).getSeries().get(0).toString());
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/constant/FilterConstant.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/constant/FilterConstant.java
new file mode 100644
index 0000000..6e8219e
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/constant/FilterConstant.java
@@ -0,0 +1,95 @@
+/*
+ * 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.influxdb.qp.constant;
+
+import org.apache.iotdb.influxdb.qp.sql.InfluxDBLexer;
+
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.Map;
+
+public class FilterConstant {
+
+ public static final Map<Integer, FilterType> lexerToFilterType = new HashMap<>();
+ public static final Map<FilterType, String> filterSymbol = new EnumMap<>(FilterType.class);
+ public static final Map<FilterType, String> filterNames = new EnumMap<>(FilterType.class);
+ public static final Map<FilterType, FilterType> filterReverseWords =
+ new EnumMap<>(FilterType.class);
+
+ public enum FilterType {
+ KW_AND,
+ KW_OR,
+ KW_NOT,
+
+ EQUAL,
+ NOTEQUAL,
+ LESSTHANOREQUALTO,
+ LESSTHAN,
+ GREATERTHANOREQUALTO,
+ GREATERTHAN,
+ IN
+ }
+
+ static {
+ lexerToFilterType.put(InfluxDBLexer.OPERATOR_EQ, FilterType.EQUAL);
+ lexerToFilterType.put(InfluxDBLexer.OPERATOR_NEQ, FilterType.NOTEQUAL);
+ lexerToFilterType.put(InfluxDBLexer.OPERATOR_LTE, FilterType.LESSTHANOREQUALTO);
+ lexerToFilterType.put(InfluxDBLexer.OPERATOR_LT, FilterType.LESSTHAN);
+ lexerToFilterType.put(InfluxDBLexer.OPERATOR_GTE, FilterType.GREATERTHANOREQUALTO);
+ lexerToFilterType.put(InfluxDBLexer.OPERATOR_GT, FilterType.GREATERTHAN);
+ lexerToFilterType.put(InfluxDBLexer.OPERATOR_IN, FilterType.IN);
+ }
+
+ static {
+ filterSymbol.put(FilterType.KW_AND, "&");
+ filterSymbol.put(FilterType.KW_OR, "|");
+ filterSymbol.put(FilterType.KW_NOT, "!");
+ filterSymbol.put(FilterType.EQUAL, "=");
+ filterSymbol.put(FilterType.NOTEQUAL, "<>");
+ filterSymbol.put(FilterType.LESSTHANOREQUALTO, "<=");
+ filterSymbol.put(FilterType.LESSTHAN, "<");
+ filterSymbol.put(FilterType.GREATERTHANOREQUALTO, ">=");
+ filterSymbol.put(FilterType.GREATERTHAN, ">");
+ }
+
+ static {
+ filterNames.put(FilterType.KW_AND, "and");
+ filterNames.put(FilterType.KW_OR, "or");
+ filterNames.put(FilterType.KW_NOT, "not");
+ filterNames.put(FilterType.EQUAL, "equal");
+ filterNames.put(FilterType.NOTEQUAL, "not_equal");
+ filterNames.put(FilterType.LESSTHANOREQUALTO, "lessthan_or_equalto");
+ filterNames.put(FilterType.LESSTHAN, "lessthan");
+ filterNames.put(FilterType.GREATERTHANOREQUALTO, "greaterthan_or_equalto");
+ filterNames.put(FilterType.GREATERTHAN, "greaterthan");
+ filterNames.put(FilterType.IN, "in");
+ }
+
+ static {
+ filterReverseWords.put(FilterType.KW_AND, FilterType.KW_OR);
+ filterReverseWords.put(FilterType.KW_OR, FilterType.KW_AND);
+ filterReverseWords.put(FilterType.EQUAL, FilterType.NOTEQUAL);
+ filterReverseWords.put(FilterType.NOTEQUAL, FilterType.EQUAL);
+ filterReverseWords.put(FilterType.LESSTHAN, FilterType.GREATERTHANOREQUALTO);
+ filterReverseWords.put(FilterType.GREATERTHANOREQUALTO, FilterType.LESSTHAN);
+ filterReverseWords.put(FilterType.LESSTHANOREQUALTO, FilterType.GREATERTHAN);
+ filterReverseWords.put(FilterType.GREATERTHAN, FilterType.LESSTHANOREQUALTO);
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/constant/SQLConstant.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/constant/SQLConstant.java
new file mode 100644
index 0000000..f7b24d1
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/constant/SQLConstant.java
@@ -0,0 +1,241 @@
+/*
+ * 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.influxdb.qp.constant;
+
+import java.util.*;
+
+/** this class contains several constants used in SQL. */
+@SuppressWarnings("unused") // some fields are for future features
+public class SQLConstant {
+
+ private SQLConstant() {
+ // forbidding instantiation
+ }
+
+ private static final String[] SINGLE_ROOT_ARRAY = {"root"};
+ private static final String[] SINGLE_TIME_ARRAY = {"time"};
+ public static final String STAR = "*";
+ public static final String ALIGNBY_DEVICE_COLUMN_NAME = "Device";
+ public static final String RESERVED_TIME = "time";
+ public static final String IS_AGGREGATION = "IS_AGGREGATION";
+ public static final String NOW_FUNC = "now()";
+ public static final String START_TIME_STR = "1970-1-01T00:00:00";
+
+ public static final String LINE_FEED_SIGNAL = "\n";
+ public static final String ROOT = "root";
+ public static final String METADATA_PARAM_EQUAL = "=";
+ public static final String QUOTE = "'";
+ public static final String DQUOTE = "\"";
+ public static final String BOOLEAN_TRUE = "true";
+ public static final String BOOLEAN_FALSE = "false";
+ public static final String BOOLEAN_TRUE_NUM = "1";
+ public static final String BOOLEAN_FALSE_NUM = "0";
+
+ // names of aggregations
+ public static final String MIN_TIME = "min_time";
+ public static final String MAX_TIME = "max_time";
+
+ public static final String MAX = "max";
+ public static final String MIN = "min";
+
+ public static final String EXTREME = "extreme";
+
+ public static final String FIRST = "first";
+ public static final String LAST = "last";
+
+ public static final String COUNT = "count";
+ public static final String AVG = "avg";
+ public static final String MEAN = "mean";
+ public static final String MEDIAN = "median";
+ public static final String MODE = "mode";
+ public static final String SPREAD = "spread";
+ public static final String STDDEV = "stddev";
+ public static final String SUM = "sum";
+
+ public static final String ALL = "all";
+
+ private static final Set<String> NATIVE_FUNCTION_NAMES =
+ new HashSet<>(
+ Arrays.asList(MIN, MAX, FIRST, LAST, MEAN, COUNT, MEDIAN, MODE, SPREAD, STDDEV, SUM));
+
+ private static final Set<String> NATIVE_SELECTOR_FUNCTION_NAMES =
+ new HashSet<>(Arrays.asList(MIN, MAX, FIRST, LAST));
+
+ public static final int TOK_WHERE = 23;
+ public static final int TOK_INSERT = 24;
+ public static final int TOK_DELETE = 25;
+ public static final int TOK_UPDATE = 26;
+ public static final int TOK_QUERY = 27;
+
+ public static final int TOK_CREATE_INDEX = 31;
+ public static final int TOK_DROP_INDEX = 32;
+ public static final int TOK_QUERY_INDEX = 33;
+
+ public static final int TOK_GRANT_WATERMARK_EMBEDDING = 34;
+ public static final int TOK_REVOKE_WATERMARK_EMBEDDING = 35;
+
+ public static final int TOK_AUTHOR_CREATE = 41;
+ public static final int TOK_AUTHOR_DROP = 42;
+ public static final int TOK_AUTHOR_GRANT = 43;
+ public static final int TOK_AUTHOR_REVOKE = 44;
+ public static final int TOK_AUTHOR_UPDATE_USER = 46;
+
+ public static final int TOK_DATALOAD = 45;
+
+ public static final int TOK_METADATA_CREATE = 51;
+ public static final int TOK_METADATA_DELETE = 52;
+ public static final int TOK_METADATA_SET_FILE_LEVEL = 53;
+ public static final int TOK_PROPERTY_CREATE = 54;
+ public static final int TOK_PROPERTY_ADD_LABEL = 55;
+ public static final int TOK_PROPERTY_DELETE_LABEL = 56;
+ public static final int TOK_PROPERTY_LINK = 57;
+ public static final int TOK_PROPERTY_UNLINK = 58;
+ public static final int TOK_LIST = 59;
+
+ public static final int TOK_DURATION = 60;
+ public static final int TOK_DATE_EXPR = 61;
+ public static final int TOK_METADATA_DELETE_FILE_LEVEL = 62;
+
+ public static final int TOK_SET = 63;
+ public static final int TOK_UNSET = 64;
+ public static final int TOK_SHOW = 65;
+ public static final int TOK_LOAD_CONFIGURATION = 66;
+
+ public static final int TOK_FLUSH_TASK_INFO = 67;
+ public static final int TOK_LOAD_FILES = 69;
+ public static final int TOK_REMOVE_FILE = 70;
+ public static final int TOK_MOVE_FILE = 71;
+ public static final int TOK_VERSION = 72;
+ public static final int TOK_TIMESERIES = 73;
+ public static final int TOK_STORAGE_GROUP = 74;
+ public static final int TOK_CHILD_PATHS = 75;
+ public static final int TOK_DEVICES = 76;
+ public static final int TOK_COUNT_TIMESERIES = 77;
+ public static final int TOK_COUNT_NODE_TIMESERIES = 78;
+ public static final int TOK_COUNT_NODES = 79;
+
+ public static final int TOK_METADATA_ALTER = 80;
+
+ public static final int TOK_FLUSH = 81;
+ public static final int TOK_MERGE = 82;
+ public static final int TOK_FULL_MERGE = 83;
+
+ public static final int TOK_CLEAR_CACHE = 84;
+
+ public static final int TOK_LOAD_CONFIGURATION_GLOBAL = 85;
+ public static final int TOK_LOAD_CONFIGURATION_LOCAL = 86;
+
+ public static final int TOK_SHOW_MERGE_STATUS = 87;
+ public static final int TOK_DELETE_PARTITION = 88;
+
+ public static final int TOK_CREATE_SCHEMA_SNAPSHOT = 89;
+ public static final int TOK_TRACING = 91;
+
+ public static final int TOK_FUNCTION_CREATE = 92;
+ public static final int TOK_FUNCTION_DROP = 93;
+ public static final int TOK_SHOW_FUNCTIONS = 94;
+
+ public static final int TOK_COUNT_DEVICES = 95;
+ public static final int TOK_COUNT_STORAGE_GROUP = 96;
+ public static final int TOK_QUERY_PROCESSLIST = 97;
+ public static final int TOK_KILL_QUERY = 98;
+
+ public static final int TOK_CHILD_NODES = 99;
+
+ public static final int TOK_TRIGGER_CREATE = 100;
+ public static final int TOK_TRIGGER_DROP = 101;
+ public static final int TOK_TRIGGER_START = 102;
+ public static final int TOK_TRIGGER_STOP = 103;
+ public static final int TOK_SHOW_TRIGGERS = 104;
+ public static final int TOK_LOCK_INFO = 105;
+
+ public static final Map<Integer, String> tokenNames = new HashMap<>();
+
+ public static String[] getSingleRootArray() {
+ return SINGLE_ROOT_ARRAY;
+ }
+
+ public static String[] getSingleTimeArray() {
+ return SINGLE_TIME_ARRAY;
+ }
+
+ static {
+ tokenNames.put(TOK_WHERE, "TOK_WHERE");
+ tokenNames.put(TOK_INSERT, "TOK_INSERT");
+ tokenNames.put(TOK_DELETE, "TOK_DELETE");
+ tokenNames.put(TOK_UPDATE, "TOK_UPDATE");
+ tokenNames.put(TOK_QUERY, "TOK_QUERY");
+
+ tokenNames.put(TOK_AUTHOR_CREATE, "TOK_AUTHOR_CREATE");
+ tokenNames.put(TOK_AUTHOR_DROP, "TOK_AUTHOR_DROP");
+ tokenNames.put(TOK_AUTHOR_GRANT, "TOK_AUTHOR_GRANT");
+ tokenNames.put(TOK_AUTHOR_REVOKE, "TOK_AUTHOR_REVOKE");
+ tokenNames.put(TOK_AUTHOR_UPDATE_USER, "TOK_AUTHOR_UPDATE_USER");
+ tokenNames.put(TOK_DATALOAD, "TOK_DATALOAD");
+
+ tokenNames.put(TOK_METADATA_CREATE, "TOK_METADATA_CREATE");
+ tokenNames.put(TOK_METADATA_DELETE, "TOK_METADATA_DELETE");
+ tokenNames.put(TOK_METADATA_SET_FILE_LEVEL, "TOK_METADATA_SET_FILE_LEVEL");
+ tokenNames.put(TOK_METADATA_DELETE_FILE_LEVEL, "TOK_METADATA_DELETE_FILE_LEVEL");
+ tokenNames.put(TOK_PROPERTY_CREATE, "TOK_PROPERTY_CREATE");
+ tokenNames.put(TOK_PROPERTY_ADD_LABEL, "TOK_PROPERTY_ADD_LABEL");
+ tokenNames.put(TOK_PROPERTY_DELETE_LABEL, "TOK_PROPERTY_DELETE_LABEL");
+ tokenNames.put(TOK_PROPERTY_LINK, "TOK_PROPERTY_LINK");
+ tokenNames.put(TOK_PROPERTY_UNLINK, "TOK_PROPERTY_UNLINK");
+
+ tokenNames.put(TOK_LIST, "TOK_LIST");
+ tokenNames.put(TOK_SET, "TOK_SET");
+ tokenNames.put(TOK_UNSET, "TOK_UNSET");
+ tokenNames.put(TOK_SHOW, "TOK_SHOW");
+
+ tokenNames.put(TOK_LOAD_CONFIGURATION, "TOK_LOAD_CONFIGURATION");
+ tokenNames.put(TOK_FLUSH_TASK_INFO, "TOK_FLUSH_TASK_INFO");
+
+ tokenNames.put(TOK_LOAD_FILES, "TOK_LOAD_FILES");
+ tokenNames.put(TOK_REMOVE_FILE, "TOK_REMOVE_FILE");
+ tokenNames.put(TOK_MOVE_FILE, "TOK_MOVE_FILE");
+
+ tokenNames.put(TOK_SHOW_MERGE_STATUS, "TOK_SHOW_MERGE_STATUS");
+ tokenNames.put(TOK_DELETE_PARTITION, "TOK_DELETE_PARTITION");
+
+ tokenNames.put(TOK_TRACING, "TOK_TRACING");
+
+ tokenNames.put(TOK_FUNCTION_CREATE, "TOK_FUNCTION_CREATE");
+ tokenNames.put(TOK_FUNCTION_DROP, "TOK_FUNCTION_DROP");
+ tokenNames.put(TOK_SHOW_FUNCTIONS, "TOK_SHOW_FUNCTIONS");
+
+ tokenNames.put(TOK_CREATE_INDEX, "TOK_CREATE_INDEX");
+ tokenNames.put(TOK_DROP_INDEX, "TOK_DROP_INDEX");
+ tokenNames.put(TOK_QUERY_INDEX, "TOK_QUERY_INDEX");
+
+ tokenNames.put(TOK_TRIGGER_CREATE, "TOK_TRIGGER_CREATE");
+ tokenNames.put(TOK_TRIGGER_DROP, "TOK_TRIGGER_DROP");
+ tokenNames.put(TOK_TRIGGER_START, "TOK_TRIGGER_START");
+ tokenNames.put(TOK_TRIGGER_STOP, "TOK_TRIGGER_STOP");
+ tokenNames.put(TOK_SHOW_TRIGGERS, "TOK_SHOW_TRIGGERS");
+ }
+
+ public static Set<String> getNativeFunctionNames() {
+ return NATIVE_FUNCTION_NAMES;
+ }
+
+ public static Set<String> getNativeSelectorFunctionNames() {
+ return NATIVE_SELECTOR_FUNCTION_NAMES;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/Operator.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/Operator.java
new file mode 100644
index 0000000..e4a2ae2
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/Operator.java
@@ -0,0 +1,160 @@
+/*
+ * 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.influxdb.qp.logical;
+
+import org.apache.iotdb.influxdb.qp.constant.SQLConstant;
+
+/** This class is a superclass of all operator. */
+public abstract class Operator {
+
+ // operator type in int format
+ protected int tokenIntType;
+ // flag of "explain"
+ protected boolean isDebug;
+
+ protected OperatorType operatorType = OperatorType.NULL;
+
+ protected Operator(int tokenIntType) {
+ this.tokenIntType = tokenIntType;
+ this.isDebug = false;
+ }
+
+ public OperatorType getType() {
+ return operatorType;
+ }
+
+ public boolean isQuery() {
+ return operatorType == OperatorType.QUERY;
+ }
+
+ public int getTokenIntType() {
+ return tokenIntType;
+ }
+
+ public void setOperatorType(OperatorType operatorType) {
+ this.operatorType = operatorType;
+ }
+
+ public boolean isDebug() {
+ return isDebug;
+ }
+
+ public void setDebug(boolean debug) {
+ isDebug = debug;
+ }
+
+ @Override
+ public String toString() {
+ return SQLConstant.tokenNames.get(tokenIntType);
+ }
+
+ /** If you want to add new OperatorType, you must add it in the last. */
+ public enum OperatorType {
+ NULL,
+
+ AUTHOR,
+ LOAD_DATA,
+ CREATE_USER,
+ DELETE_USER,
+ MODIFY_PASSWORD,
+ GRANT_USER_PRIVILEGE,
+ REVOKE_USER_PRIVILEGE,
+ GRANT_USER_ROLE,
+ REVOKE_USER_ROLE,
+ CREATE_ROLE,
+ DELETE_ROLE,
+ GRANT_ROLE_PRIVILEGE,
+ REVOKE_ROLE_PRIVILEGE,
+ LIST_USER,
+ LIST_ROLE,
+ LIST_USER_PRIVILEGE,
+ LIST_ROLE_PRIVILEGE,
+ LIST_USER_ROLES,
+ LIST_ROLE_USERS,
+ GRANT_WATERMARK_EMBEDDING,
+ REVOKE_WATERMARK_EMBEDDING,
+
+ SET_STORAGE_GROUP,
+ DELETE_STORAGE_GROUP,
+ CREATE_TIMESERIES,
+ CREATE_ALIGNED_TIMESERIES,
+ CREATE_MULTI_TIMESERIES,
+ DELETE_TIMESERIES,
+ ALTER_TIMESERIES,
+ CHANGE_ALIAS,
+ CHANGE_TAG_OFFSET,
+
+ INSERT,
+ BATCH_INSERT,
+ BATCH_INSERT_ROWS,
+ BATCH_INSERT_ONE_DEVICE,
+ MULTI_BATCH_INSERT,
+
+ DELETE,
+
+ QUERY,
+ LAST,
+ GROUP_BY_TIME,
+ GROUP_BY_FILL,
+ AGGREGATION,
+ FILL,
+ UDAF,
+ UDTF,
+
+ CREATE_FUNCTION,
+ DROP_FUNCTION,
+
+ SHOW,
+ SHOW_MERGE_STATUS,
+
+ CREATE_INDEX,
+ DROP_INDEX,
+ QUERY_INDEX,
+
+ LOAD_FILES,
+ REMOVE_FILE,
+ MOVE_FILE,
+
+ CREATE_TRIGGER,
+ DROP_TRIGGER,
+ START_TRIGGER,
+ STOP_TRIGGER,
+
+ CREATE_TEMPLATE,
+ SET_DEVICE_TEMPLATE,
+ SET_USING_DEVICE_TEMPLATE,
+
+ MERGE,
+ FULL_MERGE,
+
+ MNODE,
+ MEASUREMENT_MNODE,
+ STORAGE_GROUP_MNODE,
+ AUTO_CREATE_DEVICE_MNODE,
+
+ TTL,
+ KILL,
+ FLUSH,
+ TRACING,
+ CLEAR_CACHE,
+ DELETE_PARTITION,
+ LOAD_CONFIGURATION,
+ CREATE_SCHEMA_SNAPSHOT,
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/BasicFunctionOperator.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/BasicFunctionOperator.java
new file mode 100644
index 0000000..be7f156
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/BasicFunctionOperator.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.influxdb.qp.logical.crud;
+
+import org.apache.iotdb.influxdb.qp.constant.FilterConstant;
+
+public class BasicFunctionOperator extends FunctionOperator {
+ public String getValue() {
+ return value;
+ }
+
+ public void setValue(String value) {
+ this.value = value;
+ }
+
+ protected String value;
+
+ public BasicFunctionOperator(FilterConstant.FilterType filterType, String keyName, String value) {
+ super(filterType);
+ this.keyName = keyName;
+ this.value = value;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/Condition.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/Condition.java
new file mode 100644
index 0000000..2d931c3
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/Condition.java
@@ -0,0 +1,63 @@
+/*
+ * 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.influxdb.qp.logical.crud;
+
+import org.apache.iotdb.influxdb.qp.constant.FilterConstant;
+
+public class Condition {
+
+ // var of current query filter criteria
+ public String Value;
+ public FilterConstant.FilterType FilterType;
+ // actual value of current query filter condition
+ public String Literal;
+
+ public Condition(String value, FilterConstant.FilterType filterType, String literal) {
+ Value = value;
+ FilterType = filterType;
+ Literal = literal;
+ }
+
+ public Condition() {}
+
+ public void setValue(String value) {
+ Value = value;
+ }
+
+ public void setFilterType(FilterConstant.FilterType filterType) {
+ FilterType = filterType;
+ }
+
+ public void setLiteral(String literal) {
+ Literal = literal;
+ }
+
+ public String getValue() {
+ return Value;
+ }
+
+ public FilterConstant.FilterType getFilterType() {
+ return FilterType;
+ }
+
+ public String getLiteral() {
+ return Literal;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/FilterOperator.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/FilterOperator.java
new file mode 100644
index 0000000..3c1912e
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/FilterOperator.java
@@ -0,0 +1,78 @@
+/*
+ * 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.influxdb.qp.logical.crud;
+
+import org.apache.iotdb.influxdb.qp.constant.FilterConstant;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class FilterOperator implements Comparable<FilterOperator> {
+
+ public FilterConstant.FilterType getFilterType() {
+ return filterType;
+ }
+
+ public void setFilterType(FilterConstant.FilterType filterType) {
+ this.filterType = filterType;
+ }
+
+ public List<FilterOperator> getChildOperators() {
+ return childOperators;
+ }
+
+ public void setChildOperators(List<FilterOperator> childOperators) {
+ this.childOperators = childOperators;
+ }
+
+ public String getKeyName() {
+ return keyName;
+ }
+
+ public void setKeyName(String keyName) {
+ this.keyName = keyName;
+ }
+
+ protected FilterConstant.FilterType filterType;
+
+ private List<FilterOperator> childOperators = new ArrayList<>();
+
+ String keyName;
+
+ public FilterOperator() {}
+
+ public FilterOperator(FilterConstant.FilterType filterType) {
+ this.filterType = filterType;
+ }
+
+ public List<FilterOperator> getChildren() {
+ return childOperators;
+ }
+
+ public boolean addChildOperator(FilterOperator op) {
+ childOperators.add(op);
+ return true;
+ }
+
+ @Override
+ public int compareTo(FilterOperator o) {
+ return 0;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/FromComponent.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/FromComponent.java
new file mode 100644
index 0000000..2e8f812
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/FromComponent.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.influxdb.qp.logical.crud;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class FromComponent {
+
+ private List<String> nameList = new ArrayList<>();
+
+ public FromComponent() {}
+
+ public void addNodeName(String name) {
+ nameList.add(name);
+ }
+
+ public List<String> getNodeName() {
+ return nameList;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/FunctionOperator.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/FunctionOperator.java
new file mode 100644
index 0000000..fff4328
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/FunctionOperator.java
@@ -0,0 +1,29 @@
+/*
+ * 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.influxdb.qp.logical.crud;
+
+import org.apache.iotdb.influxdb.qp.constant.FilterConstant;
+
+public class FunctionOperator extends FilterOperator {
+
+ public FunctionOperator(FilterConstant.FilterType filterType) {
+ super(filterType);
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/QueryOperator.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/QueryOperator.java
new file mode 100644
index 0000000..971c629
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/QueryOperator.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.influxdb.qp.logical.crud;
+
+import org.apache.iotdb.influxdb.qp.constant.SQLConstant;
+import org.apache.iotdb.influxdb.qp.logical.Operator;
+
+public class QueryOperator extends Operator {
+
+ protected SelectComponent selectComponent;
+ protected FromComponent fromComponent;
+ protected WhereComponent whereComponent;
+
+ public QueryOperator() {
+ super(SQLConstant.TOK_QUERY);
+ operatorType = Operator.OperatorType.QUERY;
+ }
+
+ public SelectComponent getSelectComponent() {
+ return selectComponent;
+ }
+
+ public void setSelectComponent(SelectComponent selectComponent) {
+ this.selectComponent = selectComponent;
+ }
+
+ public FromComponent getFromComponent() {
+ return fromComponent;
+ }
+
+ public void setFromComponent(FromComponent fromComponent) {
+ this.fromComponent = fromComponent;
+ }
+
+ public WhereComponent getWhereComponent() {
+ return whereComponent;
+ }
+
+ public void setWhereComponent(WhereComponent whereComponent) {
+ this.whereComponent = whereComponent;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/SelectComponent.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/SelectComponent.java
new file mode 100644
index 0000000..eb88117
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/SelectComponent.java
@@ -0,0 +1,101 @@
+/*
+ * 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.influxdb.qp.logical.crud;
+
+import org.apache.iotdb.influxdb.qp.constant.SQLConstant;
+import org.apache.iotdb.influxdb.query.expression.Expression;
+import org.apache.iotdb.influxdb.query.expression.ResultColumn;
+import org.apache.iotdb.influxdb.query.expression.unary.FunctionExpression;
+import org.apache.iotdb.influxdb.query.expression.unary.NodeExpression;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** this class maintains information from select clause. */
+public final class SelectComponent {
+
+ private List<ResultColumn> resultColumns = new ArrayList<>();
+
+ private boolean hasAggregationFunction = false;
+ private boolean hasSelectorFunction = false;
+ private boolean hasMoreSelectorFunction = false;
+ private boolean hasMoreFunction = false;
+ private boolean hasFunction = false;
+ private boolean hasCommonQuery = false;
+
+ public void addResultColumn(ResultColumn resultColumn) {
+ Expression expression = resultColumn.getExpression();
+ if (expression instanceof FunctionExpression) {
+ String functionName = ((FunctionExpression) expression).getFunctionName();
+ if (SQLConstant.getNativeFunctionNames().contains(functionName.toLowerCase())) {
+ if (hasFunction) {
+ hasMoreFunction = true;
+ } else {
+ hasFunction = true;
+ }
+ }
+ if (SQLConstant.getNativeSelectorFunctionNames().contains(functionName.toLowerCase())) {
+ if (hasSelectorFunction) {
+ hasMoreSelectorFunction = true;
+ } else {
+ hasSelectorFunction = true;
+ }
+ } else {
+ hasAggregationFunction = true;
+ }
+ }
+ if (expression instanceof NodeExpression) {
+ hasCommonQuery = true;
+ }
+ resultColumns.add(resultColumn);
+ }
+
+ public void setResultColumns(List<ResultColumn> resultColumns) {
+ this.resultColumns = resultColumns;
+ }
+
+ public List<ResultColumn> getResultColumns() {
+ return resultColumns;
+ }
+
+ public boolean isHasAggregationFunction() {
+ return hasAggregationFunction;
+ }
+
+ public boolean isHasMoreFunction() {
+ return hasMoreFunction;
+ }
+
+ public boolean isHasCommonQuery() {
+ return hasCommonQuery;
+ }
+
+ public boolean isHasSelectorFunction() {
+ return hasSelectorFunction;
+ }
+
+ public boolean isHasMoreSelectorFunction() {
+ return hasMoreSelectorFunction;
+ }
+
+ public boolean isHasFunction() {
+ return hasFunction;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/WhereComponent.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/WhereComponent.java
new file mode 100644
index 0000000..b030fd6
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/crud/WhereComponent.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.influxdb.qp.logical.crud;
+
+public class WhereComponent {
+
+ private FilterOperator filterOperator;
+
+ public WhereComponent() {}
+
+ public WhereComponent(FilterOperator filterOperator) {
+ this.filterOperator = filterOperator;
+ }
+
+ public FilterOperator getFilterOperator() {
+ return filterOperator;
+ }
+
+ public void setFilterOperator(FilterOperator filterOperator) {
+ this.filterOperator = filterOperator;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/Aggregate.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/Aggregate.java
new file mode 100644
index 0000000..67a8371
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/Aggregate.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.influxdb.qp.logical.function;
+
+import org.apache.iotdb.influxdb.query.expression.Expression;
+import org.apache.iotdb.session.Session;
+
+import java.util.List;
+
+public abstract class Aggregate extends Function {
+ public Aggregate() {}
+
+ public Aggregate(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ public Aggregate(List<Expression> expressionList, Session session, String path) {
+ super(expressionList, session, path);
+ }
+
+ public abstract void updateValue(FunctionValue functionValue);
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/CountFunction.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/CountFunction.java
new file mode 100644
index 0000000..59fe505
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/CountFunction.java
@@ -0,0 +1,74 @@
+/*
+ * 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.influxdb.qp.logical.function;
+
+import org.apache.iotdb.influxdb.IoTDBInfluxDBUtils;
+import org.apache.iotdb.influxdb.query.expression.Expression;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.SessionDataSet;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+
+import java.util.List;
+
+public class CountFunction extends Aggregate {
+ public int countNum = 0;
+
+ public CountFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ public CountFunction(List<Expression> expressionList, Session session, String path) {
+ super(expressionList, session, path);
+ }
+
+ @Override
+ public void updateValue(FunctionValue functionValue) {
+ this.countNum++;
+ }
+
+ public CountFunction() {}
+
+ @Override
+ public FunctionValue calculate() {
+ return new FunctionValue(this.countNum, 0L);
+ }
+
+ @Override
+ public FunctionValue calculateByIoTDBFunc() {
+ int count = 0;
+ try {
+ SessionDataSet sessionDataSet =
+ this.session.executeQueryStatement(
+ IoTDBInfluxDBUtils.generateFunctionSql("count", getParmaName(), path));
+ while (sessionDataSet.hasNext()) {
+ RowRecord record = sessionDataSet.next();
+ List<org.apache.iotdb.tsfile.read.common.Field> fields = record.getFields();
+ if (fields.get(1).getDataType() != null) {
+ count += fields.get(1).getLongV();
+ }
+ }
+ } catch (StatementExecutionException | IoTDBConnectionException e) {
+ throw new IllegalArgumentException(e.getMessage());
+ }
+ return new FunctionValue(count, 0L);
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/FirstFunction.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/FirstFunction.java
new file mode 100644
index 0000000..fe8efe8
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/FirstFunction.java
@@ -0,0 +1,108 @@
+/*
+ * 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.influxdb.qp.logical.function;
+
+import org.apache.iotdb.influxdb.IoTDBInfluxDBUtils;
+import org.apache.iotdb.influxdb.query.expression.Expression;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.SessionDataSet;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+
+import java.util.List;
+
+public class FirstFunction extends Selector {
+ private Object value;
+
+ public FirstFunction(List<Expression> expressionList) {
+ super(expressionList);
+ this.setTimestamp(Long.MAX_VALUE);
+ }
+
+ public FirstFunction(List<Expression> expressionList, Session session, String path) {
+ super(expressionList, session, path);
+ }
+
+ @Override
+ public void updateValueAndRelate(FunctionValue functionValue, List<Object> values) {
+ Object value = functionValue.getValue();
+ Long timestamp = functionValue.getTimestamp();
+ if (timestamp <= this.getTimestamp()) {
+ this.value = value;
+ this.setTimestamp(timestamp);
+ this.setRelatedValues(values);
+ }
+ }
+
+ public FirstFunction() {}
+
+ @Override
+ public FunctionValue calculate() {
+ return new FunctionValue(value, this.getTimestamp());
+ }
+
+ @Override
+ public FunctionValue calculateByIoTDBFunc() {
+ Object firstValue = null;
+ Long firstTime = null;
+ try {
+ SessionDataSet sessionDataSet =
+ this.session.executeQueryStatement(
+ IoTDBInfluxDBUtils.generateFunctionSql("first_value", getParmaName(), path));
+ while (sessionDataSet.hasNext()) {
+ RowRecord record = sessionDataSet.next();
+ List<org.apache.iotdb.tsfile.read.common.Field> fields = record.getFields();
+ Object o = IoTDBInfluxDBUtils.iotdbFiledCvt(fields.get(1));
+ if (o != null) {
+ String newPath =
+ String.format(
+ "select %s from %s where %s.%s=%s",
+ getParmaName(),
+ fields.get(0).toString(),
+ fields.get(0).toString(),
+ getParmaName(),
+ o);
+ SessionDataSet sessionDataSetNew = this.session.executeQueryStatement(newPath);
+ while (sessionDataSetNew.hasNext()) {
+ RowRecord recordNew = sessionDataSetNew.next();
+ List<org.apache.iotdb.tsfile.read.common.Field> newFields = recordNew.getFields();
+ Long time = recordNew.getTimestamp();
+ if (firstValue == null && firstTime == null) {
+ firstValue = IoTDBInfluxDBUtils.iotdbFiledCvt(newFields.get(0));
+ firstTime = time;
+ } else {
+ if (time < firstTime) {
+ firstValue = IoTDBInfluxDBUtils.iotdbFiledCvt(newFields.get(0));
+ firstTime = time;
+ }
+ }
+ }
+ }
+ }
+ } catch (StatementExecutionException | IoTDBConnectionException e) {
+ throw new IllegalArgumentException(e.getMessage());
+ }
+ if (firstValue == null) {
+ return new FunctionValue(null, null);
+ }
+ return new FunctionValue(firstValue, firstTime);
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/Function.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/Function.java
new file mode 100644
index 0000000..0135887
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/Function.java
@@ -0,0 +1,69 @@
+/*
+ * 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.influxdb.qp.logical.function;
+
+import org.apache.iotdb.influxdb.query.expression.Expression;
+import org.apache.iotdb.influxdb.query.expression.unary.NodeExpression;
+import org.apache.iotdb.session.Session;
+
+import java.util.List;
+
+public abstract class Function {
+
+ // contain possible parameters
+ private List<Expression> expressionList;
+
+ protected Session session;
+
+ protected String path;
+
+ public Function(List<Expression> expressionList) {
+ this.expressionList = expressionList;
+ }
+
+ public Function(List<Expression> expressionList, Session session, String path) {
+ this.expressionList = expressionList;
+ this.session = session;
+ this.path = path;
+ }
+
+ public Function() {}
+
+ // calculate result
+ public abstract FunctionValue calculate();
+
+ public abstract FunctionValue calculateByIoTDBFunc();
+
+ public List<Expression> getExpressions() {
+ return this.expressionList;
+ }
+
+ public Session getSession() {
+ return this.session;
+ }
+
+ public String getParmaName() {
+ if (expressionList == null) {
+ throw new IllegalArgumentException("not support param");
+ }
+ NodeExpression parmaExpression = (NodeExpression) expressionList.get(0);
+ return parmaExpression.getName();
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/FunctionFactory.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/FunctionFactory.java
new file mode 100644
index 0000000..d2487b3
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/FunctionFactory.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.influxdb.qp.logical.function;
+
+import org.apache.iotdb.influxdb.qp.constant.SQLConstant;
+import org.apache.iotdb.influxdb.query.expression.Expression;
+import org.apache.iotdb.session.Session;
+
+import java.util.List;
+
+public class FunctionFactory {
+ public static Function generateFunction(String name, List<Expression> expressionList) {
+ switch (name) {
+ case SQLConstant.MAX:
+ return new MaxFunction(expressionList);
+ case SQLConstant.MIN:
+ return new MinFunction(expressionList);
+ case SQLConstant.MEAN:
+ return new MeanFunction(expressionList);
+ case SQLConstant.LAST:
+ return new LastFunction(expressionList);
+ case SQLConstant.FIRST:
+ return new FirstFunction(expressionList);
+ case SQLConstant.COUNT:
+ return new CountFunction(expressionList);
+ case SQLConstant.MEDIAN:
+ return new MedianFunction(expressionList);
+ case SQLConstant.MODE:
+ return new ModeFunction(expressionList);
+ case SQLConstant.SPREAD:
+ return new SpreadFunction(expressionList);
+ case SQLConstant.STDDEV:
+ return new StddevFunction(expressionList);
+ case SQLConstant.SUM:
+ return new SumFunction(expressionList);
+ default:
+ throw new IllegalArgumentException("not support aggregation name:" + name);
+ }
+ }
+
+ public static Function generateFunctionBySession(
+ String name, List<Expression> expressionList, Session session, String path) {
+ switch (name) {
+ case SQLConstant.MAX:
+ return new MaxFunction(expressionList, session, path);
+ case SQLConstant.MIN:
+ return new MinFunction(expressionList, session, path);
+ case SQLConstant.MEAN:
+ return new MeanFunction(expressionList, session, path);
+ case SQLConstant.LAST:
+ return new LastFunction(expressionList, session, path);
+ case SQLConstant.FIRST:
+ return new FirstFunction(expressionList, session, path);
+ case SQLConstant.COUNT:
+ return new CountFunction(expressionList, session, path);
+ case SQLConstant.SUM:
+ return new SumFunction(expressionList, session, path);
+ case SQLConstant.SPREAD:
+ return new SpreadFunction(expressionList, session, path);
+ default:
+ throw new IllegalArgumentException("not support aggregation name:" + name);
+ }
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/FunctionValue.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/FunctionValue.java
new file mode 100644
index 0000000..71020f1
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/FunctionValue.java
@@ -0,0 +1,46 @@
+/*
+ * 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.influxdb.qp.logical.function;
+
+public class FunctionValue {
+ private Object value;
+ private Long timestamp;
+
+ public FunctionValue(Object value, Long timestamp) {
+ this.value = value;
+ this.timestamp = timestamp;
+ }
+
+ public Object getValue() {
+ return value;
+ }
+
+ public void setValue(Object value) {
+ this.value = value;
+ }
+
+ public Long getTimestamp() {
+ return timestamp;
+ }
+
+ public void setTimestamp(Long timestamp) {
+ this.timestamp = timestamp;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/LastFunction.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/LastFunction.java
new file mode 100644
index 0000000..aab51a6
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/LastFunction.java
@@ -0,0 +1,105 @@
+/*
+ * 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.influxdb.qp.logical.function;
+
+import org.apache.iotdb.influxdb.IoTDBInfluxDBUtils;
+import org.apache.iotdb.influxdb.query.expression.Expression;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.SessionDataSet;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+
+import java.util.List;
+
+public class LastFunction extends Selector {
+ private Object value;
+
+ public LastFunction(List<Expression> expressionList) {
+ super(expressionList);
+ this.setTimestamp(Long.MIN_VALUE);
+ }
+
+ public LastFunction(List<Expression> expressionList, Session session, String path) {
+ super(expressionList, session, path);
+ }
+
+ public LastFunction() {}
+
+ @Override
+ public void updateValueAndRelate(FunctionValue functionValue, List<Object> values) {
+
+ Object value = functionValue.getValue();
+ Long timestamp = functionValue.getTimestamp();
+ if (timestamp >= this.getTimestamp()) {
+ this.value = value;
+ this.setTimestamp(timestamp);
+ this.setRelatedValues(values);
+ }
+ }
+
+ @Override
+ public FunctionValue calculate() {
+ return new FunctionValue(value, this.getTimestamp());
+ }
+
+ @Override
+ public FunctionValue calculateByIoTDBFunc() {
+ Object LastValue = null;
+ Long LastTime = null;
+ try {
+ SessionDataSet sessionDataSet =
+ this.session.executeQueryStatement(
+ IoTDBInfluxDBUtils.generateFunctionSql("last_value", getParmaName(), path));
+ while (sessionDataSet.hasNext()) {
+ RowRecord record = sessionDataSet.next();
+ List<org.apache.iotdb.tsfile.read.common.Field> fields = record.getFields();
+ Object o = IoTDBInfluxDBUtils.iotdbFiledCvt(fields.get(1));
+ if (o != null) {
+ String newPath =
+ String.format(
+ "select %s from %s where %s.%s=%s",
+ getParmaName(), fields.get(0), fields.get(0), getParmaName(), o);
+ SessionDataSet sessionDataSetNew = this.session.executeQueryStatement(newPath);
+ while (sessionDataSetNew.hasNext()) {
+ RowRecord recordNew = sessionDataSetNew.next();
+ List<org.apache.iotdb.tsfile.read.common.Field> newFields = recordNew.getFields();
+ Long time = recordNew.getTimestamp();
+ if (LastValue == null && LastTime == null) {
+ LastValue = IoTDBInfluxDBUtils.iotdbFiledCvt(newFields.get(0));
+ LastTime = time;
+ } else {
+ if (time > LastTime) {
+ LastValue = IoTDBInfluxDBUtils.iotdbFiledCvt(newFields.get(0));
+ LastTime = time;
+ }
+ }
+ }
+ }
+ }
+ } catch (StatementExecutionException | IoTDBConnectionException e) {
+ throw new IllegalArgumentException(e.getMessage());
+ }
+ if (LastTime == null || LastValue == null) {
+ return new FunctionValue(null, null);
+ }
+ return new FunctionValue(LastValue, LastTime);
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/MaxFunction.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/MaxFunction.java
new file mode 100644
index 0000000..c7419ed
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/MaxFunction.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.influxdb.qp.logical.function;
+
+import org.apache.iotdb.influxdb.IoTDBInfluxDBUtils;
+import org.apache.iotdb.influxdb.query.expression.Expression;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.SessionDataSet;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+
+import java.util.List;
+
+public class MaxFunction extends Selector {
+ private Double doubleValue = Double.MIN_VALUE;
+ private String stringValue = null;
+ private boolean isNumber = false;
+ private boolean isString = false;
+
+ public MaxFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ public MaxFunction(List<Expression> expressionList, Session session, String path) {
+ super(expressionList, session, path);
+ }
+
+ @Override
+ public void updateValueAndRelate(FunctionValue functionValue, List<Object> values) {
+ Object value = functionValue.getValue();
+ Long timestamp = functionValue.getTimestamp();
+ if (value instanceof Number) {
+ if (!isNumber) {
+ isNumber = true;
+ }
+ double tmpValue = ((Number) value).doubleValue();
+ if (tmpValue >= this.doubleValue) {
+ doubleValue = tmpValue;
+ this.setTimestamp(timestamp);
+ this.setRelatedValues(values);
+ }
+ } else if (value instanceof String) {
+ String tmpValue = (String) value;
+ if (!isString) {
+ isString = true;
+ stringValue = tmpValue;
+ this.setTimestamp(timestamp);
+ this.setRelatedValues(values);
+ } else {
+ if (tmpValue.compareTo(this.stringValue) >= 0) {
+ stringValue = tmpValue;
+ this.setTimestamp(timestamp);
+ this.setRelatedValues(values);
+ }
+ }
+ }
+ }
+
+ @Override
+ public FunctionValue calculate() {
+ if (!isString && !isNumber) {
+ return new FunctionValue(null, null);
+ } else if (isString) {
+ return new FunctionValue(stringValue, this.getTimestamp());
+ } else {
+ return new FunctionValue(doubleValue, this.getTimestamp());
+ }
+ }
+
+ @Override
+ public FunctionValue calculateByIoTDBFunc() {
+ Double maxNumber = null;
+ try {
+ SessionDataSet sessionDataSet =
+ this.session.executeQueryStatement(
+ IoTDBInfluxDBUtils.generateFunctionSql("max_value", getParmaName(), path));
+ while (sessionDataSet.hasNext()) {
+ RowRecord record = sessionDataSet.next();
+ List<org.apache.iotdb.tsfile.read.common.Field> fields = record.getFields();
+ Object o = IoTDBInfluxDBUtils.iotdbFiledCvt(fields.get(1));
+ if ((o instanceof Number)) {
+ double tmpValue = ((Number) o).doubleValue();
+ if (maxNumber == null) {
+ maxNumber = tmpValue;
+ } else if (tmpValue > maxNumber) {
+ maxNumber = tmpValue;
+ }
+ }
+ }
+ } catch (StatementExecutionException | IoTDBConnectionException e) {
+ throw new IllegalArgumentException(e.getMessage());
+ }
+ return new FunctionValue(maxNumber, maxNumber == null ? null : 0L);
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/MeanFunction.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/MeanFunction.java
new file mode 100644
index 0000000..d31e390
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/MeanFunction.java
@@ -0,0 +1,95 @@
+/*
+ * 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.influxdb.qp.logical.function;
+
+import org.apache.iotdb.influxdb.IoTDBInfluxDBUtils;
+import org.apache.iotdb.influxdb.qp.utils.MathUtil;
+import org.apache.iotdb.influxdb.qp.utils.TypeUtil;
+import org.apache.iotdb.influxdb.query.expression.Expression;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.SessionDataSet;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class MeanFunction extends Aggregate {
+ public List<Double> numbers = new ArrayList<>();
+
+ public MeanFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ public MeanFunction(List<Expression> expressionList, Session session, String path) {
+ super(expressionList, session, path);
+ }
+
+ public MeanFunction() {}
+
+ @Override
+ public void updateValue(FunctionValue functionValue) {
+ Object value = functionValue.getValue();
+ if (TypeUtil.checkDecimal(value)) {
+ numbers.add(((Number) value).doubleValue());
+ } else {
+ throw new IllegalArgumentException("mean not valid type");
+ }
+ }
+
+ @Override
+ public FunctionValue calculate() {
+ return new FunctionValue(numbers.size() == 0 ? numbers : MathUtil.Mean(numbers), 0L);
+ }
+
+ @Override
+ public FunctionValue calculateByIoTDBFunc() {
+
+ int sum = 0;
+ int count = 0;
+ try {
+ SessionDataSet sessionDataSet =
+ this.session.executeQueryStatement(
+ IoTDBInfluxDBUtils.generateFunctionSql("count", getParmaName(), path));
+ while (sessionDataSet.hasNext()) {
+ RowRecord record = sessionDataSet.next();
+ List<org.apache.iotdb.tsfile.read.common.Field> fields = record.getFields();
+ if (fields.get(1).getDataType() != null) {
+ count += fields.get(1).getLongV();
+ }
+ }
+
+ sessionDataSet =
+ this.session.executeQueryStatement(
+ IoTDBInfluxDBUtils.generateFunctionSql("sum", getParmaName(), path));
+ while (sessionDataSet.hasNext()) {
+ RowRecord record = sessionDataSet.next();
+ List<org.apache.iotdb.tsfile.read.common.Field> fields = record.getFields();
+ if (fields.get(1).getDataType() != null) {
+ sum += fields.get(1).getDoubleV();
+ }
+ }
+ } catch (StatementExecutionException | IoTDBConnectionException e) {
+ throw new IllegalArgumentException(e.getMessage());
+ }
+ return new FunctionValue(count != 0 ? sum / count : null, count != 0 ? 0L : null);
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/MedianFunction.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/MedianFunction.java
new file mode 100644
index 0000000..d0c1c95
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/MedianFunction.java
@@ -0,0 +1,63 @@
+/*
+ * 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.influxdb.qp.logical.function;
+
+import org.apache.iotdb.influxdb.query.expression.Expression;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+public class MedianFunction extends Aggregate {
+ private List<Double> numbers = new ArrayList<>();
+
+ public MedianFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ @Override
+ public void updateValue(FunctionValue functionValue) {
+ Object value = functionValue.getValue();
+ if (value instanceof Number) {
+ numbers.add(((Number) value).doubleValue());
+ } else {
+ throw new IllegalArgumentException("not support this type");
+ }
+ }
+
+ public MedianFunction() {}
+
+ @Override
+ public FunctionValue calculate() {
+ Collections.sort(numbers);
+ int len = numbers.size();
+ if (len % 2 == 0) {
+ return new FunctionValue((numbers.get(len / 2) + numbers.get(len / 2 - 1)) / 2, 0L);
+
+ } else {
+ return new FunctionValue(numbers.get(len / 2), 0L);
+ }
+ }
+
+ @Override
+ public FunctionValue calculateByIoTDBFunc() {
+ return null;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/MinFunction.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/MinFunction.java
new file mode 100644
index 0000000..75205ef
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/MinFunction.java
@@ -0,0 +1,116 @@
+/*
+ * 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.influxdb.qp.logical.function;
+
+import org.apache.iotdb.influxdb.IoTDBInfluxDBUtils;
+import org.apache.iotdb.influxdb.query.expression.Expression;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.SessionDataSet;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+
+import java.util.List;
+
+public class MinFunction extends Selector {
+ private Double doubleValue = Double.MAX_VALUE;
+ private String stringValue = null;
+ private boolean isNumber = false;
+ private boolean isString = false;
+
+ public MinFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ public MinFunction(List<Expression> expressionList, Session session, String path) {
+ super(expressionList, session, path);
+ }
+
+ @Override
+ public void updateValueAndRelate(FunctionValue functionValue, List<Object> values) {
+ Object value = functionValue.getValue();
+ Long timestamp = functionValue.getTimestamp();
+ if (value instanceof Number) {
+ if (!isNumber) {
+ isNumber = true;
+ }
+ double tmpValue = ((Number) value).doubleValue();
+ if (tmpValue <= this.doubleValue) {
+ doubleValue = tmpValue;
+ this.setTimestamp(timestamp);
+ this.setRelatedValues(values);
+ }
+ } else if (value instanceof String) {
+ String tmpValue = (String) value;
+ if (!isString) {
+ isString = true;
+ stringValue = tmpValue;
+ this.setTimestamp(timestamp);
+ this.setRelatedValues(values);
+ } else {
+ if (tmpValue.compareTo(this.stringValue) <= 0) {
+ stringValue = tmpValue;
+ this.setTimestamp(timestamp);
+ this.setRelatedValues(values);
+ }
+ }
+ }
+ }
+
+ public MinFunction() {}
+
+ @Override
+ public FunctionValue calculate() {
+ if (!isString && !isNumber) {
+ return new FunctionValue(null, null);
+ } else if (isString) {
+ return new FunctionValue(stringValue, this.getTimestamp());
+ } else {
+ return new FunctionValue(doubleValue, this.getTimestamp());
+ }
+ }
+
+ @Override
+ public FunctionValue calculateByIoTDBFunc() {
+
+ Double minNumber = null;
+ try {
+ SessionDataSet sessionDataSet =
+ this.session.executeQueryStatement(
+ IoTDBInfluxDBUtils.generateFunctionSql("max_value", getParmaName(), path));
+ while (sessionDataSet.hasNext()) {
+ RowRecord record = sessionDataSet.next();
+ List<org.apache.iotdb.tsfile.read.common.Field> fields = record.getFields();
+ Object o = IoTDBInfluxDBUtils.iotdbFiledCvt(fields.get(1));
+ if ((o instanceof Number)) {
+ double tmpValue = ((Number) o).doubleValue();
+ if (minNumber == null) {
+ minNumber = tmpValue;
+ } else if (tmpValue < minNumber) {
+ minNumber = tmpValue;
+ }
+ }
+ }
+ } catch (StatementExecutionException | IoTDBConnectionException e) {
+ throw new IllegalArgumentException(e.getMessage());
+ }
+ return new FunctionValue(minNumber, minNumber == null ? null : 0L);
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/ModeFunction.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/ModeFunction.java
new file mode 100644
index 0000000..527b538
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/ModeFunction.java
@@ -0,0 +1,78 @@
+/*
+ * 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.influxdb.qp.logical.function;
+
+import org.apache.iotdb.influxdb.query.expression.Expression;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class ModeFunction extends Aggregate {
+ private Map<Object, Integer> valueOrders = new HashMap<>();
+ private Map<Object, Long> valueLastTimestamp = new HashMap<>();
+ private int maxNumber = 0;
+ private Object maxObject = null;
+
+ public ModeFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ public ModeFunction() {}
+
+ @Override
+ public void updateValue(FunctionValue functionValue) {
+ Object value = functionValue.getValue();
+ Long timestamp = functionValue.getTimestamp();
+ // update new data
+ if (!valueOrders.containsKey(value)) {
+ valueOrders.put(value, 1);
+ valueLastTimestamp.put(value, timestamp);
+ } else {
+ valueOrders.put(value, valueOrders.get(value) + 1);
+ if (timestamp < valueLastTimestamp.get(value)) {
+ valueLastTimestamp.put(value, timestamp);
+ }
+ }
+ // Judge whether the new data meets the conditions
+ if (maxObject == null) {
+ maxObject = value;
+ maxNumber = 1;
+ } else {
+ if (valueOrders.get(value) > maxNumber) {
+ maxNumber = valueOrders.get(value);
+ maxObject = value;
+ } else if (valueOrders.get(value) == maxNumber
+ && timestamp < valueLastTimestamp.get(maxObject)) {
+ maxObject = value;
+ }
+ }
+ }
+
+ @Override
+ public FunctionValue calculate() {
+ return new FunctionValue(maxObject, 0L);
+ }
+
+ @Override
+ public FunctionValue calculateByIoTDBFunc() {
+ return null;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/Selector.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/Selector.java
new file mode 100644
index 0000000..d1e76a9
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/Selector.java
@@ -0,0 +1,61 @@
+/*
+ * 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.influxdb.qp.logical.function;
+
+import org.apache.iotdb.influxdb.query.expression.Expression;
+import org.apache.iotdb.session.Session;
+
+import java.util.List;
+
+public abstract class Selector extends Function {
+
+ // The timestamp corresponding to the value
+ private Long timestamp;
+
+ private List<Object> relatedValues;
+
+ public Selector() {}
+
+ public Selector(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ public Selector(List<Expression> expressionList, Session session, String path) {
+ super(expressionList, session, path);
+ }
+
+ public abstract void updateValueAndRelate(FunctionValue functionValue, List<Object> values);
+
+ public List<Object> getRelatedValues() {
+ return this.relatedValues;
+ }
+
+ public Long getTimestamp() {
+ return timestamp;
+ }
+
+ public void setTimestamp(Long timestamp) {
+ this.timestamp = timestamp;
+ }
+
+ public void setRelatedValues(List<Object> relatedValues) {
+ this.relatedValues = relatedValues;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/SpreadFunction.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/SpreadFunction.java
new file mode 100644
index 0000000..4a2ee8d
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/SpreadFunction.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.influxdb.qp.logical.function;
+
+import org.apache.iotdb.influxdb.IoTDBInfluxDBUtils;
+import org.apache.iotdb.influxdb.query.expression.Expression;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.SessionDataSet;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+
+import java.util.List;
+
+public class SpreadFunction extends Aggregate {
+ private Double maxNum = null;
+ private Double minNum = null;
+
+ public SpreadFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ public SpreadFunction(List<Expression> expressionList, Session session, String path) {
+ super(expressionList, session, path);
+ }
+
+ @Override
+ public void updateValue(FunctionValue functionValue) {
+ Object value = functionValue.getValue();
+ if (!(value instanceof Number)) {
+ throw new IllegalArgumentException("not support this type");
+ }
+
+ double tmpValue = ((Number) value).doubleValue();
+ if (maxNum == null || tmpValue > maxNum) {
+ maxNum = tmpValue;
+ }
+ if (minNum == null || tmpValue < minNum) {
+ minNum = tmpValue;
+ }
+ }
+
+ @Override
+ public FunctionValue calculate() {
+ return new FunctionValue(maxNum - minNum, 0L);
+ }
+
+ @Override
+ public FunctionValue calculateByIoTDBFunc() {
+ Double maxNumber = null;
+ Double minNumber = null;
+ try {
+ SessionDataSet sessionDataSet =
+ this.session.executeQueryStatement(
+ IoTDBInfluxDBUtils.generateFunctionSql("max_value", getParmaName(), path));
+ while (sessionDataSet.hasNext()) {
+ RowRecord record = sessionDataSet.next();
+ List<org.apache.iotdb.tsfile.read.common.Field> fields = record.getFields();
+ Object o = IoTDBInfluxDBUtils.iotdbFiledCvt(fields.get(1));
+ if ((o instanceof Number)) {
+ double tmpValue = ((Number) o).doubleValue();
+ if (maxNumber == null) {
+ maxNumber = tmpValue;
+ } else if (tmpValue > maxNumber) {
+ maxNumber = tmpValue;
+ }
+ }
+ }
+ sessionDataSet =
+ this.session.executeQueryStatement(
+ IoTDBInfluxDBUtils.generateFunctionSql("min_value", getParmaName(), path));
+ while (sessionDataSet.hasNext()) {
+ RowRecord record = sessionDataSet.next();
+ List<org.apache.iotdb.tsfile.read.common.Field> fields = record.getFields();
+ Object o = IoTDBInfluxDBUtils.iotdbFiledCvt(fields.get(1));
+ if ((o instanceof Number)) {
+ double tmpValue = ((Number) o).doubleValue();
+ if (minNumber == null) {
+ minNumber = tmpValue;
+ } else if (tmpValue < minNumber) {
+ minNumber = tmpValue;
+ }
+ }
+ }
+ } catch (StatementExecutionException | IoTDBConnectionException e) {
+ throw new IllegalArgumentException(e.getMessage());
+ }
+ if (maxNumber == null || minNumber == null) {
+ return new FunctionValue(null, null);
+ }
+ return new FunctionValue(maxNumber - minNumber, 0L);
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/StddevFunction.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/StddevFunction.java
new file mode 100644
index 0000000..ff3f651
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/StddevFunction.java
@@ -0,0 +1,57 @@
+/*
+ * 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.influxdb.qp.logical.function;
+
+import org.apache.iotdb.influxdb.qp.utils.MathUtil;
+import org.apache.iotdb.influxdb.query.expression.Expression;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class StddevFunction extends Aggregate {
+ private List<Double> numbers = new ArrayList<>();
+
+ public StddevFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ public StddevFunction() {}
+
+ @Override
+ public void updateValue(FunctionValue functionValue) {
+ Object value = functionValue.getValue();
+ if (!(value instanceof Number)) {
+ throw new IllegalArgumentException("not support this type");
+ }
+
+ double tmpValue = ((Number) value).doubleValue();
+ numbers.add(tmpValue);
+ }
+
+ @Override
+ public FunctionValue calculate() {
+ return new FunctionValue(numbers.size() == 0 ? numbers : MathUtil.POP_STD_dev(numbers), 0L);
+ }
+
+ @Override
+ public FunctionValue calculateByIoTDBFunc() {
+ return null;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/SumFunction.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/SumFunction.java
new file mode 100644
index 0000000..fa60f4b
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/logical/function/SumFunction.java
@@ -0,0 +1,83 @@
+/*
+ * 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.influxdb.qp.logical.function;
+
+import org.apache.iotdb.influxdb.IoTDBInfluxDBUtils;
+import org.apache.iotdb.influxdb.qp.utils.MathUtil;
+import org.apache.iotdb.influxdb.query.expression.Expression;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.SessionDataSet;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class SumFunction extends Aggregate {
+
+ private List<Double> numbers = new ArrayList<>();
+
+ public SumFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ public SumFunction(List<Expression> expressionList, Session session, String path) {
+ super(expressionList, session, path);
+ }
+
+ public SumFunction() {}
+
+ @Override
+ public void updateValue(FunctionValue functionValue) {
+ Object value = functionValue.getValue();
+ if (!(value instanceof Number)) {
+ throw new IllegalArgumentException("not support this type");
+ }
+
+ double tmpValue = ((Number) value).doubleValue();
+ numbers.add(tmpValue);
+ }
+
+ @Override
+ public FunctionValue calculate() {
+ return new FunctionValue(numbers.size() == 0 ? numbers : MathUtil.Sum(numbers), 0L);
+ }
+
+ @Override
+ public FunctionValue calculateByIoTDBFunc() {
+ int sum = 0;
+ try {
+ SessionDataSet sessionDataSet =
+ this.session.executeQueryStatement(
+ IoTDBInfluxDBUtils.generateFunctionSql("sum", getParmaName(), path));
+ while (sessionDataSet.hasNext()) {
+ RowRecord record = sessionDataSet.next();
+ List<org.apache.iotdb.tsfile.read.common.Field> fields = record.getFields();
+ if (fields.get(1).getDataType() != null) {
+ sum += fields.get(1).getDoubleV();
+ }
+ }
+ } catch (StatementExecutionException | IoTDBConnectionException e) {
+ throw new IllegalArgumentException(e.getMessage());
+ }
+ return new FunctionValue(sum, 0L);
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/sql/InfluxDBSqlVisitor.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/sql/InfluxDBSqlVisitor.java
new file mode 100644
index 0000000..5924bb9
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/sql/InfluxDBSqlVisitor.java
@@ -0,0 +1,289 @@
+/*
+ * 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.influxdb.qp.sql;
+
+import org.apache.iotdb.influxdb.qp.constant.FilterConstant;
+import org.apache.iotdb.influxdb.qp.constant.SQLConstant;
+import org.apache.iotdb.influxdb.qp.logical.Operator;
+import org.apache.iotdb.influxdb.qp.logical.crud.*;
+import org.apache.iotdb.influxdb.qp.logical.crud.QueryOperator;
+import org.apache.iotdb.influxdb.qp.utils.DatetimeUtils;
+import org.apache.iotdb.influxdb.query.expression.Expression;
+import org.apache.iotdb.influxdb.query.expression.ResultColumn;
+import org.apache.iotdb.influxdb.query.expression.binary.*;
+import org.apache.iotdb.influxdb.query.expression.unary.FunctionExpression;
+import org.apache.iotdb.influxdb.query.expression.unary.NegationExpression;
+import org.apache.iotdb.influxdb.query.expression.unary.NodeExpression;
+
+public class InfluxDBSqlVisitor extends InfluxDBBaseVisitor<Operator> {
+
+ private QueryOperator queryOp;
+
+ @Override
+ public Operator visitSingleStatement(InfluxDBParser.SingleStatementContext ctx) {
+ return visit(ctx.statement());
+ }
+
+ @Override
+ public Operator visitSelectStatement(InfluxDBParser.SelectStatementContext ctx) {
+ queryOp = new QueryOperator();
+ parseSelectClause(ctx.selectClause());
+ parseFromClause(ctx.fromClause());
+ if (ctx.whereClause() != null) {
+ WhereComponent whereComponent = parseWhereClause(ctx.whereClause());
+ queryOp.setWhereComponent(whereComponent);
+ }
+ return queryOp;
+ }
+
+ public void parseSelectClause(InfluxDBParser.SelectClauseContext ctx) {
+ SelectComponent selectComponent = new SelectComponent();
+ for (InfluxDBParser.ResultColumnContext resultColumnContext : ctx.resultColumn()) {
+ selectComponent.addResultColumn(parseResultColumn(resultColumnContext));
+ }
+ queryOp.setSelectComponent(selectComponent);
+ }
+
+ private void parseFromClause(InfluxDBParser.FromClauseContext fromClause) {
+ FromComponent fromComponent = new FromComponent();
+
+ for (InfluxDBParser.NodeNameContext nodeName : fromClause.nodeName()) {
+ fromComponent.addNodeName(nodeName.getText());
+ }
+ queryOp.setFromComponent(fromComponent);
+ }
+
+ private WhereComponent parseWhereClause(InfluxDBParser.WhereClauseContext ctx) {
+ FilterOperator whereOp = new FilterOperator();
+ whereOp.addChildOperator(parseOrExpression(ctx.orExpression()));
+ return new WhereComponent(whereOp.getChildren().get(0));
+ }
+
+ private FilterOperator parseOrExpression(InfluxDBParser.OrExpressionContext ctx) {
+ if (ctx.andExpression().size() == 1) {
+ return parseAndExpression(ctx.andExpression(0));
+ }
+ FilterOperator binaryOp = new FilterOperator(FilterConstant.FilterType.KW_OR);
+ if (ctx.andExpression().size() > 2) {
+ binaryOp.addChildOperator(parseAndExpression(ctx.andExpression(0)));
+ binaryOp.addChildOperator(parseAndExpression(ctx.andExpression(1)));
+ for (int i = 2; i < ctx.andExpression().size(); i++) {
+ FilterOperator operator = new FilterOperator(FilterConstant.FilterType.KW_OR);
+ operator.addChildOperator(binaryOp);
+ operator.addChildOperator(parseAndExpression(ctx.andExpression(i)));
+ binaryOp = operator;
+ }
+ } else {
+ for (InfluxDBParser.AndExpressionContext andExpressionContext : ctx.andExpression()) {
+ binaryOp.addChildOperator(parseAndExpression(andExpressionContext));
+ }
+ }
+ return binaryOp;
+ }
+
+ private FilterOperator parseAndExpression(InfluxDBParser.AndExpressionContext ctx) {
+ if (ctx.predicate().size() == 1) {
+ return parsePredicate(ctx.predicate(0));
+ }
+ FilterOperator binaryOp = new FilterOperator(FilterConstant.FilterType.KW_AND);
+ int size = ctx.predicate().size();
+ if (size > 2) {
+ binaryOp.addChildOperator(parsePredicate(ctx.predicate(0)));
+ binaryOp.addChildOperator(parsePredicate(ctx.predicate(1)));
+ for (int i = 2; i < size; i++) {
+ FilterOperator op = new FilterOperator(FilterConstant.FilterType.KW_AND);
+ op.addChildOperator(binaryOp);
+ op.addChildOperator(parsePredicate(ctx.predicate(i)));
+ binaryOp = op;
+ }
+ } else {
+ for (InfluxDBParser.PredicateContext predicateContext : ctx.predicate()) {
+ binaryOp.addChildOperator(parsePredicate(predicateContext));
+ }
+ }
+ return binaryOp;
+ }
+
+ private FilterOperator parsePredicate(InfluxDBParser.PredicateContext ctx) {
+ if (ctx.OPERATOR_NOT() != null) {
+ FilterOperator notOp = new FilterOperator(FilterConstant.FilterType.KW_NOT);
+ notOp.addChildOperator(parseOrExpression(ctx.orExpression()));
+ return notOp;
+ } else if (ctx.LR_BRACKET() != null && ctx.OPERATOR_NOT() == null) {
+ return parseOrExpression(ctx.orExpression());
+ } else {
+ String keyName = null;
+ if (ctx.TIME() != null || ctx.TIMESTAMP() != null) {
+ keyName = SQLConstant.RESERVED_TIME;
+ }
+ if (ctx.nodeName() != null) {
+ keyName = ctx.nodeName().getText();
+ }
+ if (keyName == null) {
+ throw new IllegalArgumentException("keyName is null, please check the sql");
+ }
+ return parseBasicFunctionOperator(ctx, keyName);
+ }
+ }
+
+ private FilterOperator parseBasicFunctionOperator(
+ InfluxDBParser.PredicateContext ctx, String keyName) {
+ BasicFunctionOperator basic;
+ if (ctx.constant().dateExpression() != null) {
+ if (!keyName.equals(SQLConstant.RESERVED_TIME)) {
+ throw new IllegalArgumentException("Date can only be used to time");
+ }
+ basic =
+ new BasicFunctionOperator(
+ FilterConstant.lexerToFilterType.get(ctx.comparisonOperator().type.getType()),
+ keyName,
+ Long.toString(parseDateExpression(ctx.constant().dateExpression())));
+ } else {
+ basic =
+ new BasicFunctionOperator(
+ FilterConstant.lexerToFilterType.get(ctx.comparisonOperator().type.getType()),
+ keyName,
+ ctx.constant().getText());
+ }
+ return basic;
+ }
+
+ /**
+ * parse time expression, which is addition and subtraction expression of duration time, now() or
+ * DataTimeFormat time.
+ *
+ * <p>eg. now() + 1d - 2h
+ */
+ private Long parseDateExpression(InfluxDBParser.DateExpressionContext ctx) {
+ long time;
+ time = parseTimeFormat(ctx.getChild(0).getText());
+ for (int i = 1; i < ctx.getChildCount(); i = i + 2) {
+ if (ctx.getChild(i).getText().equals("+")) {
+ time += DatetimeUtils.convertDurationStrToLong(time, ctx.getChild(i + 1).getText());
+ } else {
+ time -= DatetimeUtils.convertDurationStrToLong(time, ctx.getChild(i + 1).getText());
+ }
+ }
+ return time;
+ }
+
+ /** function for parsing time format. */
+ public long parseTimeFormat(String timestampStr) {
+ if (timestampStr == null || timestampStr.trim().equals("")) {
+ throw new IllegalArgumentException("input timestamp cannot be empty");
+ }
+ long startupNano = System.nanoTime();
+ if (timestampStr.equalsIgnoreCase(SQLConstant.NOW_FUNC)) {
+ String timePrecision = DatetimeUtils.timestampPrecision;
+ switch (timePrecision) {
+ case "ns":
+ return System.currentTimeMillis() * 1000_000
+ + (System.nanoTime() - startupNano) % 1000_000;
+ case "us":
+ return System.currentTimeMillis() * 1000
+ + (System.nanoTime() - startupNano) / 1000 % 1000;
+ default:
+ return System.currentTimeMillis();
+ }
+ }
+ throw new IllegalArgumentException(
+ String.format(
+ "Input time format %s error. "
+ + "Input like yyyy-MM-dd HH:mm:ss, yyyy-MM-ddTHH:mm:ss or "
+ + "refer to user document for more info.",
+ timestampStr));
+ }
+
+ private ResultColumn parseResultColumn(InfluxDBParser.ResultColumnContext resultColumnContext) {
+ return new ResultColumn(
+ parseExpression(resultColumnContext.expression()),
+ resultColumnContext.AS() == null ? null : resultColumnContext.ID().getText());
+ }
+
+ private Expression parseExpression(InfluxDBParser.ExpressionContext context) {
+ // unary
+ if (context.functionName != null) {
+ return parseFunctionExpression(context);
+ }
+ if (context.nodeName() != null) {
+ return new NodeExpression(context.nodeName().getText());
+ }
+
+ if (context.literal != null) {
+ return new NodeExpression(context.literal.getText());
+ }
+ if (context.unary != null) {
+ return context.MINUS() != null
+ ? new NegationExpression(parseExpression(context.expression(0)))
+ : parseExpression(context.expression(0));
+ }
+
+ // binary
+ Expression leftExpression = parseExpression(context.leftExpression);
+ Expression rightExpression = parseExpression(context.rightExpression);
+ if (context.STAR() != null) {
+ return new MultiplicationExpression(leftExpression, rightExpression);
+ }
+ if (context.DIV() != null) {
+ return new DivisionExpression(leftExpression, rightExpression);
+ }
+ if (context.MOD() != null) {
+ return new ModuloExpression(leftExpression, rightExpression);
+ }
+ if (context.PLUS() != null) {
+ return new AdditionExpression(leftExpression, rightExpression);
+ }
+ if (context.MINUS() != null) {
+ return new SubtractionExpression(leftExpression, rightExpression);
+ }
+ throw new UnsupportedOperationException();
+ }
+
+ private Expression parseFunctionExpression(InfluxDBParser.ExpressionContext functionClause) {
+
+ FunctionExpression functionExpression =
+ new FunctionExpression(functionClause.functionName.getText());
+
+ // expressions
+ for (InfluxDBParser.ExpressionContext expression : functionClause.expression()) {
+ functionExpression.addExpression(parseExpression(expression));
+ }
+
+ // attributes
+ for (InfluxDBParser.FunctionAttributeContext functionAttribute :
+ functionClause.functionAttribute()) {
+ functionExpression.addAttribute(
+ removeStringQuote(functionAttribute.functionAttributeKey.getText()),
+ removeStringQuote(functionAttribute.functionAttributeValue.getText()));
+ }
+
+ return functionExpression;
+ }
+
+ private String removeStringQuote(String src) {
+ if (src.charAt(0) == '\'' && src.charAt(src.length() - 1) == '\'') {
+ return src.substring(1, src.length() - 1);
+ } else if (src.charAt(0) == '\"' && src.charAt(src.length() - 1) == '\"') {
+ return src.substring(1, src.length() - 1);
+ } else {
+ throw new IllegalArgumentException("error format for string with quote:" + src);
+ }
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/strategy/LogicalGenerator.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/strategy/LogicalGenerator.java
new file mode 100644
index 0000000..a643d01
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/strategy/LogicalGenerator.java
@@ -0,0 +1,66 @@
+/*
+ * 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.influxdb.qp.strategy;
+
+import org.apache.iotdb.influxdb.qp.logical.Operator;
+import org.apache.iotdb.influxdb.qp.sql.InfluxDBLexer;
+import org.apache.iotdb.influxdb.qp.sql.InfluxDBParser;
+import org.apache.iotdb.influxdb.qp.sql.InfluxDBSqlVisitor;
+
+import org.antlr.v4.runtime.CharStream;
+import org.antlr.v4.runtime.CharStreams;
+import org.antlr.v4.runtime.CommonTokenStream;
+import org.antlr.v4.runtime.atn.PredictionMode;
+import org.antlr.v4.runtime.misc.ParseCancellationException;
+import org.antlr.v4.runtime.tree.ParseTree;
+
+/** LogicalGenerator. */
+public class LogicalGenerator {
+ public static Operator generate(String sql) throws ParseCancellationException {
+ InfluxDBSqlVisitor influxDBSqlVisitor = new InfluxDBSqlVisitor();
+ CharStream charStream1 = CharStreams.fromString(sql);
+ InfluxDBLexer lexer1 = new InfluxDBLexer(charStream1);
+ lexer1.removeErrorListeners();
+ lexer1.addErrorListener(SQLParseError.INSTANCE);
+ CommonTokenStream tokens1 = new CommonTokenStream(lexer1);
+ InfluxDBParser parser1 = new InfluxDBParser(tokens1);
+ parser1.getInterpreter().setPredictionMode(PredictionMode.SLL);
+ parser1.removeErrorListeners();
+ parser1.addErrorListener(SQLParseError.INSTANCE);
+ ParseTree tree;
+ try {
+ tree = parser1.singleStatement(); // STAGE 1
+ } catch (Exception ex) {
+ CharStream charStream2 = CharStreams.fromString(sql);
+ InfluxDBLexer lexer2 = new InfluxDBLexer(charStream2);
+ lexer2.removeErrorListeners();
+ lexer2.addErrorListener(SQLParseError.INSTANCE);
+ CommonTokenStream tokens2 = new CommonTokenStream(lexer2);
+ InfluxDBParser parser2 = new InfluxDBParser(tokens2);
+ parser2.getInterpreter().setPredictionMode(PredictionMode.LL);
+ parser2.removeErrorListeners();
+ parser2.addErrorListener(SQLParseError.INSTANCE);
+ tree = parser2.singleStatement(); // STAGE 2
+ // if we parse ok, it's LL not SLL
+ }
+ return influxDBSqlVisitor.visit(tree);
+ }
+
+ private LogicalGenerator() {}
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/strategy/SQLParseError.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/strategy/SQLParseError.java
new file mode 100644
index 0000000..ed151c0
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/strategy/SQLParseError.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.influxdb.qp.strategy;
+
+import org.antlr.v4.runtime.BaseErrorListener;
+import org.antlr.v4.runtime.RecognitionException;
+import org.antlr.v4.runtime.Recognizer;
+import org.antlr.v4.runtime.misc.ParseCancellationException;
+
+public class SQLParseError extends BaseErrorListener {
+
+ public static final SQLParseError INSTANCE = new SQLParseError();
+
+ @Override
+ public void syntaxError(
+ Recognizer<?, ?> recognizer,
+ Object offendingSymbol,
+ int line,
+ int charPositionInLine,
+ String msg,
+ RecognitionException e) {
+ throw new ParseCancellationException("line " + line + ":" + charPositionInLine + " " + msg);
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/utils/DatetimeUtils.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/utils/DatetimeUtils.java
new file mode 100644
index 0000000..a592619
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/utils/DatetimeUtils.java
@@ -0,0 +1,138 @@
+/*
+ * 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.influxdb.qp.utils;
+
+import java.util.Calendar;
+
+public class DatetimeUtils {
+
+ public static String timestampPrecision = "ms";
+ /**
+ * convert duration string to time value.
+ *
+ * @param duration represent duration string like: 12d8m9ns, 1y1mo, etc.
+ * @return time in milliseconds, microseconds, or nanoseconds depending on the profile
+ */
+ public static long convertDurationStrToLong(long currentTime, String duration) {
+ // TODD
+ long total = 0;
+ long temp = 0;
+ for (int i = 0; i < duration.length(); i++) {
+ char ch = duration.charAt(i);
+ if (Character.isDigit(ch)) {
+ temp *= 10;
+ temp += (ch - '0');
+ } else {
+ String unit = duration.charAt(i) + "";
+ // This is to identify units with two letters.
+ if (i + 1 < duration.length() && !Character.isDigit(duration.charAt(i + 1))) {
+ i++;
+ unit += duration.charAt(i);
+ }
+ total +=
+ DatetimeUtils.convertDurationStrToLong(
+ currentTime == -1 ? -1 : currentTime + total,
+ temp,
+ unit.toLowerCase(),
+ timestampPrecision);
+ temp = 0;
+ }
+ }
+ return total;
+ }
+
+ /** convert duration string to millisecond, microsecond or nanosecond. */
+ public static long convertDurationStrToLong(
+ long currentTime, long value, String unit, String timestampPrecision) {
+ DurationUnit durationUnit = DurationUnit.valueOf(unit);
+ long res = value;
+ switch (durationUnit) {
+ case y:
+ res *= 365 * 86_400_000L;
+ break;
+ case mo:
+ if (currentTime == -1) {
+ res *= 30 * 86_400_000L;
+ } else {
+ Calendar calendar = Calendar.getInstance();
+ calendar.setTimeInMillis(currentTime);
+ calendar.add(Calendar.MONTH, (int) (value));
+ res = calendar.getTimeInMillis() - currentTime;
+ }
+ break;
+ case w:
+ res *= 7 * 86_400_000L;
+ break;
+ case d:
+ res *= 86_400_000L;
+ break;
+ case h:
+ res *= 3_600_000L;
+ break;
+ case m:
+ res *= 60_000L;
+ break;
+ case s:
+ res *= 1_000L;
+ break;
+ default:
+ break;
+ }
+
+ if (timestampPrecision.equals("us")) {
+ if (unit.equals(DurationUnit.ns.toString())) {
+ return value / 1000;
+ } else if (unit.equals(DurationUnit.us.toString())) {
+ return value;
+ } else {
+ return res * 1000;
+ }
+ } else if (timestampPrecision.equals("ns")) {
+ if (unit.equals(DurationUnit.ns.toString())) {
+ return value;
+ } else if (unit.equals(DurationUnit.us.toString())) {
+ return value * 1000;
+ } else {
+ return res * 1000_000;
+ }
+ } else {
+ if (unit.equals(DurationUnit.ns.toString())) {
+ return value / 1000_000;
+ } else if (unit.equals(DurationUnit.us.toString())) {
+ return value / 1000;
+ } else {
+ return res;
+ }
+ }
+ }
+
+ public enum DurationUnit {
+ y,
+ mo,
+ w,
+ d,
+ h,
+ m,
+ s,
+ ms,
+ us,
+ ns
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/utils/MathUtil.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/utils/MathUtil.java
new file mode 100644
index 0000000..f25e5c4
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/utils/MathUtil.java
@@ -0,0 +1,75 @@
+/*
+ * 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.influxdb.qp.utils;
+
+import java.util.List;
+
+public class MathUtil {
+ /**
+ * calculate sum of list
+ *
+ * @param data need to calculate list
+ * @return sum of list
+ */
+ public static double Sum(List<Double> data) {
+ double sum = 0;
+ for (int i = 0; i < data.size(); i++) sum = sum + data.get(i);
+ return sum;
+ }
+
+ /**
+ * calculate mean of list
+ *
+ * @param data need to calculate list
+ * @return mean of list
+ */
+ public static double Mean(List<Double> data) {
+ double mean = 0;
+ mean = Sum(data) / data.size();
+ return mean;
+ }
+
+ /**
+ * calculate pop variance of list
+ *
+ * @param data need to calculate list
+ * @return pop variance of list
+ */
+ public static double POP_Variance(List<Double> data) {
+ double variance = 0;
+ for (int i = 0; i < data.size(); i++) {
+ variance = variance + (Math.pow((data.get(i) - Mean(data)), 2));
+ }
+ variance = variance / data.size();
+ return variance;
+ }
+
+ /**
+ * calculate pop std dev of list
+ *
+ * @param data need to calculate list
+ * @return pop std dev of list
+ */
+ public static double POP_STD_dev(List<Double> data) {
+ double std_dev;
+ std_dev = Math.sqrt(POP_Variance(data));
+ return std_dev;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/utils/TypeUtil.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/utils/TypeUtil.java
new file mode 100644
index 0000000..23d750b
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/qp/utils/TypeUtil.java
@@ -0,0 +1,26 @@
+/*
+ * 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.influxdb.qp.utils;
+
+public class TypeUtil {
+ public static boolean checkDecimal(Object object) {
+ return object instanceof Number;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/Expression.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/Expression.java
new file mode 100644
index 0000000..017b930
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/Expression.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.influxdb.query.expression;
+
+public interface Expression {}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/ResultColumn.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/ResultColumn.java
new file mode 100644
index 0000000..93373a6
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/ResultColumn.java
@@ -0,0 +1,52 @@
+/*
+ * 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.influxdb.query.expression;
+
+public class ResultColumn {
+
+ private final Expression expression;
+ private final String alias;
+
+ public ResultColumn(Expression expression, String alias) {
+ this.expression = expression;
+ this.alias = alias;
+ }
+
+ public ResultColumn(Expression expression) {
+ this.expression = expression;
+ alias = null;
+ }
+
+ public Expression getExpression() {
+ return expression;
+ }
+
+ public boolean hasAlias() {
+ return alias != null;
+ }
+
+ public String getAlias() {
+ return alias;
+ }
+
+ public String getResultColumnName() {
+ return alias != null ? alias : expression.toString();
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/AdditionExpression.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/AdditionExpression.java
new file mode 100644
index 0000000..e9e7545
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/AdditionExpression.java
@@ -0,0 +1,34 @@
+/*
+ * 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.influxdb.query.expression.binary;
+
+import org.apache.iotdb.influxdb.query.expression.Expression;
+
+public class AdditionExpression extends BinaryExpression {
+
+ public AdditionExpression(Expression leftExpression, Expression rightExpression) {
+ super(leftExpression, rightExpression);
+ }
+
+ @Override
+ protected String operator() {
+ return "+";
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/BinaryExpression.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/BinaryExpression.java
new file mode 100644
index 0000000..557317c
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/BinaryExpression.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.influxdb.query.expression.binary;
+
+import org.apache.iotdb.influxdb.query.expression.Expression;
+
+public abstract class BinaryExpression implements Expression {
+
+ protected final Expression leftExpression;
+ protected final Expression rightExpression;
+
+ protected BinaryExpression(Expression leftExpression, Expression rightExpression) {
+ this.leftExpression = leftExpression;
+ this.rightExpression = rightExpression;
+ }
+
+ public Expression getLeftExpression() {
+ return leftExpression;
+ }
+
+ public Expression getRightExpression() {
+ return rightExpression;
+ }
+
+ @Override
+ public final String toString() {
+ return String.format(
+ "%s %s %s", leftExpression.toString(), operator(), rightExpression.toString());
+ }
+
+ protected abstract String operator();
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/DivisionExpression.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/DivisionExpression.java
new file mode 100644
index 0000000..8288788
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/DivisionExpression.java
@@ -0,0 +1,34 @@
+/*
+ * 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.influxdb.query.expression.binary;
+
+import org.apache.iotdb.influxdb.query.expression.Expression;
+
+public class DivisionExpression extends BinaryExpression {
+
+ public DivisionExpression(Expression leftExpression, Expression rightExpression) {
+ super(leftExpression, rightExpression);
+ }
+
+ @Override
+ protected String operator() {
+ return "/";
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/ModuloExpression.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/ModuloExpression.java
new file mode 100644
index 0000000..24adb3a
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/ModuloExpression.java
@@ -0,0 +1,34 @@
+/*
+ * 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.influxdb.query.expression.binary;
+
+import org.apache.iotdb.influxdb.query.expression.Expression;
+
+public class ModuloExpression extends BinaryExpression {
+
+ public ModuloExpression(Expression leftExpression, Expression rightExpression) {
+ super(leftExpression, rightExpression);
+ }
+
+ @Override
+ protected String operator() {
+ return "%";
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/MultiplicationExpression.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/MultiplicationExpression.java
new file mode 100644
index 0000000..e7e263d
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/MultiplicationExpression.java
@@ -0,0 +1,34 @@
+/*
+ * 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.influxdb.query.expression.binary;
+
+import org.apache.iotdb.influxdb.query.expression.Expression;
+
+public class MultiplicationExpression extends BinaryExpression {
+
+ public MultiplicationExpression(Expression leftExpression, Expression rightExpression) {
+ super(leftExpression, rightExpression);
+ }
+
+ @Override
+ protected String operator() {
+ return "*";
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/SubtractionExpression.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/SubtractionExpression.java
new file mode 100644
index 0000000..aa37023
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/binary/SubtractionExpression.java
@@ -0,0 +1,34 @@
+/*
+ * 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.influxdb.query.expression.binary;
+
+import org.apache.iotdb.influxdb.query.expression.Expression;
+
+public class SubtractionExpression extends BinaryExpression {
+
+ public SubtractionExpression(Expression leftExpression, Expression rightExpression) {
+ super(leftExpression, rightExpression);
+ }
+
+ @Override
+ protected String operator() {
+ return "-";
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/unary/FunctionExpression.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/unary/FunctionExpression.java
new file mode 100644
index 0000000..b131c42
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/unary/FunctionExpression.java
@@ -0,0 +1,119 @@
+/*
+ * 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.influxdb.query.expression.unary;
+
+import org.apache.iotdb.influxdb.query.expression.Expression;
+
+import java.util.*;
+
+public class FunctionExpression implements Expression {
+
+ private final String functionName;
+ private final Map<String, String> functionAttributes;
+
+ private List<Expression> expressions;
+
+ private String expressionString;
+ private String parametersString;
+
+ public FunctionExpression(String functionName) {
+ this.functionName = functionName;
+ functionAttributes = new LinkedHashMap<>();
+ expressions = new ArrayList<>();
+ }
+
+ public void addAttribute(String key, String value) {
+ functionAttributes.put(key, value);
+ }
+
+ public void addExpression(Expression expression) {
+ expressions.add(expression);
+ }
+
+ public void setExpressions(List<Expression> expressions) {
+ this.expressions = expressions;
+ }
+
+ public String getFunctionName() {
+ return functionName;
+ }
+
+ public Map<String, String> getFunctionAttributes() {
+ return functionAttributes;
+ }
+
+ public List<Expression> getExpressions() {
+ return expressions;
+ }
+
+ @Override
+ public String toString() {
+ if (expressionString == null) {
+ expressionString = functionName + "(" + getParametersString() + ")";
+ }
+ return expressionString;
+ }
+
+ /**
+ * Generates the parameter part of the function column name.
+ *
+ * <p>Example:
+ *
+ * <p>Full column name -> udf(root.sg.d.s1, sin(root.sg.d.s1), 'key1'='value1', 'key2'='value2')
+ *
+ * <p>The parameter part -> root.sg.d.s1, sin(root.sg.d.s1), 'key1'='value1', 'key2'='value2'
+ */
+ public String getParametersString() {
+ if (parametersString == null) {
+ StringBuilder builder = new StringBuilder();
+ if (!expressions.isEmpty()) {
+ builder.append(expressions.get(0).toString());
+ for (int i = 1; i < expressions.size(); ++i) {
+ builder.append(", ").append(expressions.get(i).toString());
+ }
+ }
+ if (!functionAttributes.isEmpty()) {
+ if (!expressions.isEmpty()) {
+ builder.append(", ");
+ }
+ Iterator<Map.Entry<String, String>> iterator = functionAttributes.entrySet().iterator();
+ Map.Entry<String, String> entry = iterator.next();
+ builder
+ .append("\"")
+ .append(entry.getKey())
+ .append("\"=\"")
+ .append(entry.getValue())
+ .append("\"");
+ while (iterator.hasNext()) {
+ entry = iterator.next();
+ builder
+ .append(", ")
+ .append("\"")
+ .append(entry.getKey())
+ .append("\"=\"")
+ .append(entry.getValue())
+ .append("\"");
+ }
+ }
+ parametersString = builder.toString();
+ }
+ return parametersString;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/unary/NegationExpression.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/unary/NegationExpression.java
new file mode 100644
index 0000000..88227fe
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/unary/NegationExpression.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.influxdb.query.expression.unary;
+
+import org.apache.iotdb.influxdb.query.expression.Expression;
+
+public class NegationExpression implements Expression {
+ protected Expression expression;
+
+ public NegationExpression(Expression expression) {
+ this.expression = expression;
+ }
+
+ public Expression getExpression() {
+ return expression;
+ }
+
+ public void setExpression(Expression expression) {
+ this.expression = expression;
+ }
+}
diff --git a/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/unary/NodeExpression.java b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/unary/NodeExpression.java
new file mode 100644
index 0000000..36750bf
--- /dev/null
+++ b/iotdb-influxdb/src/main/java/org/apache/iotdb/influxdb/query/expression/unary/NodeExpression.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.influxdb.query.expression.unary;
+
+import org.apache.iotdb.influxdb.query.expression.Expression;
+
+public class NodeExpression implements Expression {
+ protected String name;
+
+ public NodeExpression(String name) {
+ this.name = name;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+}
diff --git a/iotdb-influxdb/src/test/java/org/apache/iotdb/influxdb/DesignSchemePerformanceTest.java b/iotdb-influxdb/src/test/java/org/apache/iotdb/influxdb/DesignSchemePerformanceTest.java
new file mode 100644
index 0000000..84140b8
--- /dev/null
+++ b/iotdb-influxdb/src/test/java/org/apache/iotdb/influxdb/DesignSchemePerformanceTest.java
@@ -0,0 +1,158 @@
+/*
+ * 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.influxdb;
+
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.SessionDataSet;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+import org.junit.Before;
+
+import java.util.*;
+
+/**
+ * The purpose of this class is to test the performance of two design schemes, store and read
+ * large-scale data for the two schemes respectively, and compare the reading efficiency
+ */
+public class DesignSchemePerformanceTest {
+
+ private static final String LOCAL_HOST = "127.0.0.1";
+ private static Session session;
+ private static final String ROOT_T1 = "root.teststress.test1";
+ private static final String ROOT_T2 = "root.teststress.test2";
+
+ @Before
+ public void setUp() throws IoTDBConnectionException {
+ // create session
+ //
+ session = new Session(LOCAL_HOST, 6667, "root", "root");
+ session.open(false);
+
+ // set session fetchSize
+ session.setFetchSize(10000);
+ }
+
+ public void Insert1() throws IoTDBConnectionException, StatementExecutionException { // 测试数据插入
+ String deviceId = ROOT_T1;
+ int fieldNum = 100;
+ int tagNum = 10;
+ List<String> measurements = new ArrayList<>();
+ List<TSDataType> types = new ArrayList<>();
+ for (int i = 0; i < fieldNum; i++) {
+ measurements.add(Util.generateWord(i));
+ types.add(TSDataType.INT64);
+ }
+ for (int i = 0; i < tagNum; i++) {
+ int baseline = 300;
+ String tmpTag = Util.generateWord(i + baseline);
+ System.out.println("tag:" + tmpTag);
+ measurements.add(tmpTag);
+ types.add(TSDataType.INT64);
+ }
+
+ for (long time = 0; time < 200000; time++) {
+ List<Object> values = new ArrayList<>();
+ for (long i = 0; i < fieldNum + tagNum; i++) {
+ values.add(time);
+ }
+ session.insertRecord(deviceId, time, measurements, types, values);
+ }
+ }
+
+ public void Insert2() throws IoTDBConnectionException, StatementExecutionException { // 测试数据插入
+ String deviceId = ROOT_T2;
+ int fieldNum = 100;
+ int tagNum = 10;
+ for (int i = 0; i < tagNum; i++) {
+ List<String> measurements = new ArrayList<>();
+ List<TSDataType> types = new ArrayList<>();
+ int baseline = 300;
+ String tmpTag = Util.generateWord(i + baseline);
+ deviceId += "." + tmpTag;
+ for (int j = 0; j < fieldNum; j++) {
+ measurements.add(Util.generateWord(j));
+ types.add(TSDataType.INT64);
+ }
+
+ for (long time = 0; time < 20000; time++) {
+ List<Object> values = new ArrayList<>();
+ for (long k = 0; k < fieldNum; k++) {
+ values.add(time);
+ }
+ session.insertRecord(deviceId, time, measurements, types, values);
+ }
+ }
+ }
+
+ public void Query1() throws IoTDBConnectionException, StatementExecutionException { // 测试数据查询
+
+ long before = System.currentTimeMillis();
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select * from root.teststress.test1 where RL = 1 and A = 1 and B =1 and C=1");
+ long after = System.currentTimeMillis();
+ long duration = (after - before);
+ System.out.println(duration);
+ // System.out.println(dataSet.getColumnNames());
+ // dataSet.setFetchSize(1024); // default is 10000
+ // int index = 0;
+ // while (dataSet.hasNext()) {
+ //// System.out.println(dataSet.next());
+ // index++;
+ // }
+ // System.out.println(index);
+
+ // dataSet.closeOperationHandle();
+ }
+
+ public void Query2() throws IoTDBConnectionException, StatementExecutionException { // 测试数据查询
+ long before = System.currentTimeMillis();
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select * from root.teststress.test2.*.*.*.*.SL where A=1 and B=1 and C=1");
+ dataSet =
+ session.executeQueryStatement(
+ "select * from root.teststress.test2.*.*.*.*.SL.* where A=1 and B=1 and C=1");
+ dataSet =
+ session.executeQueryStatement(
+ "select * from root.teststress.test2.*.*.*.*.SL.*.*.* where A=1 and B=1 and C=1");
+ dataSet =
+ session.executeQueryStatement(
+ "select * from root.teststress.test2.*.*.*.*.SL.*.*.*.* where A=1 and B=1 and C=1");
+ dataSet =
+ session.executeQueryStatement(
+ "select * from root.teststress.test2.*.*.*.*.SL.*.*.*.*.* where A=1 and B=1 ");
+ long after = System.currentTimeMillis();
+ long duration = (after - before);
+ System.out.println(duration);
+ System.out.println(dataSet.getColumnNames());
+ dataSet.setFetchSize(1024); // default is 10000
+ int index = 0;
+ while (dataSet.hasNext()) {
+ // System.out.println(dataSet.next());
+ index++;
+ }
+ System.out.println(index);
+
+ dataSet.closeOperationHandle();
+ }
+}
diff --git a/iotdb-influxdb/src/test/java/org/apache/iotdb/influxdb/IoTDBInfluxDBUtilsTest.java b/iotdb-influxdb/src/test/java/org/apache/iotdb/influxdb/IoTDBInfluxDBUtilsTest.java
new file mode 100644
index 0000000..8f12953
--- /dev/null
+++ b/iotdb-influxdb/src/test/java/org/apache/iotdb/influxdb/IoTDBInfluxDBUtilsTest.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.influxdb;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+
+public class IoTDBInfluxDBUtilsTest {
+
+ @Test
+ public void testGetSame() {
+ ArrayList<String> columns = new ArrayList<>();
+ columns.addAll(Arrays.asList("time", "root.111.1", "root.111.2", "root.222.1"));
+ ArrayList<Integer> list =
+ IoTDBInfluxDBUtils.getSamePathForList(columns.subList(1, columns.size()));
+ assert list.get(0) == 1;
+ assert list.get(1) == 2;
+ assert list.size() == 2;
+ }
+}
diff --git a/iotdb-influxdb/src/test/java/org/apache/iotdb/influxdb/Util.java b/iotdb-influxdb/src/test/java/org/apache/iotdb/influxdb/Util.java
new file mode 100644
index 0000000..5ca99e9
--- /dev/null
+++ b/iotdb-influxdb/src/test/java/org/apache/iotdb/influxdb/Util.java
@@ -0,0 +1,34 @@
+/*
+ * 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.influxdb;
+
+public class Util {
+ public static String generateWord(int index) {
+ if (index == 0) {
+ return "A";
+ }
+ StringBuilder result = new StringBuilder();
+ while (index > 0) {
+ result.append((char) ((index % 26) + 'A'));
+ index /= 26;
+ }
+ return result.toString();
+ }
+}
diff --git a/iotdb-influxdb/src/test/java/org/apache/iotdb/influxdb/integration/IoTDBInfluxDBIT.java b/iotdb-influxdb/src/test/java/org/apache/iotdb/influxdb/integration/IoTDBInfluxDBIT.java
new file mode 100644
index 0000000..84731bc
--- /dev/null
+++ b/iotdb-influxdb/src/test/java/org/apache/iotdb/influxdb/integration/IoTDBInfluxDBIT.java
@@ -0,0 +1,133 @@
+/*
+ * 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.influxdb.integration;
+
+import org.apache.iotdb.influxdb.IoTDBInfluxDBFactory;
+
+import org.influxdb.InfluxDB;
+import org.influxdb.dto.Point;
+import org.influxdb.dto.Query;
+import org.influxdb.dto.QueryResult;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.testcontainers.containers.GenericContainer;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+
+public class IoTDBInfluxDBIT {
+
+ @Rule
+ public GenericContainer IotDB =
+ new GenericContainer("apache/iotdb:latest").withExposedPorts(6667);
+
+ private InfluxDB influxDB;
+
+ @Before
+ public void setUp() {
+ influxDB =
+ IoTDBInfluxDBFactory.connect(
+ IotDB.getContainerIpAddress(), IotDB.getMappedPort(6667), "root", "root");
+ influxDB.createDatabase("database");
+ influxDB.setDatabase("database");
+ insertData();
+ }
+
+ private void insertData() {
+ // insert the build parameter to construct the influxdb
+ Point.Builder builder = Point.measurement("student");
+ Map<String, String> tags = new HashMap<>();
+ Map<String, Object> fields = new HashMap<>();
+ tags.put("name", "xie");
+ tags.put("sex", "m");
+ fields.put("score", 87);
+ fields.put("tel", "110");
+ fields.put("country", "china");
+ builder.tag(tags);
+ builder.fields(fields);
+ builder.time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
+ Point point = builder.build();
+ // after the build construction is completed, start writing
+ influxDB.write(point);
+
+ builder = Point.measurement("student");
+ tags = new HashMap<>();
+ fields = new HashMap<>();
+ tags.put("name", "xie");
+ tags.put("sex", "m");
+ tags.put("province", "anhui");
+ fields.put("score", 99);
+ fields.put("country", "china");
+ builder.tag(tags);
+ builder.fields(fields);
+ builder.time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
+ point = builder.build();
+ influxDB.write(point);
+ }
+
+ @Test
+ public void testCommonQueryColumn() {
+ Query query =
+ new Query(
+ "select * from student where (name=\"xie\" and sex=\"m\")or time<now()-7d", "database");
+ QueryResult result = influxDB.query(query);
+ QueryResult.Series series = result.getResults().get(0).getSeries().get(0);
+
+ String[] retArray = new String[] {"time", "name", "sex", "province", "country", "score", "tel"};
+ for (int i = 0; i < series.getColumns().size(); i++) {
+ assertEquals(retArray[i], series.getColumns().get(i));
+ }
+ }
+
+ @Test
+ public void testFuncWithoutFilter() {
+ Query query =
+ new Query(
+ "select max(score),min(score),sum(score),count(score),spread(score),mean(score),first(score),last(score) from student ",
+ "database");
+ QueryResult result = influxDB.query(query);
+ QueryResult.Series series = result.getResults().get(0).getSeries().get(0);
+
+ Object[] retArray = new Object[] {0, 99.0, 87.0, 186, 2, 12.0, 93, 87, 99};
+ for (int i = 0; i < series.getColumns().size(); i++) {
+ assertEquals(retArray[i], series.getValues().get(0).get(i));
+ }
+ }
+
+ @Test
+ public void testFunc() {
+ Query query =
+ new Query(
+ "select count(score),first(score),last(country),max(score),mean(score),median(score),min(score),mode(score),spread(score),stddev(score),sum(score) from student where (name=\"xie\" and sex=\"m\")or score<99",
+ "database");
+ QueryResult result = influxDB.query(query);
+ QueryResult.Series series = result.getResults().get(0).getSeries().get(0);
+ System.out.println("qqq" + series.toString());
+
+ Object[] retArray =
+ new Object[] {0, 2, 87, "china", 99.0, 93.0, 93.0, 87.0, 87, 12.0, 6.0, 186.0};
+ for (int i = 0; i < series.getColumns().size(); i++) {
+ assertEquals(retArray[i], series.getValues().get(0).get(i));
+ }
+ }
+}
diff --git a/site/src/main/.vuepress/config.js b/site/src/main/.vuepress/config.js
index 6a9d52b..dea51dc 100644
--- a/site/src/main/.vuepress/config.js
+++ b/site/src/main/.vuepress/config.js
@@ -1501,7 +1501,8 @@ var config = {
['API/Programming-Cpp-Native-API','C++ 原生接口'],
['API/Programming-Go-Native-API','Go 原生接口'],
['API/Programming-TsFile-API','TsFile API'],
- ['API/Time-zone','时区']
+ ['API/Time-zone','时区'],
+ ['API/IoTDB-InfluxDB','InfluxDB 协议适配器']
]
},
{