This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new fb08f7b [IOTDB-1808] Compatibility of Apache IoTDB with InfluxDB -
Query Func (#5104)
fb08f7b is described below
commit fb08f7bc277925ca5b8e26978e714556b90e415c
Author: Xieqijun <[email protected]>
AuthorDate: Sun Mar 20 22:45:56 2022 +0800
[IOTDB-1808] Compatibility of Apache IoTDB with InfluxDB - Query Func
(#5104)
Co-authored-by: Timcross-WEI <[email protected]>
Co-authored-by: Steve Yurong Su <[email protected]>
---
.../apache/iotdb/db/qp/sql/InfluxDBSqlParser.g4 | 2 +
.../org/apache/iotdb/influxdb/IoTDBInfluxDB.java | 4 +-
.../iotdb/influxdb/example/InfluxDBExample.java | 38 +-
.../protocol/impl/IoTDBInfluxDBService.java | 22 +-
.../iotdb/influxdb/session/InfluxDBSession.java | 29 +
.../influxdb/integration/IoTDBInfluxDBIT.java | 77 +-
.../{InfluxDBConstant.java => InfluxConstant.java} | 2 +-
.../influxdb/constant/InfluxSQLConstant.java | 7 +
.../protocol/influxdb/function/InfluxFunction.java | 58 ++
.../influxdb/function/InfluxFunctionFactory.java | 62 ++
.../InfluxFunctionValue.java} | 28 +-
.../aggregator/InfluxAggregator.java} | 17 +-
.../function/aggregator/InfluxCountFunction.java | 59 ++
.../function/aggregator/InfluxMeanFunction.java | 72 ++
.../function/aggregator/InfluxMedianFunction.java | 75 ++
.../function/aggregator/InfluxModeFunction.java | 88 ++
.../function/aggregator/InfluxSpreadFunction.java | 88 ++
.../function/aggregator/InfluxStddevFunction.java | 68 ++
.../function/aggregator/InfluxSumFunction.java | 68 ++
.../function/selector/InfluxFirstFunction.java | 73 ++
.../function/selector/InfluxLastFunction.java | 73 ++
.../function/selector/InfluxMaxFunction.java | 102 +++
.../function/selector/InfluxMinFunction.java | 102 +++
.../influxdb/function/selector/InfluxSelector.java | 57 ++
.../db/protocol/influxdb/handler/QueryHandler.java | 956 +++++++++++++++++++++
.../influxdb/meta/InfluxDBMetaManager.java | 8 +-
.../influxdb/operator/InfluxSelectComponent.java | 8 +-
.../CommonUtils.java} | 24 +-
.../InfluxDBConstant.java => util/FieldUtils.java} | 33 +-
.../db/protocol/influxdb/util/FilterUtils.java | 83 ++
.../protocol/influxdb/util/QueryResultUtils.java | 291 +++++++
.../db/protocol/influxdb/util/StringUtils.java | 97 +++
.../db/query/dataset/AlignByDeviceDataSet.java | 4 +
.../service/thrift/impl/InfluxDBServiceImpl.java | 14 +
.../java/org/apache/iotdb/db/utils/MathUtils.java | 51 ++
thrift-influxdb/src/main/thrift/influxdb.thrift | 19 +
36 files changed, 2814 insertions(+), 45 deletions(-)
diff --git
a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/InfluxDBSqlParser.g4
b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/InfluxDBSqlParser.g4
index 2df0017..316c4db 100644
--- a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/InfluxDBSqlParser.g4
+++ b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/InfluxDBSqlParser.g4
@@ -74,6 +74,8 @@ nodeName
| ID
| QUTOED_ID
| QUTOED_ID_IN_NODE_NAME
+ | LAST
+ | COUNT
;
// Identifier
diff --git
a/influxdb-protocol/src/main/java/org/apache/iotdb/influxdb/IoTDBInfluxDB.java
b/influxdb-protocol/src/main/java/org/apache/iotdb/influxdb/IoTDBInfluxDB.java
index 8b8e153..54d5065 100644
---
a/influxdb-protocol/src/main/java/org/apache/iotdb/influxdb/IoTDBInfluxDB.java
+++
b/influxdb-protocol/src/main/java/org/apache/iotdb/influxdb/IoTDBInfluxDB.java
@@ -187,8 +187,8 @@ public class IoTDBInfluxDB implements InfluxDB {
}
@Override
- public QueryResult query(final Query query) {
- throw new
UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ public QueryResult query(final Query queryReq) {
+ return influxDBService.query(queryReq);
}
@Override
diff --git
a/influxdb-protocol/src/main/java/org/apache/iotdb/influxdb/example/InfluxDBExample.java
b/influxdb-protocol/src/main/java/org/apache/iotdb/influxdb/example/InfluxDBExample.java
index 734bf07..995948a 100644
---
a/influxdb-protocol/src/main/java/org/apache/iotdb/influxdb/example/InfluxDBExample.java
+++
b/influxdb-protocol/src/main/java/org/apache/iotdb/influxdb/example/InfluxDBExample.java
@@ -23,6 +23,8 @@ 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 java.util.HashMap;
import java.util.Map;
@@ -37,6 +39,7 @@ public class InfluxDBExample {
influxDB.createDatabase("database");
influxDB.setDatabase("database");
insertData();
+ queryData();
influxDB.close();
}
@@ -61,7 +64,7 @@ public class InfluxDBExample {
tags.put("name", "xie");
tags.put("sex", "m");
tags.put("province", "anhui");
- fields.put("score", 99);
+ fields.put("score", 99.0);
fields.put("country", "china");
builder.tag(tags);
builder.fields(fields);
@@ -69,4 +72,37 @@ public class InfluxDBExample {
point = builder.build();
influxDB.write(point);
}
+
+ private static void queryData() {
+ 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());
+
+ // the selector query is parallel to the field value
+ query = new Query("select * from student ", "database");
+ result = influxDB.query(query);
+ System.out.println("query2 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),first(score),last(score) from
student ",
+ "database");
+ result = influxDB.query(query);
+ System.out.println("query3 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("query4 result:" +
result.getResults().get(0).getSeries().get(0).toString());
+ }
}
diff --git
a/influxdb-protocol/src/main/java/org/apache/iotdb/influxdb/protocol/impl/IoTDBInfluxDBService.java
b/influxdb-protocol/src/main/java/org/apache/iotdb/influxdb/protocol/impl/IoTDBInfluxDBService.java
index a701f4a..73944fc 100644
---
a/influxdb-protocol/src/main/java/org/apache/iotdb/influxdb/protocol/impl/IoTDBInfluxDBService.java
+++
b/influxdb-protocol/src/main/java/org/apache/iotdb/influxdb/protocol/impl/IoTDBInfluxDBService.java
@@ -21,11 +21,14 @@ package org.apache.iotdb.influxdb.protocol.impl;
import org.apache.iotdb.influxdb.session.InfluxDBSession;
import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSCreateDatabaseReq;
+import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSQueryReq;
import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSWritePointsReq;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.influxdb.InfluxDBException;
+import org.influxdb.dto.Query;
+import org.influxdb.dto.QueryResult;
public class IoTDBInfluxDBService {
@@ -54,10 +57,11 @@ public class IoTDBInfluxDBService {
} else {
tsWritePointsReq.setDatabase(database);
}
- tsWritePointsReq.setRetentionPolicy(retentionPolicy);
- tsWritePointsReq.setPrecision(precision);
- tsWritePointsReq.setConsistency(consistency);
- tsWritePointsReq.setLineProtocol(lineProtocol);
+ tsWritePointsReq
+ .setRetentionPolicy(retentionPolicy)
+ .setPrecision(precision)
+ .setConsistency(consistency)
+ .setLineProtocol(lineProtocol);
try {
influxDBSession.writePoints(tsWritePointsReq);
} catch (StatementExecutionException | IoTDBConnectionException e) {
@@ -79,6 +83,16 @@ public class IoTDBInfluxDBService {
}
}
+ public QueryResult query(Query queryReq) {
+ TSQueryReq tsQueryReq = new TSQueryReq();
+
tsQueryReq.setDatabase(queryReq.getDatabase()).setCommand(queryReq.getCommand());
+ try {
+ return influxDBSession.query(tsQueryReq);
+ } catch (StatementExecutionException | IoTDBConnectionException e) {
+ throw new InfluxDBException(e.getMessage());
+ }
+ }
+
public void close() {
influxDBSession.close();
}
diff --git
a/influxdb-protocol/src/main/java/org/apache/iotdb/influxdb/session/InfluxDBSession.java
b/influxdb-protocol/src/main/java/org/apache/iotdb/influxdb/session/InfluxDBSession.java
index 348ccb9..ba7a59c 100644
---
a/influxdb-protocol/src/main/java/org/apache/iotdb/influxdb/session/InfluxDBSession.java
+++
b/influxdb-protocol/src/main/java/org/apache/iotdb/influxdb/session/InfluxDBSession.java
@@ -26,6 +26,8 @@ import
org.apache.iotdb.protocol.influxdb.rpc.thrift.TSCloseSessionReq;
import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSCreateDatabaseReq;
import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSOpenSessionReq;
import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSOpenSessionResp;
+import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSQueryReq;
+import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSQueryResultRsp;
import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSWritePointsReq;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.RpcTransportFactory;
@@ -33,12 +35,14 @@ import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.Config;
+import com.google.gson.Gson;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.influxdb.InfluxDBException;
+import org.influxdb.dto.QueryResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -172,6 +176,31 @@ public class InfluxDBSession {
}
}
+ public QueryResult query(TSQueryReq request)
+ throws StatementExecutionException, IoTDBConnectionException {
+ request.setSessionId(sessionId);
+ try {
+ TSQueryResultRsp tsQueryResultRsp = client.query(request);
+ RpcUtils.verifySuccess(tsQueryResultRsp.status);
+ return new Gson().fromJson(tsQueryResultRsp.ResultJsonString,
QueryResult.class);
+ } catch (TException e) {
+ e.printStackTrace();
+ logger.error(e.getMessage());
+ if (reconnect()) {
+ try {
+ request.setSessionId(sessionId);
+ TSQueryResultRsp tsQueryResultRsp = client.query(request);
+ RpcUtils.verifySuccess(tsQueryResultRsp.status);
+ return new Gson().fromJson(tsQueryResultRsp.ResultJsonString,
QueryResult.class);
+ } catch (TException e1) {
+ throw new IoTDBConnectionException(e1);
+ }
+ } else {
+ throw new IoTDBConnectionException(MSG_RECONNECTION_FAIL);
+ }
+ }
+ }
+
public void createDatabase(TSCreateDatabaseReq request)
throws StatementExecutionException, IoTDBConnectionException {
request.setSessionId(sessionId);
diff --git
a/influxdb-protocol/src/test/java/org/apache/iotdb/influxdb/integration/IoTDBInfluxDBIT.java
b/influxdb-protocol/src/test/java/org/apache/iotdb/influxdb/integration/IoTDBInfluxDBIT.java
index 767962d..acf7d6e 100644
---
a/influxdb-protocol/src/test/java/org/apache/iotdb/influxdb/integration/IoTDBInfluxDBIT.java
+++
b/influxdb-protocol/src/test/java/org/apache/iotdb/influxdb/integration/IoTDBInfluxDBIT.java
@@ -24,6 +24,8 @@ import org.apache.iotdb.session.Session;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBException;
import org.influxdb.dto.Point;
+import org.influxdb.dto.Query;
+import org.influxdb.dto.QueryResult;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
@@ -64,40 +66,30 @@ public class IoTDBInfluxDBIT {
}
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", "A");
- tags.put("phone", "B");
- tags.put("sex", "C");
- fields.put("score", 99);
+ 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("address", "D");
- fields.put("score", 98);
- builder.tag(tags);
- builder.fields(fields);
- builder.time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
- point = builder.build();
- influxDB.write(point);
-
- builder = Point.measurement("student");
- tags = new HashMap<>();
- fields = new HashMap<>();
- tags.put("address", "D");
- tags.put("name", "A");
- tags.put("phone", "B");
- tags.put("sex", "C");
- fields.put("score", 97);
+ 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);
@@ -162,4 +154,49 @@ public class IoTDBInfluxDBIT {
public void testConnectAuthFailed() {
InfluxDB influxDB = IoTDBInfluxDBFactory.connect(host, port, "error",
"error");
}
+
+ @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);
+
+ 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/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxDBConstant.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxConstant.java
similarity index 96%
copy from
server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxDBConstant.java
copy to
server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxConstant.java
index 6b59698..3850201 100644
---
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxDBConstant.java
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxConstant.java
@@ -18,7 +18,7 @@
*/
package org.apache.iotdb.db.protocol.influxdb.constant;
-public class InfluxDBConstant {
+public class InfluxConstant {
public static final String PLACE_HOLDER = "PH";
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxSQLConstant.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxSQLConstant.java
index cbc77b7..3232778 100644
---
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxSQLConstant.java
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxSQLConstant.java
@@ -27,6 +27,7 @@ import java.util.Set;
/** this class contains several constants used in SQL. */
public class InfluxSQLConstant extends SQLConstant {
+ public static final String STAR = "*";
public static final String MAX = "max";
public static final String MIN = "min";
@@ -41,6 +42,8 @@ public class InfluxSQLConstant extends SQLConstant {
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));
+ private static final Set<String> ONLY_TRAVERSE_FUNCTION_NAMES =
+ new HashSet<>(Arrays.asList(MEDIAN, MODE, SPREAD, STDDEV));
public InfluxSQLConstant() throws InstantiationException {
// forbidding instantiation
@@ -54,4 +57,8 @@ public class InfluxSQLConstant extends SQLConstant {
public static Set<String> getNativeSelectorFunctionNames() {
return NATIVE_SELECTOR_FUNCTION_NAMES;
}
+
+ public static Set<String> getOnlyTraverseFunctionNames() {
+ return ONLY_TRAVERSE_FUNCTION_NAMES;
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/InfluxFunction.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/InfluxFunction.java
new file mode 100644
index 0000000..f687378
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/InfluxFunction.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.protocol.influxdb.function;
+
+import org.apache.iotdb.db.query.expression.Expression;
+import org.apache.iotdb.db.query.expression.unary.TimeSeriesOperand;
+
+import java.util.List;
+
+public abstract class InfluxFunction {
+
+ // contain possible parameters
+ private List<Expression> expressionList;
+
+ public InfluxFunction(List<Expression> expressionList) {
+ this.expressionList = expressionList;
+ }
+
+ public InfluxFunction() {}
+
+ public List<Expression> getExpressions() {
+ return this.expressionList;
+ }
+
+ public String getParmaName() {
+ if (expressionList == null) {
+ throw new IllegalArgumentException("not support param");
+ }
+ TimeSeriesOperand parmaExpression = (TimeSeriesOperand)
expressionList.get(0);
+ return parmaExpression.getPath().getFullPath();
+ }
+
+ public abstract String getFunctionName();
+
+ // calculate result by brute force
+ public abstract InfluxFunctionValue calculateBruteForce();
+
+ // calculate result by iotdb func
+ public abstract InfluxFunctionValue calculateByIoTDBFunc();
+
+ public abstract void updateValueIoTDBFunc(InfluxFunctionValue...
functionValues);
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/InfluxFunctionFactory.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/InfluxFunctionFactory.java
new file mode 100644
index 0000000..505a0b4
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/InfluxFunctionFactory.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.protocol.influxdb.function;
+
+import org.apache.iotdb.db.protocol.influxdb.constant.InfluxSQLConstant;
+import org.apache.iotdb.db.protocol.influxdb.function.aggregator.*;
+import
org.apache.iotdb.db.protocol.influxdb.function.selector.InfluxFirstFunction;
+import
org.apache.iotdb.db.protocol.influxdb.function.selector.InfluxLastFunction;
+import
org.apache.iotdb.db.protocol.influxdb.function.selector.InfluxMaxFunction;
+import
org.apache.iotdb.db.protocol.influxdb.function.selector.InfluxMinFunction;
+import org.apache.iotdb.db.query.expression.Expression;
+
+import java.util.List;
+
+public class InfluxFunctionFactory {
+
+ public static InfluxFunction generateFunction(
+ String functionName, List<Expression> expressionList) {
+ switch (functionName) {
+ case InfluxSQLConstant.MAX:
+ return new InfluxMaxFunction(expressionList);
+ case InfluxSQLConstant.MIN:
+ return new InfluxMinFunction(expressionList);
+ case InfluxSQLConstant.LAST:
+ return new InfluxLastFunction(expressionList);
+ case InfluxSQLConstant.FIRST:
+ return new InfluxFirstFunction(expressionList);
+ case InfluxSQLConstant.COUNT:
+ return new InfluxCountFunction(expressionList);
+ case InfluxSQLConstant.MEAN:
+ return new InfluxMeanFunction(expressionList);
+ case InfluxSQLConstant.MEDIAN:
+ return new InfluxMedianFunction(expressionList);
+ case InfluxSQLConstant.MODE:
+ return new InfluxModeFunction(expressionList);
+ case InfluxSQLConstant.SPREAD:
+ return new InfluxSpreadFunction(expressionList);
+ case InfluxSQLConstant.STDDEV:
+ return new InfluxStddevFunction(expressionList);
+ case InfluxSQLConstant.SUM:
+ return new InfluxSumFunction(expressionList);
+ default:
+ throw new IllegalArgumentException("not support aggregation name:" +
functionName);
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxDBConstant.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/InfluxFunctionValue.java
similarity index 60%
copy from
server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxDBConstant.java
copy to
server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/InfluxFunctionValue.java
index 6b59698..104988b 100644
---
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxDBConstant.java
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/InfluxFunctionValue.java
@@ -16,9 +16,31 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.protocol.influxdb.constant;
+package org.apache.iotdb.db.protocol.influxdb.function;
-public class InfluxDBConstant {
+public class InfluxFunctionValue {
- public static final String PLACE_HOLDER = "PH";
+ private Object value;
+ private Long timestamp;
+
+ public InfluxFunctionValue(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/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxDBConstant.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxAggregator.java
similarity index 60%
copy from
server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxDBConstant.java
copy to
server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxAggregator.java
index 6b59698..5d82ad9 100644
---
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxDBConstant.java
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxAggregator.java
@@ -16,9 +16,20 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.protocol.influxdb.constant;
-public class InfluxDBConstant {
+package org.apache.iotdb.db.protocol.influxdb.function.aggregator;
- public static final String PLACE_HOLDER = "PH";
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunction;
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunctionValue;
+import org.apache.iotdb.db.query.expression.Expression;
+
+import java.util.List;
+
+public abstract class InfluxAggregator extends InfluxFunction {
+
+ public InfluxAggregator(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ public abstract void updateValueBruteForce(InfluxFunctionValue
functionValue);
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxCountFunction.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxCountFunction.java
new file mode 100644
index 0000000..f6acb42
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxCountFunction.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.db.protocol.influxdb.function.aggregator;
+
+import org.apache.iotdb.db.protocol.influxdb.constant.InfluxSQLConstant;
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunctionValue;
+import org.apache.iotdb.db.query.expression.Expression;
+
+import java.util.List;
+
+public class InfluxCountFunction extends InfluxAggregator {
+ private int countNum = 0;
+
+ public InfluxCountFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateBruteForce() {
+ return new InfluxFunctionValue(this.countNum, 0L);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateByIoTDBFunc() {
+ return calculateBruteForce();
+ }
+
+ @Override
+ public String getFunctionName() {
+ return InfluxSQLConstant.COUNT;
+ }
+
+ @Override
+ public void updateValueBruteForce(InfluxFunctionValue functionValue) {
+ this.countNum++;
+ }
+
+ @Override
+ public void updateValueIoTDBFunc(InfluxFunctionValue... functionValues) {
+ this.countNum += (long) functionValues[0].getValue();
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxMeanFunction.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxMeanFunction.java
new file mode 100644
index 0000000..8a5a0c6
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxMeanFunction.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.protocol.influxdb.function.aggregator;
+
+import org.apache.iotdb.db.protocol.influxdb.constant.InfluxSQLConstant;
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunctionValue;
+import org.apache.iotdb.db.query.expression.Expression;
+import org.apache.iotdb.db.utils.MathUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class InfluxMeanFunction extends InfluxAggregator {
+ private final List<Double> numbers = new ArrayList<>();
+ long sum = 0;
+ long count = 0;
+
+ public InfluxMeanFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateBruteForce() {
+ return new InfluxFunctionValue(numbers.size() == 0 ? numbers :
MathUtils.mean(numbers), 0L);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateByIoTDBFunc() {
+ return new InfluxFunctionValue(count != 0 ? sum / count : null, count != 0
? 0L : null);
+ }
+
+ @Override
+ public String getFunctionName() {
+ return InfluxSQLConstant.MEAN;
+ }
+
+ @Override
+ public void updateValueBruteForce(InfluxFunctionValue functionValue) {
+ Object value = functionValue.getValue();
+ if (value instanceof Number) {
+ numbers.add(((Number) value).doubleValue());
+ } else {
+ throw new IllegalArgumentException("mean is not a valid type");
+ }
+ }
+
+ @Override
+ public void updateValueIoTDBFunc(InfluxFunctionValue... functionValues) {
+ if (functionValues.length == 1) {
+ count += (long) functionValues[0].getValue();
+ } else if (functionValues.length == 2) {
+ sum += (long) functionValues[1].getValue();
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxMedianFunction.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxMedianFunction.java
new file mode 100644
index 0000000..33d919a
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxMedianFunction.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.db.protocol.influxdb.function.aggregator;
+
+import org.apache.iotdb.db.protocol.influxdb.constant.InfluxSQLConstant;
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunctionValue;
+import org.apache.iotdb.db.query.expression.Expression;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+public class InfluxMedianFunction extends InfluxAggregator {
+ private final List<Double> numbers = new ArrayList<>();
+
+ public InfluxMedianFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateBruteForce() {
+ Collections.sort(numbers);
+ int len = numbers.size();
+ if (len > 0) {
+ if (len % 2 == 0) {
+ return new InfluxFunctionValue((numbers.get(len / 2) + numbers.get(len
/ 2 - 1)) / 2, 0L);
+ } else {
+ return new InfluxFunctionValue(numbers.get(len / 2), 0L);
+ }
+ }
+ return new InfluxFunctionValue(null, null);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateByIoTDBFunc() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getFunctionName() {
+ return InfluxSQLConstant.MEDIAN;
+ }
+
+ @Override
+ public void updateValueBruteForce(InfluxFunctionValue functionValue) {
+ Object value = functionValue.getValue();
+ if (value instanceof Number) {
+ numbers.add(((Number) value).doubleValue());
+ } else {
+ throw new IllegalArgumentException("not support this type");
+ }
+ }
+
+ @Override
+ public void updateValueIoTDBFunc(InfluxFunctionValue... functionValues) {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxModeFunction.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxModeFunction.java
new file mode 100644
index 0000000..4850104
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxModeFunction.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.protocol.influxdb.function.aggregator;
+
+import org.apache.iotdb.db.protocol.influxdb.constant.InfluxSQLConstant;
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunctionValue;
+import org.apache.iotdb.db.query.expression.Expression;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class InfluxModeFunction extends InfluxAggregator {
+ private final Map<Object, Integer> valueOrders = new HashMap<>();
+ private final Map<Object, Long> valueLastTimestamp = new HashMap<>();
+ private int maxNumber = 0;
+ private Object maxObject = null;
+
+ public InfluxModeFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateBruteForce() {
+ return new InfluxFunctionValue(maxObject, 0L);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateByIoTDBFunc() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getFunctionName() {
+ return InfluxSQLConstant.MODE;
+ }
+
+ @Override
+ public void updateValueBruteForce(InfluxFunctionValue 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 void updateValueIoTDBFunc(InfluxFunctionValue... functionValues) {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxSpreadFunction.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxSpreadFunction.java
new file mode 100644
index 0000000..b76ac49
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxSpreadFunction.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.protocol.influxdb.function.aggregator;
+
+import org.apache.iotdb.db.protocol.influxdb.constant.InfluxSQLConstant;
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunctionValue;
+import org.apache.iotdb.db.query.expression.Expression;
+
+import java.util.List;
+
+public class InfluxSpreadFunction extends InfluxAggregator {
+ private Double maxNum = null;
+ private Double minNum = null;
+
+ public InfluxSpreadFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateBruteForce() {
+ return new InfluxFunctionValue(maxNum - minNum, 0L);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateByIoTDBFunc() {
+ if (maxNum == null || minNum == null) {
+ return new InfluxFunctionValue(null, null);
+ }
+ return new InfluxFunctionValue(maxNum - minNum, 0L);
+ }
+
+ @Override
+ public String getFunctionName() {
+ return InfluxSQLConstant.SPREAD;
+ }
+
+ @Override
+ public void updateValueBruteForce(InfluxFunctionValue 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 void updateValueIoTDBFunc(InfluxFunctionValue... functionValues) {
+ if (functionValues.length == 1) {
+ double tmpValue = ((Number) functionValues[0].getValue()).doubleValue();
+ if (maxNum == null) {
+ maxNum = tmpValue;
+ } else if (tmpValue > maxNum) {
+ maxNum = tmpValue;
+ }
+ } else if (functionValues.length == 2) {
+ double tmpValue = ((Number) functionValues[1].getValue()).doubleValue();
+ if (minNum == null) {
+ minNum = tmpValue;
+ } else if (tmpValue < minNum) {
+ minNum = tmpValue;
+ }
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxStddevFunction.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxStddevFunction.java
new file mode 100644
index 0000000..614f402
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxStddevFunction.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.protocol.influxdb.function.aggregator;
+
+import org.apache.iotdb.db.protocol.influxdb.constant.InfluxSQLConstant;
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunctionValue;
+import org.apache.iotdb.db.query.expression.Expression;
+import org.apache.iotdb.db.utils.MathUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class InfluxStddevFunction extends InfluxAggregator {
+ private final List<Double> numbers = new ArrayList<>();
+
+ public InfluxStddevFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateBruteForce() {
+ return new InfluxFunctionValue(
+ numbers.size() == 0 ? numbers : MathUtils.popStdDev(numbers), 0L);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateByIoTDBFunc() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getFunctionName() {
+ return InfluxSQLConstant.STDDEV;
+ }
+
+ @Override
+ public void updateValueBruteForce(InfluxFunctionValue 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 void updateValueIoTDBFunc(InfluxFunctionValue... functionValues) {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxSumFunction.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxSumFunction.java
new file mode 100644
index 0000000..f175a1a
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/aggregator/InfluxSumFunction.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.protocol.influxdb.function.aggregator;
+
+import org.apache.iotdb.db.protocol.influxdb.constant.InfluxSQLConstant;
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunctionValue;
+import org.apache.iotdb.db.query.expression.Expression;
+import org.apache.iotdb.db.utils.MathUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class InfluxSumFunction extends InfluxAggregator {
+ private final List<Double> numbers = new ArrayList<>();
+ private long sum = 0;
+
+ public InfluxSumFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateBruteForce() {
+ return new InfluxFunctionValue(numbers.size() == 0 ? numbers :
MathUtils.sum(numbers), 0L);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateByIoTDBFunc() {
+ return new InfluxFunctionValue(sum, 0L);
+ }
+
+ @Override
+ public String getFunctionName() {
+ return InfluxSQLConstant.SUM;
+ }
+
+ @Override
+ public void updateValueBruteForce(InfluxFunctionValue 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 void updateValueIoTDBFunc(InfluxFunctionValue... functionValues) {
+ sum += (double) functionValues[0].getValue();
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/selector/InfluxFirstFunction.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/selector/InfluxFirstFunction.java
new file mode 100644
index 0000000..bba41c0
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/selector/InfluxFirstFunction.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.protocol.influxdb.function.selector;
+
+import org.apache.iotdb.db.protocol.influxdb.constant.InfluxSQLConstant;
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunctionValue;
+import org.apache.iotdb.db.query.expression.Expression;
+
+import java.util.List;
+
+public class InfluxFirstFunction extends InfluxSelector {
+ private Object value;
+
+ public InfluxFirstFunction(List<Expression> expressionList) {
+ super(expressionList);
+ this.setTimestamp(Long.MAX_VALUE);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateBruteForce() {
+ return new InfluxFunctionValue(value, this.getTimestamp());
+ }
+
+ @Override
+ public InfluxFunctionValue calculateByIoTDBFunc() {
+ return calculateBruteForce();
+ }
+
+ @Override
+ public void updateValueIoTDBFunc(InfluxFunctionValue... functionValues) {
+ if (value == null && getTimestamp() == null) {
+ value = functionValues[0].getValue();
+ setTimestamp(functionValues[0].getTimestamp());
+ } else if (getTimestamp() < functionValues[0].getTimestamp()) {
+ value = functionValues[0].getValue();
+ setTimestamp(functionValues[0].getTimestamp());
+ }
+ }
+
+ @Override
+ public void updateValueAndRelateValues(
+ InfluxFunctionValue functionValue, List<Object> relatedValues) {
+ Object value = functionValue.getValue();
+ Long timestamp = functionValue.getTimestamp();
+ if (timestamp <= this.getTimestamp()) {
+ this.value = value;
+ this.setTimestamp(timestamp);
+ this.setRelatedValues(relatedValues);
+ }
+ }
+
+ @Override
+ public String getFunctionName() {
+ return InfluxSQLConstant.FIRST;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/selector/InfluxLastFunction.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/selector/InfluxLastFunction.java
new file mode 100644
index 0000000..2674e0e
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/selector/InfluxLastFunction.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.protocol.influxdb.function.selector;
+
+import org.apache.iotdb.db.protocol.influxdb.constant.InfluxSQLConstant;
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunctionValue;
+import org.apache.iotdb.db.query.expression.Expression;
+
+import java.util.List;
+
+public class InfluxLastFunction extends InfluxSelector {
+ private Object value;
+
+ public InfluxLastFunction(List<Expression> expressionList) {
+ super(expressionList);
+ this.setTimestamp(Long.MIN_VALUE);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateBruteForce() {
+ return new InfluxFunctionValue(value, this.getTimestamp());
+ }
+
+ @Override
+ public InfluxFunctionValue calculateByIoTDBFunc() {
+ return calculateBruteForce();
+ }
+
+ @Override
+ public void updateValueIoTDBFunc(InfluxFunctionValue... functionValues) {
+ if (value == null && getTimestamp() == null) {
+ value = functionValues[0].getValue();
+ setTimestamp(functionValues[0].getTimestamp());
+ } else if (getTimestamp() > functionValues[0].getTimestamp()) {
+ value = functionValues[0].getValue();
+ setTimestamp(functionValues[0].getTimestamp());
+ }
+ }
+
+ @Override
+ public void updateValueAndRelateValues(
+ InfluxFunctionValue functionValue, List<Object> relatedValues) {
+ Object value = functionValue.getValue();
+ Long timestamp = functionValue.getTimestamp();
+ if (timestamp >= this.getTimestamp()) {
+ this.value = value;
+ this.setTimestamp(timestamp);
+ this.setRelatedValues(relatedValues);
+ }
+ }
+
+ @Override
+ public String getFunctionName() {
+ return InfluxSQLConstant.LAST;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/selector/InfluxMaxFunction.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/selector/InfluxMaxFunction.java
new file mode 100644
index 0000000..448955f
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/selector/InfluxMaxFunction.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.protocol.influxdb.function.selector;
+
+import org.apache.iotdb.db.protocol.influxdb.constant.InfluxSQLConstant;
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunctionValue;
+import org.apache.iotdb.db.query.expression.Expression;
+
+import java.util.List;
+
+public class InfluxMaxFunction extends InfluxSelector {
+ private Double doubleValue = Double.MIN_VALUE;
+ private String stringValue = null;
+ private boolean isNumber = false;
+ private boolean isString = false;
+
+ private Double maxValue = null;
+
+ public InfluxMaxFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateBruteForce() {
+ if (!isString && !isNumber) {
+ return new InfluxFunctionValue(null, null);
+ } else if (isString) {
+ return new InfluxFunctionValue(stringValue, this.getTimestamp());
+ } else {
+ return new InfluxFunctionValue(doubleValue, this.getTimestamp());
+ }
+ }
+
+ @Override
+ public InfluxFunctionValue calculateByIoTDBFunc() {
+ return new InfluxFunctionValue(maxValue, maxValue == null ? null : 0L);
+ }
+
+ @Override
+ public void updateValueIoTDBFunc(InfluxFunctionValue... functionValues) {
+ if (functionValues[0].getValue() instanceof Number) {
+ double tmpValue = ((Number) functionValues[0].getValue()).doubleValue();
+ if (maxValue == null) {
+ maxValue = tmpValue;
+ } else if (tmpValue > maxValue) {
+ maxValue = tmpValue;
+ }
+ }
+ }
+
+ @Override
+ public String getFunctionName() {
+ return InfluxSQLConstant.MAX;
+ }
+
+ @Override
+ public void updateValueAndRelateValues(
+ InfluxFunctionValue functionValue, List<Object> relatedValues) {
+ 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(relatedValues);
+ }
+ } else if (value instanceof String) {
+ String tmpValue = (String) value;
+ if (!isString) {
+ isString = true;
+ stringValue = tmpValue;
+ this.setTimestamp(timestamp);
+ this.setRelatedValues(relatedValues);
+ } else if (tmpValue.compareTo(this.stringValue) >= 0) {
+ stringValue = tmpValue;
+ this.setTimestamp(timestamp);
+ this.setRelatedValues(relatedValues);
+ }
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/selector/InfluxMinFunction.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/selector/InfluxMinFunction.java
new file mode 100644
index 0000000..073a640
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/selector/InfluxMinFunction.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.protocol.influxdb.function.selector;
+
+import org.apache.iotdb.db.protocol.influxdb.constant.InfluxSQLConstant;
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunctionValue;
+import org.apache.iotdb.db.query.expression.Expression;
+
+import java.util.List;
+
+public class InfluxMinFunction extends InfluxSelector {
+ private Double doubleValue = Double.MAX_VALUE;
+ private String stringValue = null;
+ private boolean isNumber = false;
+ private boolean isString = false;
+
+ private Double minValue = null;
+
+ public InfluxMinFunction(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ @Override
+ public InfluxFunctionValue calculateBruteForce() {
+ if (!isString && !isNumber) {
+ return new InfluxFunctionValue(null, null);
+ } else if (isString) {
+ return new InfluxFunctionValue(stringValue, this.getTimestamp());
+ } else {
+ return new InfluxFunctionValue(doubleValue, this.getTimestamp());
+ }
+ }
+
+ @Override
+ public InfluxFunctionValue calculateByIoTDBFunc() {
+ return new InfluxFunctionValue(minValue, minValue == null ? null : 0L);
+ }
+
+ @Override
+ public void updateValueIoTDBFunc(InfluxFunctionValue... functionValues) {
+ if (functionValues[0].getValue() instanceof Number) {
+ double tmpValue = ((Number) functionValues[0].getValue()).doubleValue();
+ if (minValue == null) {
+ minValue = tmpValue;
+ } else if (tmpValue < minValue) {
+ minValue = tmpValue;
+ }
+ }
+ }
+
+ @Override
+ public String getFunctionName() {
+ return InfluxSQLConstant.MIN;
+ }
+
+ @Override
+ public void updateValueAndRelateValues(
+ InfluxFunctionValue functionValue, List<Object> relatedValues) {
+ 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(relatedValues);
+ }
+ } else if (value instanceof String) {
+ String tmpValue = (String) value;
+ if (!isString) {
+ isString = true;
+ stringValue = tmpValue;
+ this.setTimestamp(timestamp);
+ this.setRelatedValues(relatedValues);
+ } else if (tmpValue.compareTo(this.stringValue) <= 0) {
+ stringValue = tmpValue;
+ this.setTimestamp(timestamp);
+ this.setRelatedValues(relatedValues);
+ }
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/selector/InfluxSelector.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/selector/InfluxSelector.java
new file mode 100644
index 0000000..c4c9e33
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/function/selector/InfluxSelector.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.db.protocol.influxdb.function.selector;
+
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunction;
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunctionValue;
+import org.apache.iotdb.db.query.expression.Expression;
+
+import java.util.List;
+
+public abstract class InfluxSelector extends InfluxFunction {
+
+ // The timestamp corresponding to the value
+ private Long timestamp = null;
+
+ private List<Object> relatedValues;
+
+ public InfluxSelector(List<Expression> expressionList) {
+ super(expressionList);
+ }
+
+ public List<Object> getRelatedValues() {
+ return this.relatedValues;
+ }
+
+ public void setRelatedValues(List<Object> relatedValues) {
+ this.relatedValues = relatedValues;
+ }
+
+ public Long getTimestamp() {
+ return timestamp;
+ }
+
+ public void setTimestamp(Long timestamp) {
+ this.timestamp = timestamp;
+ }
+
+ public abstract void updateValueAndRelateValues(
+ InfluxFunctionValue functionValue, List<Object> relatedValues);
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/handler/QueryHandler.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/handler/QueryHandler.java
new file mode 100644
index 0000000..e2c5317
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/handler/QueryHandler.java
@@ -0,0 +1,956 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.protocol.influxdb.handler;
+
+import org.apache.iotdb.db.auth.AuthException;
+import org.apache.iotdb.db.conf.IoTDBConstant;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.protocol.influxdb.constant.InfluxSQLConstant;
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunction;
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunctionFactory;
+import org.apache.iotdb.db.protocol.influxdb.function.InfluxFunctionValue;
+import
org.apache.iotdb.db.protocol.influxdb.function.aggregator.InfluxAggregator;
+import org.apache.iotdb.db.protocol.influxdb.function.selector.InfluxSelector;
+import org.apache.iotdb.db.protocol.influxdb.meta.InfluxDBMetaManager;
+import org.apache.iotdb.db.protocol.influxdb.operator.InfluxQueryOperator;
+import org.apache.iotdb.db.protocol.influxdb.operator.InfluxSelectComponent;
+import org.apache.iotdb.db.protocol.influxdb.util.FieldUtils;
+import org.apache.iotdb.db.protocol.influxdb.util.FilterUtils;
+import org.apache.iotdb.db.protocol.influxdb.util.QueryResultUtils;
+import org.apache.iotdb.db.protocol.influxdb.util.StringUtils;
+import org.apache.iotdb.db.qp.constant.FilterConstant;
+import org.apache.iotdb.db.qp.logical.Operator;
+import org.apache.iotdb.db.qp.logical.crud.BasicFunctionOperator;
+import org.apache.iotdb.db.qp.logical.crud.FilterOperator;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.expression.Expression;
+import org.apache.iotdb.db.query.expression.ResultColumn;
+import org.apache.iotdb.db.query.expression.unary.FunctionExpression;
+import org.apache.iotdb.db.query.expression.unary.TimeSeriesOperand;
+import org.apache.iotdb.db.service.basic.ServiceProvider;
+import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSQueryResultRsp;
+import org.apache.iotdb.rpc.RpcUtils;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.service.rpc.thrift.TSStatus;
+import
org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
+import org.apache.iotdb.tsfile.read.common.Field;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.expression.IExpression;
+import org.apache.iotdb.tsfile.read.expression.impl.SingleSeriesExpression;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+
+import com.google.gson.Gson;
+import org.apache.thrift.TException;
+import org.influxdb.InfluxDBException;
+import org.influxdb.dto.QueryResult;
+
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class QueryHandler {
+
+ public static TSQueryResultRsp queryInfluxDB(
+ String database,
+ InfluxQueryOperator queryOperator,
+ long sessionId,
+ ServiceProvider serviceProvider) {
+ String measurement =
queryOperator.getFromComponent().getPrefixPaths().get(0).getFullPath();
+ // The list of fields under the current measurement and the order of the
specified rules
+ Map<String, Integer> fieldOrders = getFieldOrders(database, measurement,
serviceProvider);
+ QueryResult queryResult;
+ TSQueryResultRsp tsQueryResultRsp = new TSQueryResultRsp();
+ try {
+ // contain filter condition or have common query the result of by
traversal.
+ if (queryOperator.getWhereComponent() != null
+ || queryOperator.getSelectComponent().isHasCommonQuery()
+ || queryOperator.getSelectComponent().isHasOnlyTraverseFunction()) {
+ // step1 : generate query results
+ queryResult =
+ queryExpr(
+ queryOperator.getWhereComponent() != null
+ ? queryOperator.getWhereComponent().getFilterOperator()
+ : null,
+ database,
+ measurement,
+ serviceProvider,
+ fieldOrders,
+ sessionId);
+ // step2 : select filter
+ ProcessSelectComponent(queryResult,
queryOperator.getSelectComponent());
+ }
+ // don't contain filter condition and only have function use iotdb
function.
+ else {
+ queryResult =
+ queryFuncWithoutFilter(
+ queryOperator.getSelectComponent(), database, measurement,
serviceProvider);
+ }
+ return tsQueryResultRsp
+ .setResultJsonString(new Gson().toJson(queryResult))
+ .setStatus(RpcUtils.getInfluxDBStatus(TSStatusCode.SUCCESS_STATUS));
+ } catch (AuthException e) {
+ return tsQueryResultRsp.setStatus(
+ RpcUtils.getInfluxDBStatus(
+ TSStatusCode.UNINITIALIZED_AUTH_ERROR.getStatusCode(),
e.getMessage()));
+ }
+ }
+
+ /**
+ * conditions are generated from subtrees of unique conditions
+ *
+ * @param basicFunctionOperator subtree to generate condition
+ * @return corresponding conditions
+ */
+ public static IExpression getIExpressionForBasicFunctionOperator(
+ BasicFunctionOperator basicFunctionOperator) {
+ return new SingleSeriesExpression(
+ basicFunctionOperator.getSinglePath(),
+ FilterUtils.filterTypeToFilter(
+ basicFunctionOperator.getFilterType(),
basicFunctionOperator.getValue()));
+ }
+
+ /**
+ * 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
+ */
+ public static void ProcessSelectComponent(
+ QueryResult queryResult, InfluxSelectComponent 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(InfluxSQLConstant.RESERVED_TIME);
+
+ // when have function
+ if (selectComponent.isHasFunction()) {
+ List<InfluxFunction> functions = new ArrayList<>();
+ for (ResultColumn resultColumn : selectComponent.getResultColumns()) {
+ Expression expression = resultColumn.getExpression();
+ if (expression instanceof FunctionExpression) {
+ String functionName = ((FunctionExpression)
expression).getFunctionName();
+ functions.add(
+ InfluxFunctionFactory.generateFunction(functionName,
expression.getExpressions()));
+ newColumns.add(functionName);
+ } else if (expression instanceof TimeSeriesOperand) {
+ String columnName = ((TimeSeriesOperand)
expression).getPath().getFullPath();
+ if (!columnName.equals(InfluxSQLConstant.STAR)) {
+ newColumns.add(columnName);
+ } else {
+ newColumns.addAll(columns.subList(1, columns.size()));
+ }
+ }
+ }
+ for (List<Object> value : values) {
+ for (InfluxFunction function : functions) {
+ List<Expression> expressions = function.getExpressions();
+ if (expressions == null) {
+ throw new IllegalArgumentException("not support param");
+ }
+ TimeSeriesOperand parmaExpression = (TimeSeriesOperand)
expressions.get(0);
+ String parmaName = parmaExpression.getPath().getFullPath();
+ 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 InfluxSelector) {
+ ((InfluxSelector) function)
+ .updateValueAndRelateValues(
+ new InfluxFunctionValue(selectedValue,
selectedTimestamp), value);
+ } else {
+ // aggregate function
+ ((InfluxAggregator) function)
+ .updateValueBruteForce(
+ new InfluxFunctionValue(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()) {
+ InfluxSelector selector = (InfluxSelector) functions.get(0);
+ List<Object> relatedValue = selector.getRelatedValues();
+ for (String column : newColumns) {
+ if
(InfluxSQLConstant.getNativeSelectorFunctionNames().contains(column)) {
+ value.add(selector.calculateBruteForce().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 (InfluxFunction function : functions) {
+ if (value.size() == 0) {
+ value.add(function.calculateBruteForce().getTimestamp());
+ } else {
+ value.set(0, function.calculateBruteForce().getTimestamp());
+ }
+ value.add(function.calculateBruteForce().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 TimeSeriesOperand) {
+ // not star case
+ if (!((TimeSeriesOperand) expression)
+ .getPath()
+ .getFullPath()
+ .equals(InfluxSQLConstant.STAR)) {
+ newColumns.add(((TimeSeriesOperand)
expression).getPath().getFullPath());
+ } 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;
+ }
+ QueryResultUtils.updateQueryResultColumnValue(
+ queryResult, StringUtils.removeDuplicate(newColumns), values);
+ }
+
+ public static Map<String, Integer> getFieldOrders(
+ String database, String measurement, ServiceProvider serviceProvider) {
+ Map<String, Integer> fieldOrders = new HashMap<>();
+ long queryId = ServiceProvider.SESSION_MANAGER.requestQueryId(true);
+ try {
+ String showTimeseriesSql = "show timeseries root." + database + '.' +
measurement + ".**";
+ PhysicalPlan physicalPlan =
+
serviceProvider.getPlanner().parseSQLToPhysicalPlan(showTimeseriesSql);
+ QueryContext queryContext =
+ serviceProvider.genQueryContext(
+ queryId,
+ true,
+ System.currentTimeMillis(),
+ showTimeseriesSql,
+ IoTDBConstant.DEFAULT_CONNECTION_TIMEOUT_MS);
+ QueryDataSet queryDataSet =
+ serviceProvider.createQueryDataSet(
+ queryContext, physicalPlan, IoTDBConstant.DEFAULT_FETCH_SIZE);
+ int fieldNums = 0;
+ Map<String, Integer> tagOrders =
+
InfluxDBMetaManager.getDatabase2Measurement2TagOrders().get(database).get(measurement);
+ int tagOrderNums = tagOrders.size();
+ while (queryDataSet.hasNext()) {
+ List<Field> fields = queryDataSet.next().getFields();
+ String filed =
StringUtils.getFieldByPath(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);
+ fieldNums++;
+ }
+ }
+ } catch (QueryProcessException
+ | TException
+ | StorageEngineException
+ | SQLException
+ | IOException
+ | InterruptedException
+ | QueryFilterOptimizationException
+ | MetadataException e) {
+ throw new InfluxDBException(e.getMessage());
+ } finally {
+
ServiceProvider.SESSION_MANAGER.releaseQueryResourceNoExceptions(queryId);
+ }
+ return fieldOrders;
+ }
+
+ /**
+ * 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
+ */
+ public static QueryResult queryFuncWithoutFilter(
+ InfluxSelectComponent selectComponent,
+ String database,
+ String measurement,
+ ServiceProvider serviceProvider) {
+ // columns
+ List<String> columns = new ArrayList<>();
+ columns.add(InfluxSQLConstant.RESERVED_TIME);
+
+ List<InfluxFunction> functions = new ArrayList<>();
+ String path = "root." + database + "." + measurement;
+ for (ResultColumn resultColumn : selectComponent.getResultColumns()) {
+ Expression expression = resultColumn.getExpression();
+ if (expression instanceof FunctionExpression) {
+ String functionName = ((FunctionExpression)
expression).getFunctionName();
+ functions.add(
+ InfluxFunctionFactory.generateFunction(functionName,
expression.getExpressions()));
+ columns.add(functionName);
+ }
+ }
+
+ List<Object> value = new ArrayList<>();
+ List<List<Object>> values = new ArrayList<>();
+ for (InfluxFunction function : functions) {
+ InfluxFunctionValue functionValue = updateByIoTDBFunc(function,
serviceProvider, path);
+ // InfluxFunctionValue 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;
+ }
+
+ private static InfluxFunctionValue updateByIoTDBFunc(
+ InfluxFunction function, ServiceProvider serviceProvider, String path) {
+ switch (function.getFunctionName()) {
+ case InfluxSQLConstant.COUNT:
+ {
+ long queryId = ServiceProvider.SESSION_MANAGER.requestQueryId(true);
+ String functionSql =
+ StringUtils.generateFunctionSql(
+ function.getFunctionName(), function.getParmaName(), path);
+ try {
+ QueryPlan queryPlan =
+ (QueryPlan)
serviceProvider.getPlanner().parseSQLToPhysicalPlan(functionSql);
+ QueryContext queryContext =
+ serviceProvider.genQueryContext(
+ queryId,
+ true,
+ System.currentTimeMillis(),
+ functionSql,
+ IoTDBConstant.DEFAULT_CONNECTION_TIMEOUT_MS);
+ QueryDataSet queryDataSet =
+ serviceProvider.createQueryDataSet(
+ queryContext, queryPlan, IoTDBConstant.DEFAULT_FETCH_SIZE);
+ while (queryDataSet.hasNext()) {
+ List<Field> fields = queryDataSet.next().getFields();
+ for (Field field : fields) {
+ function.updateValueIoTDBFunc(new
InfluxFunctionValue(field.getLongV(), null));
+ }
+ }
+ } catch (QueryProcessException
+ | QueryFilterOptimizationException
+ | StorageEngineException
+ | IOException
+ | MetadataException
+ | SQLException
+ | TException
+ | InterruptedException e) {
+ e.printStackTrace();
+ throw new InfluxDBException(e.getMessage());
+ } finally {
+
ServiceProvider.SESSION_MANAGER.releaseQueryResourceNoExceptions(queryId);
+ }
+ break;
+ }
+ case InfluxSQLConstant.MEAN:
+ {
+ long queryId = ServiceProvider.SESSION_MANAGER.requestQueryId(true);
+ try {
+ String functionSqlCount =
+ StringUtils.generateFunctionSql("count",
function.getParmaName(), path);
+ QueryPlan queryPlan =
+ (QueryPlan)
serviceProvider.getPlanner().parseSQLToPhysicalPlan(functionSqlCount);
+ QueryContext queryContext =
+ serviceProvider.genQueryContext(
+ queryId,
+ true,
+ System.currentTimeMillis(),
+ functionSqlCount,
+ IoTDBConstant.DEFAULT_CONNECTION_TIMEOUT_MS);
+ QueryDataSet queryDataSet =
+ serviceProvider.createQueryDataSet(
+ queryContext, queryPlan, IoTDBConstant.DEFAULT_FETCH_SIZE);
+ while (queryDataSet.hasNext()) {
+ List<Field> fields = queryDataSet.next().getFields();
+ for (Field field : fields) {
+ function.updateValueIoTDBFunc(new
InfluxFunctionValue(field.getLongV(), null));
+ }
+ }
+ } catch (QueryProcessException
+ | TException
+ | StorageEngineException
+ | SQLException
+ | IOException
+ | InterruptedException
+ | QueryFilterOptimizationException
+ | MetadataException e) {
+ throw new InfluxDBException(e.getMessage());
+ } finally {
+
ServiceProvider.SESSION_MANAGER.releaseQueryResourceNoExceptions(queryId);
+ }
+ long queryId1 = ServiceProvider.SESSION_MANAGER.requestQueryId(true);
+ try {
+ String functionSqlSum =
+ StringUtils.generateFunctionSql("sum",
function.getParmaName(), path);
+ QueryPlan queryPlan =
+ (QueryPlan)
serviceProvider.getPlanner().parseSQLToPhysicalPlan(functionSqlSum);
+ QueryContext queryContext =
+ serviceProvider.genQueryContext(
+ queryId,
+ true,
+ System.currentTimeMillis(),
+ functionSqlSum,
+ IoTDBConstant.DEFAULT_CONNECTION_TIMEOUT_MS);
+ QueryDataSet queryDataSet =
+ serviceProvider.createQueryDataSet(
+ queryContext, queryPlan, IoTDBConstant.DEFAULT_FETCH_SIZE);
+ while (queryDataSet.hasNext()) {
+ List<Field> fields = queryDataSet.next().getFields();
+ for (Field field : fields) {
+ function.updateValueIoTDBFunc(
+ null, new InfluxFunctionValue(field.getDoubleV(), null));
+ }
+ }
+ } catch (QueryProcessException
+ | TException
+ | StorageEngineException
+ | SQLException
+ | IOException
+ | InterruptedException
+ | QueryFilterOptimizationException
+ | MetadataException e) {
+ e.printStackTrace();
+ throw new InfluxDBException(e.getMessage());
+ } finally {
+
ServiceProvider.SESSION_MANAGER.releaseQueryResourceNoExceptions(queryId1);
+ }
+ break;
+ }
+ case InfluxSQLConstant.SPREAD:
+ {
+ long queryId = ServiceProvider.SESSION_MANAGER.requestQueryId(true);
+ try {
+ String functionSqlMaxValue =
+ StringUtils.generateFunctionSql("max_value",
function.getParmaName(), path);
+ QueryPlan queryPlan =
+ (QueryPlan)
+
serviceProvider.getPlanner().parseSQLToPhysicalPlan(functionSqlMaxValue);
+ QueryContext queryContext =
+ serviceProvider.genQueryContext(
+ queryId,
+ true,
+ System.currentTimeMillis(),
+ functionSqlMaxValue,
+ IoTDBConstant.DEFAULT_CONNECTION_TIMEOUT_MS);
+ QueryDataSet queryDataSet =
+ serviceProvider.createQueryDataSet(
+ queryContext, queryPlan, IoTDBConstant.DEFAULT_FETCH_SIZE);
+ while (queryDataSet.hasNext()) {
+ List<Path> paths = queryDataSet.getPaths();
+ List<Field> fields = queryDataSet.next().getFields();
+ for (int i = 0; i < paths.size(); i++) {
+ Object o = FieldUtils.iotdbFieldConvert(fields.get(i));
+ if (o instanceof Number) {
+ function.updateValueIoTDBFunc(
+ new InfluxFunctionValue(((Number) o).doubleValue(),
null));
+ }
+ }
+ }
+ } catch (QueryProcessException
+ | TException
+ | StorageEngineException
+ | SQLException
+ | IOException
+ | InterruptedException
+ | QueryFilterOptimizationException
+ | MetadataException e) {
+ throw new InfluxDBException(e.getMessage());
+ } finally {
+
ServiceProvider.SESSION_MANAGER.releaseQueryResourceNoExceptions(queryId);
+ }
+ long queryId1 = ServiceProvider.SESSION_MANAGER.requestQueryId(true);
+ try {
+ String functionSqlMinValue =
+ StringUtils.generateFunctionSql("min_value",
function.getParmaName(), path);
+ QueryPlan queryPlan =
+ (QueryPlan)
+
serviceProvider.getPlanner().parseSQLToPhysicalPlan(functionSqlMinValue);
+ QueryContext queryContext =
+ serviceProvider.genQueryContext(
+ queryId,
+ true,
+ System.currentTimeMillis(),
+ functionSqlMinValue,
+ IoTDBConstant.DEFAULT_CONNECTION_TIMEOUT_MS);
+ QueryDataSet queryDataSet =
+ serviceProvider.createQueryDataSet(
+ queryContext, queryPlan, IoTDBConstant.DEFAULT_FETCH_SIZE);
+ while (queryDataSet.hasNext()) {
+ List<Path> paths = queryDataSet.getPaths();
+ List<Field> fields = queryDataSet.next().getFields();
+ for (int i = 0; i < paths.size(); i++) {
+ Object o = FieldUtils.iotdbFieldConvert(fields.get(i));
+ if (o instanceof Number) {
+ function.updateValueIoTDBFunc(
+ null, new InfluxFunctionValue(((Number)
o).doubleValue(), null));
+ }
+ }
+ }
+ } catch (QueryProcessException
+ | TException
+ | StorageEngineException
+ | SQLException
+ | IOException
+ | InterruptedException
+ | QueryFilterOptimizationException
+ | MetadataException e) {
+ throw new InfluxDBException(e.getMessage());
+ } finally {
+
ServiceProvider.SESSION_MANAGER.releaseQueryResourceNoExceptions(queryId1);
+ }
+ break;
+ }
+ case InfluxSQLConstant.SUM:
+ {
+ long queryId = ServiceProvider.SESSION_MANAGER.requestQueryId(true);
+ try {
+ String functionSql =
+ StringUtils.generateFunctionSql("sum",
function.getParmaName(), path);
+ QueryPlan queryPlan =
+ (QueryPlan)
serviceProvider.getPlanner().parseSQLToPhysicalPlan(functionSql);
+ QueryContext queryContext =
+ serviceProvider.genQueryContext(
+ queryId,
+ true,
+ System.currentTimeMillis(),
+ functionSql,
+ IoTDBConstant.DEFAULT_CONNECTION_TIMEOUT_MS);
+ QueryDataSet queryDataSet =
+ serviceProvider.createQueryDataSet(
+ queryContext, queryPlan, IoTDBConstant.DEFAULT_FETCH_SIZE);
+ while (queryDataSet.hasNext()) {
+ List<Field> fields = queryDataSet.next().getFields();
+ if (fields.get(1).getDataType() != null) {
+ function.updateValueIoTDBFunc(
+ new InfluxFunctionValue(fields.get(1).getDoubleV(), null));
+ }
+ }
+ } catch (QueryProcessException
+ | TException
+ | StorageEngineException
+ | SQLException
+ | IOException
+ | InterruptedException
+ | QueryFilterOptimizationException
+ | MetadataException e) {
+ throw new InfluxDBException(e.getMessage());
+ } finally {
+
ServiceProvider.SESSION_MANAGER.releaseQueryResourceNoExceptions(queryId);
+ }
+ break;
+ }
+ case InfluxSQLConstant.FIRST:
+ case InfluxSQLConstant.LAST:
+ {
+ String functionSql;
+ if (function.getFunctionName().equals(InfluxSQLConstant.FIRST)) {
+ functionSql =
+ StringUtils.generateFunctionSql("first_value",
function.getParmaName(), path);
+ } else {
+ functionSql =
+ StringUtils.generateFunctionSql("last_value",
function.getParmaName(), path);
+ }
+ List<Long> queryIds = new ArrayList<>();
+ queryIds.add(ServiceProvider.SESSION_MANAGER.requestQueryId(true));
+ try {
+ QueryPlan queryPlan =
+ (QueryPlan)
serviceProvider.getPlanner().parseSQLToPhysicalPlan(functionSql);
+ QueryContext queryContext =
+ serviceProvider.genQueryContext(
+ queryIds.get(0),
+ true,
+ System.currentTimeMillis(),
+ functionSql,
+ IoTDBConstant.DEFAULT_CONNECTION_TIMEOUT_MS);
+ QueryDataSet queryDataSet =
+ serviceProvider.createQueryDataSet(
+ queryContext, queryPlan, IoTDBConstant.DEFAULT_FETCH_SIZE);
+ while (queryDataSet.hasNext()) {
+ List<Path> paths = queryDataSet.getPaths();
+ List<Field> fields = queryDataSet.next().getFields();
+ for (int i = 0; i < paths.size(); i++) {
+ Object o = FieldUtils.iotdbFieldConvert(fields.get(i));
+ long queryId =
ServiceProvider.SESSION_MANAGER.requestQueryId(true);
+ queryIds.add(queryId);
+ if (o != null) {
+ String specificSql =
+ String.format(
+ "select %s from %s where %s=%s",
+ function.getParmaName(),
+ paths.get(i).getDevice(),
+ paths.get(i).getFullPath(),
+ o);
+ QueryPlan queryPlanNew =
+ (QueryPlan)
serviceProvider.getPlanner().parseSQLToPhysicalPlan(specificSql);
+ QueryContext queryContextNew =
+ serviceProvider.genQueryContext(
+ queryId,
+ true,
+ System.currentTimeMillis(),
+ specificSql,
+ IoTDBConstant.DEFAULT_CONNECTION_TIMEOUT_MS);
+ QueryDataSet queryDataSetNew =
+ serviceProvider.createQueryDataSet(
+ queryContextNew, queryPlanNew,
IoTDBConstant.DEFAULT_FETCH_SIZE);
+ while (queryDataSetNew.hasNext()) {
+ RowRecord recordNew = queryDataSetNew.next();
+ List<Field> newFields = recordNew.getFields();
+ long time = recordNew.getTimestamp();
+ function.updateValueIoTDBFunc(new
InfluxFunctionValue(newFields.get(0), time));
+ }
+ }
+ }
+ }
+ } catch (QueryProcessException
+ | TException
+ | StorageEngineException
+ | SQLException
+ | IOException
+ | InterruptedException
+ | QueryFilterOptimizationException
+ | MetadataException e) {
+ throw new InfluxDBException(e.getMessage());
+ } finally {
+ for (long queryId : queryIds) {
+
ServiceProvider.SESSION_MANAGER.releaseQueryResourceNoExceptions(queryId);
+ }
+ }
+ break;
+ }
+ case InfluxSQLConstant.MAX:
+ case InfluxSQLConstant.MIN:
+ {
+ String functionSql;
+ if (function.getFunctionName().equals(InfluxSQLConstant.MAX)) {
+ functionSql =
+ StringUtils.generateFunctionSql("max_value",
function.getParmaName(), path);
+ } else {
+ functionSql =
+ StringUtils.generateFunctionSql("min_value",
function.getParmaName(), path);
+ }
+ long queryId = ServiceProvider.SESSION_MANAGER.requestQueryId(true);
+ try {
+ QueryPlan queryPlan =
+ (QueryPlan)
serviceProvider.getPlanner().parseSQLToPhysicalPlan(functionSql);
+ QueryContext queryContext =
+ serviceProvider.genQueryContext(
+ queryId,
+ true,
+ System.currentTimeMillis(),
+ functionSql,
+ IoTDBConstant.DEFAULT_CONNECTION_TIMEOUT_MS);
+ QueryDataSet queryDataSet =
+ serviceProvider.createQueryDataSet(
+ queryContext, queryPlan, IoTDBConstant.DEFAULT_FETCH_SIZE);
+ while (queryDataSet.hasNext()) {
+ List<Path> paths = queryDataSet.getPaths();
+ List<Field> fields = queryDataSet.next().getFields();
+ for (int i = 0; i < paths.size(); i++) {
+ Object o = FieldUtils.iotdbFieldConvert(fields.get(i));
+ function.updateValueIoTDBFunc(new InfluxFunctionValue(o,
null));
+ }
+ }
+ } catch (QueryProcessException
+ | TException
+ | StorageEngineException
+ | SQLException
+ | IOException
+ | InterruptedException
+ | QueryFilterOptimizationException
+ | MetadataException e) {
+ throw new InfluxDBException(e.getMessage());
+ } finally {
+
ServiceProvider.SESSION_MANAGER.releaseQueryResourceNoExceptions(queryId);
+ }
+ break;
+ }
+ default:
+ throw new IllegalStateException("Unexpected value: " +
function.getFunctionName());
+ }
+ return function.calculateByIoTDBFunc();
+ }
+
+ public static void checkInfluxDBQueryOperator(Operator operator) {
+ if (!(operator instanceof InfluxQueryOperator)) {
+ throw new IllegalArgumentException("not query sql");
+ }
+ InfluxSelectComponent selectComponent = ((InfluxQueryOperator)
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");
+ }
+ }
+
+ public static QueryResult queryExpr(
+ FilterOperator operator,
+ String database,
+ String measurement,
+ ServiceProvider serviceProvider,
+ Map<String, Integer> fieldOrders,
+ Long sessionId)
+ throws AuthException {
+ if (operator == null) {
+ List<IExpression> expressions = new ArrayList<>();
+ return queryByConditions(
+ expressions, database, measurement, serviceProvider, fieldOrders,
sessionId);
+ } else if (operator instanceof BasicFunctionOperator) {
+ List<IExpression> iExpressions = new ArrayList<>();
+
iExpressions.add(getIExpressionForBasicFunctionOperator((BasicFunctionOperator)
operator));
+ return queryByConditions(
+ iExpressions, database, measurement, serviceProvider, fieldOrders,
sessionId);
+ } else {
+ FilterOperator leftOperator = operator.getChildren().get(0);
+ FilterOperator rightOperator = operator.getChildren().get(1);
+ if (operator.getFilterType() == FilterConstant.FilterType.KW_OR) {
+ return QueryResultUtils.orQueryResultProcess(
+ queryExpr(leftOperator, database, measurement, serviceProvider,
fieldOrders, sessionId),
+ queryExpr(
+ rightOperator, database, measurement, serviceProvider,
fieldOrders, sessionId));
+ } else if (operator.getFilterType() == FilterConstant.FilterType.KW_AND)
{
+ if (canMergeOperator(leftOperator) && canMergeOperator(rightOperator))
{
+ List<IExpression> iExpressions1 =
getIExpressionByFilterOperatorOperator(leftOperator);
+ List<IExpression> iExpressions2 =
getIExpressionByFilterOperatorOperator(rightOperator);
+ iExpressions1.addAll(iExpressions2);
+ return queryByConditions(
+ iExpressions1, database, measurement, serviceProvider,
fieldOrders, sessionId);
+ } else {
+ return QueryResultUtils.andQueryResultProcess(
+ queryExpr(
+ leftOperator, database, measurement, serviceProvider,
fieldOrders, sessionId),
+ queryExpr(
+ rightOperator, database, measurement, serviceProvider,
fieldOrders, sessionId));
+ }
+ }
+ }
+ throw new IllegalArgumentException("unknown operator " + operator);
+ }
+
+ /**
+ * get query results in the format of influxdb through conditions
+ *
+ * @param expressions list of conditions, including tag and field condition
+ * @return returns the results of the influxdb query
+ */
+ private static QueryResult queryByConditions(
+ List<IExpression> expressions,
+ String database,
+ String measurement,
+ ServiceProvider serviceProvider,
+ Map<String, Integer> fieldOrders,
+ Long sessionId)
+ throws AuthException {
+ // used to store the actual order according to the tag
+ Map<Integer, SingleSeriesExpression> realTagOrders = new HashMap<>();
+ // stores a list of conditions belonging to the field
+ List<SingleSeriesExpression> fieldExpressions = new ArrayList<>();
+ // maximum number of tags in the current query criteria
+ int currentQueryMaxTagNum = 0;
+ Map<String, Integer> tagOrders =
+
InfluxDBMetaManager.getDatabase2Measurement2TagOrders().get(database).get(measurement);
+ for (IExpression expression : expressions) {
+ SingleSeriesExpression singleSeriesExpression =
((SingleSeriesExpression) expression);
+ // the current condition is in tag
+ if
(tagOrders.containsKey(singleSeriesExpression.getSeriesPath().getFullPath())) {
+ int curOrder =
tagOrders.get(singleSeriesExpression.getSeriesPath().getFullPath());
+ // put it into the map according to the tag
+ realTagOrders.put(curOrder, singleSeriesExpression);
+ // update the maximum tag order of the current query criteria
+ currentQueryMaxTagNum = Math.max(currentQueryMaxTagNum, curOrder);
+ } else {
+ fieldExpressions.add(singleSeriesExpression);
+ }
+ }
+ // 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(
+ StringUtils.removeQuotation(
+
FilterUtils.getFilterStringValue(realTagOrders.get(i).getFilter())));
+ } else {
+ curQueryPath.append(".").append("*");
+ }
+ }
+ curQueryPath.append(".**");
+ // construct actual query condition
+ StringBuilder realIotDBCondition = new StringBuilder();
+ for (int i = 0; i < fieldExpressions.size(); i++) {
+ SingleSeriesExpression singleSeriesExpression = fieldExpressions.get(i);
+ if (i != 0) {
+ realIotDBCondition.append(" and ");
+ }
+ realIotDBCondition
+ .append(singleSeriesExpression.getSeriesPath().getFullPath())
+ .append(" ")
+
.append((FilterUtils.getFilerSymbol(singleSeriesExpression.getFilter())))
+ .append(" ")
+
.append(FilterUtils.getFilterStringValue(singleSeriesExpression.getFilter()));
+ }
+ // actual query SQL statement
+ String realQuerySql;
+
+ realQuerySql = "select * from " + curQueryPath;
+ if (!(realIotDBCondition.length() == 0)) {
+ realQuerySql += " where " + realIotDBCondition;
+ }
+ realQuerySql += " align by device";
+
+ long queryId = ServiceProvider.SESSION_MANAGER.requestQueryId(true);
+ try {
+ QueryPlan queryPlan =
+ (QueryPlan)
serviceProvider.getPlanner().parseSQLToPhysicalPlan(realQuerySql);
+ TSStatus tsStatus = serviceProvider.checkAuthority(queryPlan, sessionId);
+ if (tsStatus != null) {
+ throw new AuthException(tsStatus.getMessage());
+ }
+ QueryContext queryContext =
+ serviceProvider.genQueryContext(
+ queryId,
+ true,
+ System.currentTimeMillis(),
+ realQuerySql,
+ IoTDBConstant.DEFAULT_CONNECTION_TIMEOUT_MS);
+ QueryDataSet queryDataSet =
+ serviceProvider.createQueryDataSet(
+ queryContext, queryPlan, IoTDBConstant.DEFAULT_FETCH_SIZE);
+ return QueryResultUtils.iotdbResultConvertInfluxResult(
+ queryDataSet, database, measurement, fieldOrders);
+ } catch (QueryProcessException
+ | TException
+ | StorageEngineException
+ | SQLException
+ | IOException
+ | InterruptedException
+ | QueryFilterOptimizationException
+ | MetadataException e) {
+ throw new InfluxDBException(e.getMessage());
+ } finally {
+
ServiceProvider.SESSION_MANAGER.releaseQueryResourceNoExceptions(queryId);
+ }
+ }
+
+ /**
+ * 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<IExpression> getIExpressionByFilterOperatorOperator(
+ FilterOperator filterOperator) {
+ if (filterOperator instanceof BasicFunctionOperator) {
+ // It must be a non-or situation
+ List<IExpression> expressions = new ArrayList<>();
+ expressions.add(
+ getIExpressionForBasicFunctionOperator((BasicFunctionOperator)
filterOperator));
+ return expressions;
+ } else {
+ FilterOperator leftOperator = filterOperator.getChildren().get(0);
+ FilterOperator rightOperator = filterOperator.getChildren().get(1);
+ List<IExpression> expressions1 =
getIExpressionByFilterOperatorOperator(leftOperator);
+ List<IExpression> expressions2 =
getIExpressionByFilterOperatorOperator(rightOperator);
+ expressions1.addAll(expressions2);
+ return expressions1;
+ }
+ }
+
+ /**
+ * 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.getChildren().get(0);
+ FilterOperator rightOperator = operator.getChildren().get(1);
+ return canMergeOperator(leftOperator) &&
canMergeOperator(rightOperator);
+ }
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/meta/InfluxDBMetaManager.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/meta/InfluxDBMetaManager.java
index ff20c64..261a0e9 100644
---
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/meta/InfluxDBMetaManager.java
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/meta/InfluxDBMetaManager.java
@@ -25,7 +25,7 @@ import
org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.path.PartialPath;
-import org.apache.iotdb.db.protocol.influxdb.constant.InfluxDBConstant;
+import org.apache.iotdb.db.protocol.influxdb.constant.InfluxConstant;
import org.apache.iotdb.db.qp.Planner;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
@@ -188,7 +188,7 @@ public class InfluxDBMetaManager {
.append(
layerOrderToTagKeysInPath.containsKey(i)
? tags.get(layerOrderToTagKeysInPath.get(i))
- : InfluxDBConstant.PLACE_HOLDER);
+ : InfluxConstant.PLACE_HOLDER);
}
return path.toString();
}
@@ -204,6 +204,10 @@ public class InfluxDBMetaManager {
}
}
+ public static Map<String, Map<String, Map<String, Integer>>>
getDatabase2Measurement2TagOrders() {
+ return database2Measurement2TagOrders;
+ }
+
private static class InfluxDBMetaManagerHolder {
private static final InfluxDBMetaManager INSTANCE = new
InfluxDBMetaManager();
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/operator/InfluxSelectComponent.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/operator/InfluxSelectComponent.java
index 4e98caa..b5b9113 100644
---
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/operator/InfluxSelectComponent.java
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/operator/InfluxSelectComponent.java
@@ -36,6 +36,7 @@ public final class InfluxSelectComponent
private boolean hasMoreFunction = false;
private boolean hasFunction = false;
private boolean hasCommonQuery = false;
+ private boolean hasOnlyTraverseFunction = false;
public InfluxSelectComponent() {
super((ZoneId) null);
@@ -61,6 +62,9 @@ public final class InfluxSelectComponent
} else {
hasAggregationFunction = true;
}
+ if
(InfluxSQLConstant.getOnlyTraverseFunctionNames().contains(functionName.toLowerCase()))
{
+ hasOnlyTraverseFunction = true;
+ }
}
if (expression instanceof TimeSeriesOperand) {
hasCommonQuery = true;
@@ -80,8 +84,8 @@ public final class InfluxSelectComponent
return hasCommonQuery;
}
- public boolean isHasSelectorFunction() {
- return hasSelectorFunction;
+ public boolean isHasOnlyTraverseFunction() {
+ return hasOnlyTraverseFunction;
}
public boolean isHasMoreSelectorFunction() {
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxDBConstant.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/util/CommonUtils.java
similarity index 61%
copy from
server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxDBConstant.java
copy to
server/src/main/java/org/apache/iotdb/db/protocol/influxdb/util/CommonUtils.java
index 6b59698..fcdc7cd 100644
---
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxDBConstant.java
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/util/CommonUtils.java
@@ -16,9 +16,25 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.protocol.influxdb.constant;
+package org.apache.iotdb.db.protocol.influxdb.util;
-public class InfluxDBConstant {
-
- public static final String PLACE_HOLDER = "PH";
+public class CommonUtils {
+ /**
+ * 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);
+ }
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxDBConstant.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/util/FieldUtils.java
similarity index 50%
rename from
server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxDBConstant.java
rename to
server/src/main/java/org/apache/iotdb/db/protocol/influxdb/util/FieldUtils.java
index 6b59698..9b49e87 100644
---
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/constant/InfluxDBConstant.java
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/util/FieldUtils.java
@@ -16,9 +16,36 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.protocol.influxdb.constant;
+package org.apache.iotdb.db.protocol.influxdb.util;
-public class InfluxDBConstant {
+import org.apache.iotdb.tsfile.read.common.Field;
- public static final String PLACE_HOLDER = "PH";
+public class FieldUtils {
+ /**
+ * convert the value of field in iotdb to object
+ *
+ * @param field filed to be converted
+ * @return value stored in field
+ */
+ public static Object iotdbFieldConvert(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;
+ }
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/util/FilterUtils.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/util/FilterUtils.java
new file mode 100644
index 0000000..8bad260
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/util/FilterUtils.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.db.protocol.influxdb.util;
+
+import org.apache.iotdb.db.qp.constant.FilterConstant;
+import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
+import org.apache.iotdb.tsfile.read.filter.ValueFilter;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+
+public class FilterUtils {
+
+ public static String getFilterStringValue(Filter filter) {
+ String filterString = filter.toString();
+ if (filter instanceof ValueFilter.ValueEq) {
+ return filterString.split("== ")[1];
+ } else if (filter instanceof ValueFilter.ValueNotEq) {
+ return filterString.split("!= ")[1];
+ } else if (filter instanceof ValueFilter.ValueLtEq) {
+ return filterString.split("<= ")[1];
+ } else if (filter instanceof ValueFilter.ValueLt) {
+ return filterString.split("< ")[1];
+ } else if (filter instanceof ValueFilter.ValueGtEq) {
+ return filterString.split(">= ")[1];
+ } else if (filter instanceof ValueFilter.ValueGt) {
+ return filterString.split("> ")[1];
+ } else {
+ throw new UnSupportedDataTypeException("Unsupported filter :" + filter);
+ }
+ }
+
+ public static String getFilerSymbol(Filter filter) {
+ if (filter instanceof ValueFilter.ValueEq) {
+ return "=";
+ } else if (filter instanceof ValueFilter.ValueNotEq) {
+ return "!=";
+ } else if (filter instanceof ValueFilter.ValueLtEq) {
+ return "<=";
+ } else if (filter instanceof ValueFilter.ValueLt) {
+ return "<";
+ } else if (filter instanceof ValueFilter.ValueGtEq) {
+ return ">=";
+ } else if (filter instanceof ValueFilter.ValueGt) {
+ return ">";
+ } else {
+ throw new UnSupportedDataTypeException("Unsupported filter :" + filter);
+ }
+ }
+
+ public static Filter filterTypeToFilter(FilterConstant.FilterType
filterType, String value) {
+ switch (filterType) {
+ case EQUAL:
+ return ValueFilter.eq(value);
+ case NOTEQUAL:
+ return ValueFilter.notEq(value);
+ case LESSTHANOREQUALTO:
+ return ValueFilter.ltEq(value);
+ case LESSTHAN:
+ return ValueFilter.lt(value);
+ case GREATERTHANOREQUALTO:
+ return ValueFilter.gtEq(value);
+ case GREATERTHAN:
+ return ValueFilter.gt(value);
+ default:
+ throw new UnSupportedDataTypeException("Unsupported data type:" +
filterType);
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/util/QueryResultUtils.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/util/QueryResultUtils.java
new file mode 100644
index 0000000..df4c431
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/util/QueryResultUtils.java
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.protocol.influxdb.util;
+
+import org.apache.iotdb.db.protocol.influxdb.constant.InfluxConstant;
+import org.apache.iotdb.db.protocol.influxdb.meta.InfluxDBMetaManager;
+import org.apache.iotdb.db.query.dataset.AlignByDeviceDataSet;
+import org.apache.iotdb.tsfile.read.common.Field;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+
+import org.influxdb.dto.QueryResult;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class QueryResultUtils {
+ /**
+ * 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);
+ }
+
+ /**
+ * Convert align by device query result of iotdb to the query result of
influxdb
+ *
+ * @param queryDataSet iotdb query results to be converted
+ * @return query results in influxdb format
+ */
+ public static QueryResult iotdbResultConvertInfluxResult(
+ QueryDataSet queryDataSet,
+ String database,
+ String measurement,
+ Map<String, Integer> fieldOrders)
+ throws IOException {
+
+ if (queryDataSet == null) {
+ return getNullQueryResult();
+ }
+ // generate series
+ QueryResult.Series series = new QueryResult.Series();
+ series.setName(measurement);
+ // gets the reverse map of the tag
+ Map<String, Integer> tagOrders =
+
InfluxDBMetaManager.getDatabase2Measurement2TagOrders().get(database).get(measurement);
+ Map<Integer, String> tagOrderReversed =
+ tagOrders.entrySet().stream()
+ .collect(Collectors.toMap(Map.Entry::getValue, Map.Entry::getKey));
+ Map<Integer, String> fieldOrdersReversed =
+ fieldOrders.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<>();
+
+ while (queryDataSet.hasNext()) {
+ Object[] value = new Object[columns.size()];
+
+ RowRecord record = queryDataSet.next();
+ List<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(InfluxConstant.PLACE_HOLDER)) {
+ value[i - 2] = deviceNameList[i];
+ }
+ }
+ for (int i = 1; i < fields.size(); i++) {
+ Object o = FieldUtils.iotdbFieldConvert(fields.get(i));
+ if (o != null) {
+ // insert the value of filed into it
+ value[
+ fieldOrders.get(
+ ((AlignByDeviceDataSet)
queryDataSet).getMeasurements().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;
+ }
+
+ /**
+ * 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;
+ }
+
+ /**
+ * 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);
+ }
+
+ /**
+ * 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())
+ && StringUtils.checkSameStringList(
+ queryResult1.getResults().get(0).getSeries().get(0).getColumns(),
+ queryResult2.getResults().get(0).getSeries().get(0).getColumns());
+ }
+
+ /**
+ * 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 (!CommonUtils.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;
+ }
+
+ /**
+ * 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 (!CommonUtils.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;
+ }
+
+ /**
+ * 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;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/util/StringUtils.java
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/util/StringUtils.java
new file mode 100644
index 0000000..39449b6
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/protocol/influxdb/util/StringUtils.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.protocol.influxdb.util;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class StringUtils {
+
+ /**
+ * 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;
+ }
+
+ /**
+ * 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;
+ }
+
+ /**
+ * get the last node through the path in iotdb
+ *
+ * @param path path to process
+ * @return last node
+ */
+ public static String getFieldByPath(String path) {
+ String[] tmpList = path.split("\\.");
+ return tmpList[tmpList.length - 1];
+ }
+
+ /**
+ * 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
+ */
+ public 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;
+ }
+
+ public static String generateFunctionSql(String functionName, String
parameter, String path) {
+ return String.format("select %s(%s) from %s.**", functionName, parameter,
path);
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java
b/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java
index a2ab715..cd38f79 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java
@@ -134,6 +134,10 @@ public class AlignByDeviceDataSet extends QueryDataSet {
return pathsNum;
}
+ public List<String> getMeasurements() {
+ return measurements;
+ }
+
@Override
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity
warning
public boolean hasNextWithoutConstraint() throws IOException {
diff --git
a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/InfluxDBServiceImpl.java
b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/InfluxDBServiceImpl.java
index 9b97cf6..8789142 100644
---
a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/InfluxDBServiceImpl.java
+++
b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/InfluxDBServiceImpl.java
@@ -25,8 +25,12 @@ import
org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.protocol.influxdb.dto.IoTDBPoint;
+import org.apache.iotdb.db.protocol.influxdb.handler.QueryHandler;
import org.apache.iotdb.db.protocol.influxdb.input.InfluxLineParser;
import org.apache.iotdb.db.protocol.influxdb.meta.InfluxDBMetaManager;
+import org.apache.iotdb.db.protocol.influxdb.operator.InfluxQueryOperator;
+import org.apache.iotdb.db.protocol.influxdb.sql.InfluxDBLogicalGenerator;
+import org.apache.iotdb.db.qp.logical.Operator;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
@@ -39,6 +43,8 @@ import
org.apache.iotdb.protocol.influxdb.rpc.thrift.TSCloseSessionReq;
import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSCreateDatabaseReq;
import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSOpenSessionReq;
import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSOpenSessionResp;
+import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSQueryReq;
+import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSQueryResultRsp;
import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSStatus;
import org.apache.iotdb.protocol.influxdb.rpc.thrift.TSWritePointsReq;
import org.apache.iotdb.rpc.IoTDBConnectionException;
@@ -135,6 +141,14 @@ public class InfluxDBServiceImpl implements
InfluxDBService.Iface {
}
}
+ @Override
+ public TSQueryResultRsp query(TSQueryReq req) throws TException {
+ Operator operator = InfluxDBLogicalGenerator.generate(req.command);
+ QueryHandler.checkInfluxDBQueryOperator(operator);
+ return QueryHandler.queryInfluxDB(
+ req.database, (InfluxQueryOperator) operator, req.sessionId,
serviceProvider);
+ }
+
public void handleClientExit() {
Long sessionId = ServiceProvider.SESSION_MANAGER.getCurrSessionId();
if (sessionId != null) {
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/MathUtils.java
b/server/src/main/java/org/apache/iotdb/db/utils/MathUtils.java
index e5809d5..aceec18 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/MathUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/MathUtils.java
@@ -21,6 +21,8 @@ package org.apache.iotdb.db.utils;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+import java.util.List;
+
public class MathUtils {
private MathUtils() {
@@ -77,4 +79,53 @@ public class MathUtils {
* Math.pow(10,
TSFileDescriptor.getInstance().getConfig().getFloatPrecision())))
/ Math.pow(10,
TSFileDescriptor.getInstance().getConfig().getFloatPrecision());
}
+
+ /**
+ * 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 (Double datum : data) {
+ sum = sum + datum;
+ }
+ return sum;
+ }
+
+ /**
+ * calculate mean of list
+ *
+ * @param data need to calculate list
+ * @return mean of list
+ */
+ public static double mean(List<Double> data) {
+ return sum(data) / data.size();
+ }
+
+ /**
+ * calculate pop variance of list
+ *
+ * @param data need to calculate list
+ * @return pop variance of list
+ */
+ public static double popVariance(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 popStdDev(List<Double> data) {
+ return Math.sqrt(popVariance(data));
+ }
}
diff --git a/thrift-influxdb/src/main/thrift/influxdb.thrift
b/thrift-influxdb/src/main/thrift/influxdb.thrift
index 4ab0a19..a62cd91 100644
--- a/thrift-influxdb/src/main/thrift/influxdb.thrift
+++ b/thrift-influxdb/src/main/thrift/influxdb.thrift
@@ -81,6 +81,23 @@ struct TSCreateDatabaseReq{
2: required string database
}
+// query()
+// query in influxdb
+struct TSQueryReq{
+ // The session to execute the statement against
+ 1: required i64 sessionId
+
+ 2: required string command
+ 3: required string database
+}
+
+struct TSQueryResultRsp{
+ 1: required TSStatus status
+
+ 2: optional string resultJsonString
+
+}
+
service InfluxDBService {
TSOpenSessionResp openSession(1:TSOpenSessionReq req);
@@ -89,4 +106,6 @@ service InfluxDBService {
TSStatus writePoints(1:TSWritePointsReq req);
TSStatus createDatabase(1:TSCreateDatabaseReq req);
+
+ TSQueryResultRsp query(1:TSQueryReq req);
}
