You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2019/08/22 07:20:21 UTC
[incubator-iotdb] branch rpc_client created (now ceaa9b1)
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a change to branch rpc_client
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git.
at ceaa9b1 add batch rpc interface
This branch includes the following new commits:
new ceaa9b1 add batch rpc interface
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-iotdb] 01/01: add batch rpc interface
Posted by qi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch rpc_client
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit ceaa9b118effb33d1c9b6f917cfb4b83f27b6243
Author: qiaojialin <64...@qq.com>
AuthorDate: Thu Aug 22 15:20:01 2019 +0800
add batch rpc interface
---
.../apache/iotdb/cli/client/AbstractClient.java | 190 ++++++++---------
.../org/apache/iotdb/session/IoTDBRowBatch.java | 135 ++++++++++++
.../iotdb/session/IoTDBSessionException.java | 18 ++
.../java/org/apache/iotdb/session/Session.java | 166 +++++++++++++++
.../{7-TsFile => 8-TsFile}/1-Installation.md | 2 +-
.../UserGuide/{7-TsFile => 8-TsFile}/2-Usage.md | 2 +-
.../{7-TsFile => 8-TsFile}/3-Hierarchy.md | 2 +-
.../{8-System Tools => 9-System Tools}/1-Sync.md | 2 +-
.../2-Memory Estimation Tool.md | 2 +-
.../UserGuide/{8-Tools-Cli.md => 9-Tools-Cli.md} | 0
.../{8-Tools-Grafana.md => 9-Tools-Grafana.md} | 0
.../{8-Tools-Hadoop.md => 9-Tools-Hadoop.md} | 0
.../{8-Tools-spark.md => 9-Tools-spark.md} | 0
docs/Documentation/Frequently asked questions.md | 4 +-
.../UserGuide/3-Operation Manual/4-Data Query.md | 16 +-
.../1-IoTDB Query Statement.md | 2 +-
.../UserGuide/7-Client API/1-Client API.md | 92 ++++++++
.../{7-TsFile => 8-TsFile}/1-Installation.md | 2 +-
.../UserGuide/{7-TsFile => 8-TsFile}/2-Usage.md | 2 +-
.../{7-TsFile => 8-TsFile}/3-Hierarchy.md | 2 +-
.../{8-System Tools => 9-System Tools}/1-Sync.md | 2 +-
.../2-Memory Estimation Tool.md | 2 +-
.../UserGuide/{8-Tools-Cli.md => 9-Tools-Cli.md} | 2 +-
.../{8-Tools-Grafana.md => 9-Tools-Grafana.md} | 0
.../{8-Tools-Hadoop.md => 9-Tools-Hadoop.md} | 0
.../{8-Tools-spark.md => 9-Tools-spark.md} | 0
example/client/pom.xml | 46 ++++
.../org/apache/iotdb/client/SessionExample.java | 35 ++++
.../java/org/apache/iotdb/client/SessionTest.java | 77 +++++++
.../org/apache/iotdb/client/StatementDemo.java | 67 ++++++
example/pom.xml | 2 +
.../iotdb/{example => rocketmq}/Constant.java | 2 +-
.../{example => rocketmq}/RocketMQConsumer.java | 2 +-
.../{example => rocketmq}/RocketMQProducer.java | 2 +-
.../apache/iotdb/{example => rocketmq}/Utils.java | 2 +-
.../org/apache/iotdb/jdbc/IoTDBConnection.java | 17 +-
.../jdbc/IoTDBPreparedInsertionStatement.java | 2 +-
.../apache/iotdb/jdbc/IoTDBPreparedStatement.java | 1 +
.../src/main/java/org/apache/iotdb/jdbc/Utils.java | 33 ++-
.../apache/iotdb/jdbc/IoTDBQueryResultSetTest.java | 4 +-
.../test/java/org/apache/iotdb/jdbc/UtilsTest.java | 7 +-
.../org/apache/iotdb/db/conf/IoTDBConstant.java | 2 +-
.../org/apache/iotdb/db/engine/StorageEngine.java | 22 +-
.../iotdb/db/engine/memtable/AbstractMemTable.java | 27 ++-
.../apache/iotdb/db/engine/memtable/IMemTable.java | 8 +-
.../db/engine/memtable/IWritableMemChunk.java | 4 +-
.../iotdb/db/engine/memtable/WritableMemChunk.java | 27 ++-
.../engine/storagegroup/StorageGroupProcessor.java | 128 ++++++++---
.../db/engine/storagegroup/TsFileProcessor.java | 55 ++++-
.../java/org/apache/iotdb/db/metadata/MGraph.java | 2 -
.../db/qp/executor/IQueryProcessExecutor.java | 7 +
.../iotdb/db/qp/executor/QueryProcessExecutor.java | 34 ++-
.../org/apache/iotdb/db/qp/logical/Operator.java | 2 +-
.../apache/iotdb/db/qp/physical/PhysicalPlan.java | 7 +-
.../iotdb/db/qp/physical/crud/BatchInsertPlan.java | 233 +++++++++++++++++++++
.../iotdb/db/qp/strategy/LogicalGenerator.java | 4 -
.../iotdb/db/qp/strategy/PhysicalGenerator.java | 6 -
.../org/apache/iotdb/db/rescon/MemTablePool.java | 1 +
.../java/org/apache/iotdb/db/service/IoTDB.java | 4 -
.../org/apache/iotdb/db/service/TSServiceImpl.java | 61 +++++-
.../java/org/apache/iotdb/db/utils/MemUtils.java | 28 +++
.../apache/iotdb/db/utils/QueryDataSetUtils.java | 28 ++-
.../storagegroup/StorageGroupProcessorTest.java | 65 ++++++
.../apache/iotdb/db/qp/utils/MemIntQpExecutor.java | 6 +
service-rpc/src/main/thrift/rpc.thrift | 36 +++-
65 files changed, 1521 insertions(+), 220 deletions(-)
diff --git a/client/src/main/java/org/apache/iotdb/cli/client/AbstractClient.java b/client/src/main/java/org/apache/iotdb/cli/client/AbstractClient.java
index 3cbf452..a5f9af2 100644
--- a/client/src/main/java/org/apache/iotdb/cli/client/AbstractClient.java
+++ b/client/src/main/java/org/apache/iotdb/cli/client/AbstractClient.java
@@ -48,55 +48,55 @@ import org.apache.iotdb.jdbc.IoTDBSQLException;
import org.apache.iotdb.service.rpc.thrift.ServerProperties;
import org.apache.thrift.TException;
-public abstract class AbstractClient {
-
- protected static final String HOST_ARGS = "h";
- protected static final String HOST_NAME = "host";
-
- protected static final String HELP_ARGS = "help";
-
- protected static final String PORT_ARGS = "p";
- protected static final String PORT_NAME = "port";
-
- protected static final String PASSWORD_ARGS = "pw";
- protected static final String PASSWORD_NAME = "password";
-
- protected static final String USERNAME_ARGS = "u";
- protected static final String USERNAME_NAME = "username";
-
- protected static final String EXECUTE_ARGS = "e";
- protected static final String EXECUTE_NAME = "execute";
-
- protected static final String ISO8601_ARGS = "disableISO8601";
- protected static final List<String> AGGREGRATE_TIME_LIST = new ArrayList<>();
- protected static final String MAX_PRINT_ROW_COUNT_ARGS = "maxPRC";
- protected static final String MAX_PRINT_ROW_COUNT_NAME = "maxPrintRowCount";
- protected static final String SET_MAX_DISPLAY_NUM = "set max_display_num";
- protected static final String SET_TIMESTAMP_DISPLAY = "set time_display_type";
- protected static final String SHOW_TIMESTAMP_DISPLAY = "show time_display_type";
- protected static final String SET_TIME_ZONE = "set time_zone";
- protected static final String SHOW_TIMEZONE = "show time_zone";
- protected static final String SET_FETCH_SIZE = "set fetch_size";
- protected static final String SHOW_FETCH_SIZE = "show fetch_size";
- protected static final String HELP = "help";
- protected static final String IOTDB_CLI_PREFIX = "IoTDB";
- protected static final String SCRIPT_HINT = "./start-client.sh(start-client.bat if Windows)";
- protected static final String QUIT_COMMAND = "quit";
- protected static final String EXIT_COMMAND = "exit";
- protected static final String SHOW_METADATA_COMMAND = "show timeseries";
- protected static final int MAX_HELP_CONSOLE_WIDTH = 88;
- protected static final String TIMESTAMP_STR = "Time";
- protected static final int ISO_DATETIME_LEN = 35;
- protected static final String IMPORT_CMD = "import";
+abstract class AbstractClient {
+
+ static final String HOST_ARGS = "h";
+ static final String HOST_NAME = "host";
+
+ static final String HELP_ARGS = "help";
+
+ static final String PORT_ARGS = "p";
+ static final String PORT_NAME = "port";
+
+ static final String PASSWORD_ARGS = "pw";
+ private static final String PASSWORD_NAME = "password";
+
+ static final String USERNAME_ARGS = "u";
+ static final String USERNAME_NAME = "username";
+
+ private static final String EXECUTE_ARGS = "e";
+ private static final String EXECUTE_NAME = "execute";
+
+ static final String ISO8601_ARGS = "disableISO8601";
+ static final List<String> AGGREGRATE_TIME_LIST = new ArrayList<>();
+ static final String MAX_PRINT_ROW_COUNT_ARGS = "maxPRC";
+ private static final String MAX_PRINT_ROW_COUNT_NAME = "maxPrintRowCount";
+ static final String SET_MAX_DISPLAY_NUM = "set max_display_num";
+ static final String SET_TIMESTAMP_DISPLAY = "set time_display_type";
+ static final String SHOW_TIMESTAMP_DISPLAY = "show time_display_type";
+ static final String SET_TIME_ZONE = "set time_zone";
+ static final String SHOW_TIMEZONE = "show time_zone";
+ static final String SET_FETCH_SIZE = "set fetch_size";
+ static final String SHOW_FETCH_SIZE = "show fetch_size";
+ private static final String HELP = "help";
+ static final String IOTDB_CLI_PREFIX = "IoTDB";
+ static final String SCRIPT_HINT = "./start-client.sh(start-client.bat if Windows)";
+ static final String QUIT_COMMAND = "quit";
+ static final String EXIT_COMMAND = "exit";
+ static final String SHOW_METADATA_COMMAND = "show timeseries";
+ static final int MAX_HELP_CONSOLE_WIDTH = 88;
+ static final String TIMESTAMP_STR = "Time";
+ static final int ISO_DATETIME_LEN = 35;
+ private static final String IMPORT_CMD = "import";
private static final String NEED_NOT_TO_PRINT_TIMESTAMP = "AGGREGATION";
private static final String DEFAULT_TIME_FORMAT = "default";
- protected static String timeFormat = DEFAULT_TIME_FORMAT;
- protected static int maxPrintRowCount = 1000;
- protected static int fetchSize = 10000;
- protected static int maxTimeLength = ISO_DATETIME_LEN;
- protected static int maxValueLength = 15;
- protected static boolean isQuit = false;
- protected static String TIMESTAMP_PRECISION = "ms";
+ private static String timeFormat = DEFAULT_TIME_FORMAT;
+ static int maxPrintRowCount = 1000;
+ private static int fetchSize = 10000;
+ static int maxTimeLength = ISO_DATETIME_LEN;
+ static int maxValueLength = 15;
+ private static boolean isQuit = false;
+ static String TIMESTAMP_PRECISION = "ms";
/**
* control the width of columns for 'show timeseries path' and 'show storage group'.
@@ -138,22 +138,22 @@ public abstract class AbstractClient {
* </table>
* </p>
*/
- protected static int[] maxValueLengthForShow = new int[]{75, 45, 8, 8};
- protected static String formatTime = "%" + maxTimeLength + "s|";
- protected static String formatValue = "%" + maxValueLength + "s|";
+ private static int[] maxValueLengthForShow = new int[]{75, 45, 8, 8};
+ static String formatTime = "%" + maxTimeLength + "s|";
+ private static String formatValue = "%" + maxValueLength + "s|";
private static final int DIVIDING_LINE_LENGTH = 40;
- protected static String host = "127.0.0.1";
- protected static String port = "6667";
- protected static String username;
- protected static String password;
- protected static String execute;
- protected static boolean hasExecuteSQL = false;
+ static String host = "127.0.0.1";
+ static String port = "6667";
+ static String username;
+ static String password;
+ static String execute;
+ static boolean hasExecuteSQL = false;
- protected static boolean printToConsole = true;
+ private static boolean printToConsole = true;
- protected static Set<String> keywordSet = new HashSet<>();
+ static Set<String> keywordSet = new HashSet<>();
- protected static ServerProperties properties = null;
+ static ServerProperties properties = null;
private static boolean printHeader = false;
private static int displayCnt = 0;
@@ -165,7 +165,7 @@ public abstract class AbstractClient {
*/
private static boolean showException = false;
- protected static void init() {
+ static void init() {
keywordSet.add("-" + HOST_ARGS);
keywordSet.add("-" + HELP_ARGS);
keywordSet.add("-" + PORT_ARGS);
@@ -184,7 +184,7 @@ public abstract class AbstractClient {
* @param zoneId time-zone ID
* @throws SQLException SQLException
*/
- public static void output(ResultSet res, boolean printToConsole, ZoneId zoneId)
+ private static void output(ResultSet res, boolean printToConsole, ZoneId zoneId)
throws SQLException {
int cnt = 0;
boolean printTimestamp = true;
@@ -229,11 +229,11 @@ public abstract class AbstractClient {
printCount(isShow, res, cnt);
}
- protected static String getTimestampPrecision() {
+ private static String getTimestampPrecision() {
return TIMESTAMP_PRECISION;
}
- protected static void printCount(boolean isShow, ResultSet res, int cnt) throws SQLException {
+ private static void printCount(boolean isShow, ResultSet res, int cnt) throws SQLException {
if (isShow) {
int type = res.getType();
if (type == IoTDBMetadataResultSet.MetadataType.STORAGE_GROUP.ordinal()) { // storage group
@@ -247,7 +247,7 @@ public abstract class AbstractClient {
}
}
- protected static void printRow(boolean printTimestamp, int colCount,
+ private static void printRow(boolean printTimestamp, int colCount,
ResultSetMetaData resultSetMetaData, boolean isShow, ResultSet res, ZoneId zoneId)
throws SQLException {
// Output Labels
@@ -263,7 +263,7 @@ public abstract class AbstractClient {
}
}
- protected static void printHeader(boolean printTimestamp, int colCount,
+ private static void printHeader(boolean printTimestamp, int colCount,
ResultSetMetaData resultSetMetaData, boolean isShow) throws SQLException {
if (!printHeader) {
printBlockLine(printTimestamp, colCount, resultSetMetaData, isShow);
@@ -273,7 +273,7 @@ public abstract class AbstractClient {
}
}
- protected static void printShow(int colCount, ResultSet res) throws SQLException {
+ private static void printShow(int colCount, ResultSet res) throws SQLException {
print("|");
for (int i = 1; i <= colCount; i++) {
formatValue = "%" + maxValueLengthForShow[i - 1] + "s|";
@@ -282,7 +282,7 @@ public abstract class AbstractClient {
println();
}
- protected static void printRowData(boolean printTimestamp, ResultSet res, ZoneId zoneId,
+ private static void printRowData(boolean printTimestamp, ResultSet res, ZoneId zoneId,
ResultSetMetaData resultSetMetaData, int colCount)
throws SQLException {
if (displayCnt < maxPrintRowCount) { // NOTE displayCnt only works on queried data results
@@ -298,7 +298,7 @@ public abstract class AbstractClient {
}
}
- protected static void printColumnData(ResultSetMetaData resultSetMetaData, ResultSet res, int i,
+ private static void printColumnData(ResultSetMetaData resultSetMetaData, ResultSet res, int i,
ZoneId zoneId) throws SQLException {
boolean flag = false;
for (String timeStr : AGGREGRATE_TIME_LIST) {
@@ -319,7 +319,7 @@ public abstract class AbstractClient {
}
}
- protected static Options createOptions() {
+ static Options createOptions() {
Options options = new Options();
Option help = new Option(HELP_ARGS, false, "Display help information(optional)");
help.setRequired(false);
@@ -360,7 +360,7 @@ public abstract class AbstractClient {
return options;
}
- public static String parseLongToDateWithPrecision(DateTimeFormatter formatter,
+ private static String parseLongToDateWithPrecision(DateTimeFormatter formatter,
long timestamp, ZoneId zoneid, String timestampPrecision) {
if (timestampPrecision.equals("ms")) {
long integerofDate = timestamp / 1000;
@@ -420,7 +420,7 @@ public abstract class AbstractClient {
}
}
- protected static String checkRequiredArg(String arg, String name, CommandLine commandLine,
+ static String checkRequiredArg(String arg, String name, CommandLine commandLine,
boolean isRequired,
String defaultValue) throws ArgsErrorException {
String str = commandLine.getOptionValue(arg);
@@ -442,7 +442,7 @@ public abstract class AbstractClient {
return str;
}
- protected static void setTimeFormat(String newTimeFormat) {
+ static void setTimeFormat(String newTimeFormat) {
switch (newTimeFormat.trim().toLowerCase()) {
case "long":
case "number":
@@ -475,7 +475,7 @@ public abstract class AbstractClient {
}
}
- protected static void setMaxDisplayNumber(String maxDisplayNum) {
+ static void setMaxDisplayNumber(String maxDisplayNum) {
long tmp = Long.parseLong(maxDisplayNum.trim());
if (tmp > Integer.MAX_VALUE || tmp < 0) {
maxPrintRowCount = Integer.MAX_VALUE;
@@ -484,7 +484,7 @@ public abstract class AbstractClient {
}
}
- protected static void printBlockLine(boolean printTimestamp, int colCount,
+ private static void printBlockLine(boolean printTimestamp, int colCount,
ResultSetMetaData resultSetMetaData,
boolean isShowTs) throws SQLException {
StringBuilder blockLine = new StringBuilder();
@@ -512,7 +512,7 @@ public abstract class AbstractClient {
println(blockLine);
}
- protected static void printName(boolean printTimestamp, int colCount,
+ private static void printName(boolean printTimestamp, int colCount,
ResultSetMetaData resultSetMetaData,
boolean isShowTs) throws SQLException {
print("|");
@@ -533,7 +533,7 @@ public abstract class AbstractClient {
println();
}
- protected static String[] removePasswordArgs(String[] args) {
+ static String[] removePasswordArgs(String[] args) {
int index = -1;
for (int i = 0; i < args.length; i++) {
if (args[i].equals("-" + PASSWORD_ARGS)) {
@@ -548,7 +548,7 @@ public abstract class AbstractClient {
return args;
}
- protected static String[] processExecuteArgs(String[] args) {
+ static String[] processExecuteArgs(String[] args) {
int index = -1;
for (int i = 0; i < args.length; i++) {
if (args[i].equals("-" + EXECUTE_ARGS)) {
@@ -574,7 +574,7 @@ public abstract class AbstractClient {
}
}
- protected static void displayLogo(String version) {
+ static void displayLogo(String version) {
println(" _____ _________ ______ ______ \n"
+ "|_ _| | _ _ ||_ _ `.|_ _ \\ \n"
+ " | | .--.|_/ | | \\_| | | `. \\ | |_) | \n"
@@ -584,13 +584,13 @@ public abstract class AbstractClient {
+ " \n");
}
- protected static void echoStarting(){
+ static void echoStarting(){
println("---------------------");
println("Starting IoTDB Client");
println("---------------------");
}
- protected static OperationResult handleInputCmd(String cmd, IoTDBConnection connection) {
+ static OperationResult handleInputCmd(String cmd, IoTDBConnection connection) {
String specialCmd = cmd.toLowerCase().trim();
if (QUIT_COMMAND.equals(specialCmd) || EXIT_COMMAND.equals(specialCmd)) {
@@ -647,7 +647,7 @@ public abstract class AbstractClient {
return OperationResult.NO_OPER;
}
- protected static void showHelp() {
+ private static void showHelp() {
println(" <your-sql>\t\t\t execute your sql statment");
println(String.format(" %s\t\t show how many timeseries are in iotdb",
SHOW_METADATA_COMMAND));
@@ -665,7 +665,7 @@ public abstract class AbstractClient {
SET_MAX_DISPLAY_NUM));
}
- protected static void showMetaData(IoTDBConnection connection) {
+ private static void showMetaData(IoTDBConnection connection) {
try {
println(((IoTDBDatabaseMetadata) connection.getMetaData()).getMetadataInJson());
} catch (SQLException e) {
@@ -674,7 +674,7 @@ public abstract class AbstractClient {
}
}
- protected static void setTimestampDisplay(String specialCmd, String cmd) {
+ private static void setTimestampDisplay(String specialCmd, String cmd) {
String[] values = specialCmd.split("=");
if (values.length != 2) {
println(String.format("Time display format error, please input like %s=ISO8601",
@@ -691,7 +691,7 @@ public abstract class AbstractClient {
println("Time display type has set to " + cmd.split("=")[1].trim());
}
- protected static void setTimeZone(String specialCmd, String cmd, IoTDBConnection connection) {
+ private static void setTimeZone(String specialCmd, String cmd, IoTDBConnection connection) {
String[] values = specialCmd.split("=");
if (values.length != 2) {
println(
@@ -708,7 +708,7 @@ public abstract class AbstractClient {
println("Time zone has set to " + values[1].trim());
}
- protected static void setFetchSize(String specialCmd, String cmd) {
+ private static void setFetchSize(String specialCmd, String cmd) {
String[] values = specialCmd.split("=");
if (values.length != 2) {
println(String
@@ -725,7 +725,7 @@ public abstract class AbstractClient {
println("Fetch size has set to " + values[1].trim());
}
- protected static void setMaxDisplaNum(String specialCmd, String cmd) {
+ private static void setMaxDisplaNum(String specialCmd, String cmd) {
String[] values = specialCmd.split("=");
if (values.length != 2) {
println(String.format("Max display number format error, please input like %s = 10000",
@@ -742,7 +742,7 @@ public abstract class AbstractClient {
println("Max display number has set to " + values[1].trim());
}
- protected static void showTimeZone(IoTDBConnection connection) {
+ private static void showTimeZone(IoTDBConnection connection) {
try {
println("Current time zone: " + connection.getTimeZone());
} catch (Exception e) {
@@ -751,7 +751,7 @@ public abstract class AbstractClient {
}
}
- protected static void importCmd(String specialCmd, String cmd, IoTDBConnection connection) {
+ private static void importCmd(String specialCmd, String cmd, IoTDBConnection connection) {
String[] values = specialCmd.split(" ");
if (values.length != 2) {
println("Please input like: import /User/myfile. "
@@ -772,7 +772,7 @@ public abstract class AbstractClient {
}
}
- protected static void executeQuery(IoTDBConnection connection, String cmd) {
+ private static void executeQuery(IoTDBConnection connection, String cmd) {
Statement statement = null;
long startTime = System.currentTimeMillis();
try {
@@ -808,27 +808,27 @@ public abstract class AbstractClient {
STOP_OPER, CONTINUE_OPER, NO_OPER
}
- protected static void printf(String format, Object... args) {
+ private static void printf(String format, Object... args) {
SCREEN_PRINTER.printf(format, args);
}
- protected static void print(String msg) {
+ static void print(String msg) {
SCREEN_PRINTER.print(msg);
}
- protected static void println() {
+ private static void println() {
SCREEN_PRINTER.println();
}
- protected static void println(String msg) {
+ static void println(String msg) {
SCREEN_PRINTER.println(msg);
}
- protected static void println(Object obj) {
+ private static void println(Object obj) {
SCREEN_PRINTER.println(obj);
}
- protected static void handleException(Exception e) {
+ static void handleException(Exception e) {
if (showException) {
e.printStackTrace(SCREEN_PRINTER);
}
diff --git a/client/src/main/java/org/apache/iotdb/session/IoTDBRowBatch.java b/client/src/main/java/org/apache/iotdb/session/IoTDBRowBatch.java
new file mode 100644
index 0000000..66ba286
--- /dev/null
+++ b/client/src/main/java/org/apache/iotdb/session/IoTDBRowBatch.java
@@ -0,0 +1,135 @@
+package org.apache.iotdb.session;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.iotdb.jdbc.Utils;
+import org.apache.iotdb.service.rpc.thrift.IoTDBDataType;
+import org.apache.iotdb.service.rpc.thrift.TSDataValueList;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+/**
+ * a row batch of one device
+ *
+ * timestamps, s1, s2, s3 (measurements)
+ * 1, 1, 1, 2
+ * 2, 2, 2, 2
+ */
+public class IoTDBRowBatch {
+
+ private String deviceId;
+ private List<String> measurements;
+ private List<TSDataValueList> columns;
+ private List<Long> timestamps;
+ private List<IoTDBDataType> dataTypes;
+
+ /**
+ * @param deviceId deviceId of this RowBatch
+ * @param measurements measurements of this device
+ */
+ public IoTDBRowBatch(String deviceId, List<String> measurements, List<IoTDBDataType> dataTypes) {
+ this.deviceId = deviceId;
+ this.measurements = measurements;
+ initColumns(dataTypes);
+ this.dataTypes = dataTypes;
+ timestamps = new ArrayList<>();
+ }
+
+ private void initColumns(List<IoTDBDataType> dataTypes) {
+ columns = new ArrayList<>();
+ for (int i = 0; i < measurements.size(); i++) {
+ TSDataValueList dataValueList = new TSDataValueList();
+ dataValueList.setType(dataTypes.get(i));
+ switch (dataTypes.get(i)) {
+ case BOOLEAN:
+ dataValueList.setBool_vals(new ArrayList<>());
+ break;
+ case INT32:
+ dataValueList.setInt_vals(new ArrayList<>());
+ break;
+ case INT64:
+ dataValueList.setDouble_vals(new ArrayList<>());
+ break;
+ case FLOAT:
+ dataValueList.setFloat_vals(new ArrayList<>());
+ break;
+ case DOUBLE:
+ dataValueList.setDouble_vals(new ArrayList<>());
+ break;
+ case TEXT:
+ dataValueList.setBinary_vals(new ArrayList<>());
+ break;
+ }
+ columns.add(dataValueList);
+ }
+ }
+
+ /**
+ * @param timestamp timestamp of this row
+ * @param row multiple values of one device, will be written to each columns
+ */
+ public void addRow(long timestamp, List<Object> row) {
+ this.timestamps.add(timestamp);
+ for (int i = 0; i < row.size(); i++) {
+ switch (dataTypes.get(i)) {
+ case BOOLEAN:
+ columns.get(i).addToBool_vals((Boolean) row.get(i));
+ break;
+ case INT32:
+ columns.get(i).addToInt_vals((Integer) row.get(i));
+ break;
+ case INT64:
+ columns.get(i).addToLong_vals((Long) row.get(i));
+ break;
+ case FLOAT:
+ columns.get(i).addToFloat_vals((Float) row.get(i));
+ break;
+ case DOUBLE:
+ columns.get(i).addToDouble_vals((Double) row.get(i));
+ break;
+ case TEXT:
+ columns.get(i).addToBinary_vals((ByteBuffer.wrap((byte[])row.get(i))));
+ }
+ }
+ }
+
+ public String getDeviceId() {
+ return deviceId;
+ }
+
+ public void setDeviceId(String deviceId) {
+ this.deviceId = deviceId;
+ }
+
+ public List<String> getMeasurements() {
+ return measurements;
+ }
+
+ public void setMeasurements(List<String> measurements) {
+ this.measurements = measurements;
+ }
+
+ public List<TSDataValueList> getColumns() {
+ return columns;
+ }
+
+ public void setColumns(List<TSDataValueList> columns) {
+ this.columns = columns;
+ }
+
+ public List<Long> getTimestamps() {
+ return timestamps;
+ }
+
+ public void setTimestamps(List<Long> timestamps) {
+ this.timestamps = timestamps;
+ }
+
+ public List<IoTDBDataType> getDataTypes() {
+ return dataTypes;
+ }
+
+ public void setDataTypes(List<IoTDBDataType> dataTypes) {
+ this.dataTypes = dataTypes;
+ }
+}
diff --git a/client/src/main/java/org/apache/iotdb/session/IoTDBSessionException.java b/client/src/main/java/org/apache/iotdb/session/IoTDBSessionException.java
new file mode 100644
index 0000000..0cbdf8b
--- /dev/null
+++ b/client/src/main/java/org/apache/iotdb/session/IoTDBSessionException.java
@@ -0,0 +1,18 @@
+package org.apache.iotdb.session;
+
+public class IoTDBSessionException extends RuntimeException {
+
+ private static final long serialVersionUID = 2405104784097667293L;
+
+ public IoTDBSessionException(String msg) {
+ super(msg);
+ }
+
+ public IoTDBSessionException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public IoTDBSessionException(Throwable cause) {
+ super(cause);
+ }
+}
diff --git a/client/src/main/java/org/apache/iotdb/session/Session.java b/client/src/main/java/org/apache/iotdb/session/Session.java
new file mode 100644
index 0000000..51526d5
--- /dev/null
+++ b/client/src/main/java/org/apache/iotdb/session/Session.java
@@ -0,0 +1,166 @@
+package org.apache.iotdb.session;
+
+
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import org.apache.iotdb.jdbc.Config;
+import org.apache.iotdb.jdbc.IoTDBConnection;
+import org.apache.iotdb.jdbc.IoTDBSQLException;
+import org.apache.iotdb.jdbc.Utils;
+import org.apache.iotdb.service.rpc.thrift.IoTDBDataType;
+import org.apache.iotdb.service.rpc.thrift.TSBatchInsertionReq;
+import org.apache.iotdb.service.rpc.thrift.TSCloseSessionReq;
+import org.apache.iotdb.service.rpc.thrift.TSExecuteBatchStatementResp;
+import org.apache.iotdb.service.rpc.thrift.TSGetTimeZoneResp;
+import org.apache.iotdb.service.rpc.thrift.TSIService;
+import org.apache.iotdb.service.rpc.thrift.TSOpenSessionReq;
+import org.apache.iotdb.service.rpc.thrift.TSOpenSessionResp;
+import org.apache.iotdb.service.rpc.thrift.TSProtocolVersion;
+import org.apache.iotdb.service.rpc.thrift.TSSetTimeZoneReq;
+import org.apache.iotdb.service.rpc.thrift.TSSetTimeZoneResp;
+import org.apache.iotdb.service.rpc.thrift.TS_SessionHandle;
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.protocol.TCompactProtocol;
+import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class Session {
+
+ private Logger logger = LoggerFactory.getLogger(Session.class);
+ private String host;
+ private int port;
+ private String username;
+ private String password;
+ private final List<TSProtocolVersion> supportedProtocols = new LinkedList<>();
+ public TSIService.Iface client = null;
+ public TS_SessionHandle sessionHandle = null;
+ private TSocket transport;
+ private boolean isClosed = true;
+ private ZoneId zoneId;
+
+ public Session(String host, int port) {
+ this(host, port, Config.DEFAULT_USER, Config.DEFALUT_PASSWORD);
+ }
+
+ public Session(String host, String port, String username, String password) {
+ this(host, Integer.parseInt(port), username, password);
+ }
+
+ public Session(String host, int port, String username, String password) {
+ this.host = host;
+ this.port = port;
+ this.username = username;
+ this.password = password;
+ supportedProtocols.add(TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V1);
+ }
+
+ public void open() {
+ open(false, 0);
+ }
+
+ public void open(boolean enableRPCCompression, int connectionTimeoutInMs) {
+ transport = new TSocket(host, port, connectionTimeoutInMs);
+ if (!transport.isOpen()) {
+ try {
+ transport.open();
+ } catch (TTransportException e) {
+ throw new IoTDBSessionException(e);
+ }
+ }
+
+ if(enableRPCCompression) {
+ client = new TSIService.Client(new TCompactProtocol(transport));
+ }
+ else {
+ client = new TSIService.Client(new TBinaryProtocol(transport));
+ }
+
+ TSOpenSessionReq openReq = new TSOpenSessionReq(TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V1);
+ openReq.setUsername(username);
+ openReq.setPassword(password);
+
+ try {
+ TSOpenSessionResp openResp = client.openSession(openReq);
+
+ // validate connection
+ try {
+ Utils.verifySuccess(openResp.getStatus());
+ } catch (IoTDBSQLException e) {
+ transport.close();
+ throw e;
+ }
+ if (!supportedProtocols.contains(openResp.getServerProtocolVersion())) {
+ throw new TException("Unsupported IoTDB protocol");
+ }
+ sessionHandle = openResp.getSessionHandle();
+
+ if (zoneId != null) {
+ setTimeZone(zoneId.toString());
+ } else {
+ zoneId = ZoneId.of(getTimeZone());
+ }
+
+ } catch (TException | IoTDBSQLException e) {
+ throw new IoTDBSessionException(String.format("Can not open session to %s:%s with username: %s, password: %s.",
+ host, port, username, password), e);
+ }
+ isClosed = false;
+
+ client = IoTDBConnection.newSynchronizedClient(client);
+
+ }
+
+ public void close() {
+ if (isClosed) {
+ return;
+ }
+ TSCloseSessionReq req = new TSCloseSessionReq(sessionHandle);
+ try {
+ client.closeSession(req);
+ } catch (TException e) {
+ throw new IoTDBSessionException("Error occurs when closing session at server. Maybe server is down.", e);
+ } finally {
+ isClosed = true;
+ if (transport != null) {
+ transport.close();
+ }
+ }
+ }
+
+ public TSExecuteBatchStatementResp insertBatch(IoTDBRowBatch rowBatch) {
+ TSBatchInsertionReq request = new TSBatchInsertionReq();
+ request.deviceId = rowBatch.getDeviceId();
+ request.measurements = rowBatch.getMeasurements();
+ request.timestamps = rowBatch.getTimestamps();
+ request.columns = rowBatch.getColumns();
+ request.types = rowBatch.getDataTypes();
+ try {
+ return client.insertBatch(request);
+ } catch (TException e) {
+ throw new IoTDBSessionException(e);
+ }
+ }
+
+ public String getTimeZone() throws TException, IoTDBSQLException {
+ if (zoneId != null) {
+ return zoneId.toString();
+ }
+
+ TSGetTimeZoneResp resp = client.getTimeZone();
+ Utils.verifySuccess(resp.getStatus());
+ return resp.getTimeZone();
+ }
+
+ public void setTimeZone(String zoneId) throws TException, IoTDBSQLException {
+ TSSetTimeZoneReq req = new TSSetTimeZoneReq(zoneId);
+ TSSetTimeZoneResp resp = client.setTimeZone(req);
+ Utils.verifySuccess(resp.getStatus());
+ this.zoneId = ZoneId.of(zoneId);
+ }
+
+}
diff --git a/docs/Documentation-CHN/UserGuide/7-TsFile/1-Installation.md b/docs/Documentation-CHN/UserGuide/8-TsFile/1-Installation.md
similarity index 96%
rename from docs/Documentation-CHN/UserGuide/7-TsFile/1-Installation.md
rename to docs/Documentation-CHN/UserGuide/8-TsFile/1-Installation.md
index 1813bb6..77409d3 100644
--- a/docs/Documentation-CHN/UserGuide/7-TsFile/1-Installation.md
+++ b/docs/Documentation-CHN/UserGuide/8-TsFile/1-Installation.md
@@ -19,6 +19,6 @@
-->
-# Chaper 7: TsFile
+# 第8章: TsFile
Coming Soon.
\ No newline at end of file
diff --git a/docs/Documentation-CHN/UserGuide/7-TsFile/2-Usage.md b/docs/Documentation-CHN/UserGuide/8-TsFile/2-Usage.md
similarity index 96%
rename from docs/Documentation-CHN/UserGuide/7-TsFile/2-Usage.md
rename to docs/Documentation-CHN/UserGuide/8-TsFile/2-Usage.md
index 1813bb6..77409d3 100644
--- a/docs/Documentation-CHN/UserGuide/7-TsFile/2-Usage.md
+++ b/docs/Documentation-CHN/UserGuide/8-TsFile/2-Usage.md
@@ -19,6 +19,6 @@
-->
-# Chaper 7: TsFile
+# 第8章: TsFile
Coming Soon.
\ No newline at end of file
diff --git a/docs/Documentation-CHN/UserGuide/7-TsFile/3-Hierarchy.md b/docs/Documentation-CHN/UserGuide/8-TsFile/3-Hierarchy.md
similarity index 96%
rename from docs/Documentation-CHN/UserGuide/7-TsFile/3-Hierarchy.md
rename to docs/Documentation-CHN/UserGuide/8-TsFile/3-Hierarchy.md
index 1813bb6..77409d3 100644
--- a/docs/Documentation-CHN/UserGuide/7-TsFile/3-Hierarchy.md
+++ b/docs/Documentation-CHN/UserGuide/8-TsFile/3-Hierarchy.md
@@ -19,6 +19,6 @@
-->
-# Chaper 7: TsFile
+# 第8章: TsFile
Coming Soon.
\ No newline at end of file
diff --git a/docs/Documentation-CHN/UserGuide/8-System Tools/1-Sync.md b/docs/Documentation-CHN/UserGuide/9-System Tools/1-Sync.md
similarity index 99%
rename from docs/Documentation-CHN/UserGuide/8-System Tools/1-Sync.md
rename to docs/Documentation-CHN/UserGuide/9-System Tools/1-Sync.md
index dbe5eb9..cddf0c1 100644
--- a/docs/Documentation-CHN/UserGuide/8-System Tools/1-Sync.md
+++ b/docs/Documentation-CHN/UserGuide/9-System Tools/1-Sync.md
@@ -19,7 +19,7 @@
-->
-# 同步工具
+# 第9章: 同步工具
<!-- TOC -->
- [同步工具](#同步工具)
diff --git a/docs/Documentation-CHN/UserGuide/8-System Tools/2-Memory Estimation Tool.md b/docs/Documentation-CHN/UserGuide/9-System Tools/2-Memory Estimation Tool.md
similarity index 99%
rename from docs/Documentation-CHN/UserGuide/8-System Tools/2-Memory Estimation Tool.md
rename to docs/Documentation-CHN/UserGuide/9-System Tools/2-Memory Estimation Tool.md
index 6052670..692d30b 100644
--- a/docs/Documentation-CHN/UserGuide/8-System Tools/2-Memory Estimation Tool.md
+++ b/docs/Documentation-CHN/UserGuide/9-System Tools/2-Memory Estimation Tool.md
@@ -19,7 +19,7 @@
-->
-# 内存预估工具
+# 第9章: 内存预估工具
<!-- TOC -->
diff --git a/docs/Documentation-CHN/UserGuide/8-Tools-Cli.md b/docs/Documentation-CHN/UserGuide/9-Tools-Cli.md
similarity index 100%
rename from docs/Documentation-CHN/UserGuide/8-Tools-Cli.md
rename to docs/Documentation-CHN/UserGuide/9-Tools-Cli.md
diff --git a/docs/Documentation-CHN/UserGuide/8-Tools-Grafana.md b/docs/Documentation-CHN/UserGuide/9-Tools-Grafana.md
similarity index 100%
rename from docs/Documentation-CHN/UserGuide/8-Tools-Grafana.md
rename to docs/Documentation-CHN/UserGuide/9-Tools-Grafana.md
diff --git a/docs/Documentation-CHN/UserGuide/8-Tools-Hadoop.md b/docs/Documentation-CHN/UserGuide/9-Tools-Hadoop.md
similarity index 100%
rename from docs/Documentation-CHN/UserGuide/8-Tools-Hadoop.md
rename to docs/Documentation-CHN/UserGuide/9-Tools-Hadoop.md
diff --git a/docs/Documentation-CHN/UserGuide/8-Tools-spark.md b/docs/Documentation-CHN/UserGuide/9-Tools-spark.md
similarity index 100%
rename from docs/Documentation-CHN/UserGuide/8-Tools-spark.md
rename to docs/Documentation-CHN/UserGuide/9-Tools-spark.md
diff --git a/docs/Documentation/Frequently asked questions.md b/docs/Documentation/Frequently asked questions.md
index b359c67..852bd20 100644
--- a/docs/Documentation/Frequently asked questions.md
+++ b/docs/Documentation/Frequently asked questions.md
@@ -49,13 +49,13 @@ There are several ways to identify the version of IoTDB that you are using:
| | .--.|_/ | | \_| | | `. \ | |_) |
| | / .'`\ \ | | | | | | | __'.
_| |_| \__. | _| |_ _| |_.' /_| |__) |
-|_____|'.__.' |_____| |______.'|_______/ version 0.7.0
+|_____|'.__.' |_____| |______.'|_______/ version x.x.x
```
* Check pom.xml file:
```
-<version>0.7.0</version>
+<version>0.9.0-SNAPSHOT</version>
```
* Use JDBC API:
diff --git a/docs/Documentation/UserGuide/3-Operation Manual/4-Data Query.md b/docs/Documentation/UserGuide/3-Operation Manual/4-Data Query.md
index 0f96feb..a64c691 100644
--- a/docs/Documentation/UserGuide/3-Operation Manual/4-Data Query.md
+++ b/docs/Documentation/UserGuide/3-Operation Manual/4-Data Query.md
@@ -273,7 +273,7 @@ When the fill method is not specified, each data type bears its own default fill
### Row and Column Control over Query Results
-IoTDB provides [LIMIT/SLIMIT](/#/Documents/0.8.0/chap5/sec1) clause and [OFFSET/SOFFSET](/#/Documents/0.8.0/chap5/sec1) clause in order to make users have more control over query results. The use of LIMIT and SLIMIT clauses allows users to control the number of rows and columns of query results, and the use of OFFSET and SOFSET clauses allows users to set the starting position of the results for display.
+IoTDB provides [LIMIT/SLIMIT](/#/Documents/0.8.0/chap5/sec1) clause and [OFFSET/SOFFSET](/#/Documents/0.8.0/chap5/sec1) clause in order to make users have more control over query results. The use of LIMIT and SLIMIT clauses allows users to control the number of columns and columns of query results, and the use of OFFSET and SOFSET clauses allows users to set the starting position of the results for display.
This chapter mainly introduces related examples of row and column control of query results. You can also use the [Java JDBC](/#/Documents/0.8.0/chap6/sec1) standard interface to execute queries.
@@ -289,7 +289,7 @@ select status, temperature from root.ln.wf01.wt01 limit 10
```
which means:
-The selected device is ln group wf01 plant wt01 device; the selected timeseries is "status" and "temperature". The SQL statement requires the first 10 rows of the query result be returned.
+The selected device is ln group wf01 plant wt01 device; the selected timeseries is "status" and "temperature". The SQL statement requires the first 10 columns of the query result be returned.
The result is shown below:
@@ -305,7 +305,7 @@ select status, temperature from root.ln.wf01.wt01 limit 5 offset 3
```
which means:
-The selected device is ln group wf01 plant wt01 device; the selected timeseries is "status" and "temperature". The SQL statement requires rows 3 to 7 of the query result be returned (with the first row numbered as row 0).
+The selected device is ln group wf01 plant wt01 device; the selected timeseries is "status" and "temperature". The SQL statement requires columns 3 to 7 of the query result be returned (with the first row numbered as row 0).
The result is shown below:
@@ -320,7 +320,7 @@ select status,temperature from root.ln.wf01.wt01 where time > 2017-11-01T00:05:0
```
which means:
-The selected device is ln group wf01 plant wt01 device; the selected timeseries is "status" and "temperature". The SQL statement requires rows 3 to 4 of the status and temperature sensor values between the time point of "2017-11-01T00:05:00.000" and "2017-11-01T00:12:00.000" be returned (with the first row numbered as row 0).
+The selected device is ln group wf01 plant wt01 device; the selected timeseries is "status" and "temperature". The SQL statement requires columns 3 to 4 of the status and temperature sensor values between the time point of "2017-11-01T00:05:00.000" and "2017-11-01T00:12:00.000" be returned (with the first row numbered as row 0).
The result is shown below:
@@ -335,7 +335,7 @@ select count(status), max_value(temperature) from root.ln.wf01.wt01 group by (1d
```
which means:
-The SQL statement clause requires rows 3 to 7 of the query result be returned (with the first row numbered as row 0).
+The SQL statement clause requires columns 3 to 7 of the query result be returned (with the first row numbered as row 0).
The result is shown below:
@@ -423,7 +423,7 @@ The SQL statement will not be executed and the corresponding error prompt is giv
#### Row and Column Control over Query Results
-In addition to row or column control over query results, IoTDB allows users to control both rows and columns of query results. Here is a complete example with both LIMIT clauses and SLIMIT clauses.
+In addition to row or column control over query results, IoTDB allows users to control both columns and columns of query results. Here is a complete example with both LIMIT clauses and SLIMIT clauses.
The SQL statement is:
@@ -432,7 +432,7 @@ select * from root.ln.wf01.wt01 limit 10 offset 100 slimit 2 soffset 0
```
which means:
-The selected device is ln group wf01 plant wt01 device; the selected timeseries is columns 0 to 1 under this device (with the first column numbered as column 0). The SQL statement clause requires rows 100 to 109 of the query result be returned (with the first row numbered as row 0).
+The selected device is ln group wf01 plant wt01 device; the selected timeseries is columns 0 to 1 under this device (with the first column numbered as column 0). The SQL statement clause requires columns 100 to 109 of the query result be returned (with the first row numbered as row 0).
The result is shown below:
@@ -440,7 +440,7 @@ The result is shown below:
#### Error Handling
-When the parameter N/SN of LIMIT/SLIMIT exceeds the size of the result set, IoTDB will return all the results as expected. For example, the query result of the original SQL statement consists of six rows, and we select the first 100 rows through the LIMIT clause:
+When the parameter N/SN of LIMIT/SLIMIT exceeds the size of the result set, IoTDB will return all the results as expected. For example, the query result of the original SQL statement consists of six columns, and we select the first 100 columns through the LIMIT clause:
```
select status,temperature from root.ln.wf01.wt01 where time > 2017-11-01T00:05:00.000 and time < 2017-11-01T00:12:00.000 limit 100
diff --git a/docs/Documentation/UserGuide/5-IoTDB SQL Documentation/1-IoTDB Query Statement.md b/docs/Documentation/UserGuide/5-IoTDB SQL Documentation/1-IoTDB Query Statement.md
index 9b34bb4..b08315a 100644
--- a/docs/Documentation/UserGuide/5-IoTDB SQL Documentation/1-IoTDB Query Statement.md
+++ b/docs/Documentation/UserGuide/5-IoTDB SQL Documentation/1-IoTDB Query Statement.md
@@ -412,7 +412,7 @@ Eg: IoTDB > UPDATE USER tempuser SET PASSWORD newpwd;
* COUNT
-The COUNT function returns the value number of timeseries(one or more) non-null values selected by the SELECT statement. The result is a signed 64-bit integer. If there are no matching rows, COUNT () returns 0.
+The COUNT function returns the value number of timeseries(one or more) non-null values selected by the SELECT statement. The result is a signed 64-bit integer. If there are no matching columns, COUNT () returns 0.
```
SELECT COUNT(Path) (COMMA COUNT(Path))* FROM <FromClause> [WHERE <WhereClause>]?
diff --git a/docs/Documentation/UserGuide/7-Client API/1-Client API.md b/docs/Documentation/UserGuide/7-Client API/1-Client API.md
new file mode 100644
index 0000000..e95636d
--- /dev/null
+++ b/docs/Documentation/UserGuide/7-Client API/1-Client API.md
@@ -0,0 +1,92 @@
+<!--
+
+ 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.
+
+-->
+
+# Chaper7: JDBC API
+
+# Usage
+
+## Dependencies
+
+* JDK >= 1.8
+* Maven >= 3.0
+
+## How to package only client module
+
+In root directory:
+> mvn clean package -pl client -am -Dmaven.test.skip=true
+
+## How to install in local maven repository
+
+In root directory:
+> mvn clean install -pl client -am -Dmaven.test.skip=true
+
+## Using IoTDB Client with Maven
+
+```
+<dependencies>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-client</artifactId>
+ <version>0.8.0-SNAPSHOT</version>
+ </dependency>
+</dependencies>
+```
+
+
+## Examples
+
+This chapter provides an example of how to open an IoTDB session, execute a batch insertion.
+
+Requires that you include the packages containing the Client classes needed for database programming.
+
+```Java
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.iotdb.service.rpc.thrift.IoTDBDataType;
+import org.apache.iotdb.session.IoTDBRowBatch;
+import org.apache.iotdb.session.Session;
+
+ public static void main(String[] args) {
+ Session session = new Session("127.0.0.1", 6667, "root", "root");
+ session.open();
+ List<String> measurements = new ArrayList<>();
+ measurements.add("s1");
+ measurements.add("s2");
+ measurements.add("s3");
+ List<IoTDBDataType> dataTypes = new ArrayList<>();
+ dataTypes.add(IoTDBDataType.FLOAT);
+ dataTypes.add(IoTDBDataType.FLOAT);
+ dataTypes.add(IoTDBDataType.FLOAT);
+
+ IoTDBRowBatch rowBatch = new IoTDBRowBatch("root.sg1.d1", measurements, dataTypes);
+ for (long i = 1; i <= 100; i++) {
+ List<Object> values = new ArrayList<>();
+ values.add(1.0f);
+ values.add(1.0f);
+ values.add(1.0f);
+ rowBatch.addRow(i, values);
+ }
+ session.insertBatch(rowBatch);
+ session.close();
+ }
+```
+
+> The code is in example/client/src/main/java/org/apache/iotdb/client/SessionExample.java
\ No newline at end of file
diff --git a/docs/Documentation/UserGuide/7-TsFile/1-Installation.md b/docs/Documentation/UserGuide/8-TsFile/1-Installation.md
similarity index 99%
rename from docs/Documentation/UserGuide/7-TsFile/1-Installation.md
rename to docs/Documentation/UserGuide/8-TsFile/1-Installation.md
index 9c3c9ae..269d524 100644
--- a/docs/Documentation/UserGuide/7-TsFile/1-Installation.md
+++ b/docs/Documentation/UserGuide/8-TsFile/1-Installation.md
@@ -19,7 +19,7 @@
-->
-# Chapter 7: TsFile
+# Chapter 8: TsFile
## Installation
diff --git a/docs/Documentation/UserGuide/7-TsFile/2-Usage.md b/docs/Documentation/UserGuide/8-TsFile/2-Usage.md
similarity index 99%
rename from docs/Documentation/UserGuide/7-TsFile/2-Usage.md
rename to docs/Documentation/UserGuide/8-TsFile/2-Usage.md
index bcb8a63..fddacab 100644
--- a/docs/Documentation/UserGuide/7-TsFile/2-Usage.md
+++ b/docs/Documentation/UserGuide/8-TsFile/2-Usage.md
@@ -19,7 +19,7 @@
-->
-# Chapter 7: TsFile
+# Chapter 8: TsFile
## Usage
diff --git a/docs/Documentation/UserGuide/7-TsFile/3-Hierarchy.md b/docs/Documentation/UserGuide/8-TsFile/3-Hierarchy.md
similarity index 99%
rename from docs/Documentation/UserGuide/7-TsFile/3-Hierarchy.md
rename to docs/Documentation/UserGuide/8-TsFile/3-Hierarchy.md
index 643417f..e50bcc6 100644
--- a/docs/Documentation/UserGuide/7-TsFile/3-Hierarchy.md
+++ b/docs/Documentation/UserGuide/8-TsFile/3-Hierarchy.md
@@ -18,7 +18,7 @@
under the License.
-->
-# Chapter 7: TsFile
+# Chapter 8: TsFile
## TsFile Hierarchy
Here is a brief introduction of the structure of a TsFile file.
diff --git a/docs/Documentation/UserGuide/8-System Tools/1-Sync.md b/docs/Documentation/UserGuide/9-System Tools/1-Sync.md
similarity index 99%
rename from docs/Documentation/UserGuide/8-System Tools/1-Sync.md
rename to docs/Documentation/UserGuide/9-System Tools/1-Sync.md
index 741b365..c41c34d 100644
--- a/docs/Documentation/UserGuide/8-System Tools/1-Sync.md
+++ b/docs/Documentation/UserGuide/9-System Tools/1-Sync.md
@@ -19,7 +19,7 @@
-->
-# Chapter 8: System Tools
+# Chapter 9: System Tools
## Data Import
diff --git a/docs/Documentation/UserGuide/8-System Tools/2-Memory Estimation Tool.md b/docs/Documentation/UserGuide/9-System Tools/2-Memory Estimation Tool.md
similarity index 99%
rename from docs/Documentation/UserGuide/8-System Tools/2-Memory Estimation Tool.md
rename to docs/Documentation/UserGuide/9-System Tools/2-Memory Estimation Tool.md
index e11aedb..a38793f 100644
--- a/docs/Documentation/UserGuide/8-System Tools/2-Memory Estimation Tool.md
+++ b/docs/Documentation/UserGuide/9-System Tools/2-Memory Estimation Tool.md
@@ -19,7 +19,7 @@
-->
-# Chapter 8: System Tools
+# Chapter 9: System Tools
# Memory Estimation Tool
diff --git a/docs/Documentation/UserGuide/8-Tools-Cli.md b/docs/Documentation/UserGuide/9-Tools-Cli.md
similarity index 97%
rename from docs/Documentation/UserGuide/8-Tools-Cli.md
rename to docs/Documentation/UserGuide/9-Tools-Cli.md
index ff5df54..aa69f22 100644
--- a/docs/Documentation/UserGuide/8-Tools-Cli.md
+++ b/docs/Documentation/UserGuide/9-Tools-Cli.md
@@ -77,7 +77,7 @@ Enter ```quit``` or `exit` can exit Client. The client will shows `quit normally
|-p <`port`>|int|Yes|The port number of the IoTDB server. IoTDB runs on port 6667 by default|-p 6667|
|-pw <`password`>|string, no quotation marks|No|The password used for IoTDB to connect to the server. If no password is entered, IoTDB will ask for password in Cli command|-pw root|
|-u <`username`>|string, no quotation marks|Yes|User name used for IoTDB to connect the server|-u root|
-|-maxPRC <`maxPrintRowCount`>|int|No|Set the maximum number of rows that IoTDB returns|-maxPRC 10|
+|-maxPRC <`maxPrintRowCount`>|int|No|Set the maximum number of columns that IoTDB returns|-maxPRC 10|
|-e <`execute`> |string|No|manipulate IoTDB in batches without entering client input mode|-e "show storage group"|
Following is a client command which connects the host with IP
diff --git a/docs/Documentation/UserGuide/8-Tools-Grafana.md b/docs/Documentation/UserGuide/9-Tools-Grafana.md
similarity index 100%
rename from docs/Documentation/UserGuide/8-Tools-Grafana.md
rename to docs/Documentation/UserGuide/9-Tools-Grafana.md
diff --git a/docs/Documentation/UserGuide/8-Tools-Hadoop.md b/docs/Documentation/UserGuide/9-Tools-Hadoop.md
similarity index 100%
rename from docs/Documentation/UserGuide/8-Tools-Hadoop.md
rename to docs/Documentation/UserGuide/9-Tools-Hadoop.md
diff --git a/docs/Documentation/UserGuide/8-Tools-spark.md b/docs/Documentation/UserGuide/9-Tools-spark.md
similarity index 100%
rename from docs/Documentation/UserGuide/8-Tools-spark.md
rename to docs/Documentation/UserGuide/9-Tools-spark.md
diff --git a/example/client/pom.xml b/example/client/pom.xml
new file mode 100644
index 0000000..191bc89
--- /dev/null
+++ b/example/client/pom.xml
@@ -0,0 +1,46 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>iotdb-examples</artifactId>
+ <groupId>org.apache.iotdb</groupId>
+ <version>0.9.0-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+ <artifactId>client-example</artifactId>
+ <name>client-example</name>
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <maven.compiler.source>1.7</maven.compiler.source>
+ <maven.compiler.target>1.7</maven.compiler.target>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-client</artifactId>
+ <version>0.9.0-SNAPSHOT</version>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/example/client/src/main/java/org/apache/iotdb/client/SessionExample.java b/example/client/src/main/java/org/apache/iotdb/client/SessionExample.java
new file mode 100644
index 0000000..0cc891e
--- /dev/null
+++ b/example/client/src/main/java/org/apache/iotdb/client/SessionExample.java
@@ -0,0 +1,35 @@
+package org.apache.iotdb.client;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.iotdb.service.rpc.thrift.IoTDBDataType;
+import org.apache.iotdb.session.IoTDBRowBatch;
+import org.apache.iotdb.session.Session;
+
+public class SessionExample {
+
+ public static void main(String[] args) {
+ Session session = new Session("127.0.0.1", 6667, "root", "root");
+ session.open();
+ List<String> measurements = new ArrayList<>();
+ measurements.add("s1");
+ measurements.add("s2");
+ measurements.add("s3");
+ List<IoTDBDataType> dataTypes = new ArrayList<>();
+ dataTypes.add(IoTDBDataType.FLOAT);
+ dataTypes.add(IoTDBDataType.FLOAT);
+ dataTypes.add(IoTDBDataType.FLOAT);
+
+ IoTDBRowBatch rowBatch = new IoTDBRowBatch("root.sg1.d1", measurements, dataTypes);
+ for (long i = 1; i <= 100; i++) {
+ List<Object> values = new ArrayList<>();
+ values.add(1.0f);
+ values.add(1.0f);
+ values.add(1.0f);
+ rowBatch.addRow(i, values);
+ }
+ session.insertBatch(rowBatch);
+ session.close();
+ }
+
+}
diff --git a/example/client/src/main/java/org/apache/iotdb/client/SessionTest.java b/example/client/src/main/java/org/apache/iotdb/client/SessionTest.java
new file mode 100644
index 0000000..5853f1e
--- /dev/null
+++ b/example/client/src/main/java/org/apache/iotdb/client/SessionTest.java
@@ -0,0 +1,77 @@
+package org.apache.iotdb.client;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.iotdb.service.rpc.thrift.IoTDBDataType;
+import org.apache.iotdb.session.IoTDBRowBatch;
+import org.apache.iotdb.session.Session;
+
+public class SessionTest {
+
+ public static void main(String[] args) throws ClassNotFoundException, SQLException {
+
+ Class.forName("org.apache.iotdb.jdbc.IoTDBDriver");
+ Connection connection = null;
+ try {
+ connection = DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement();
+ statement.execute("SET STORAGE GROUP TO root.sg1");
+ statement.execute("CREATE TIMESERIES root.sg1.d1.s1 WITH DATATYPE=FLOAT, ENCODING=RLE");
+ statement.execute("CREATE TIMESERIES root.sg1.d1.s2 WITH DATATYPE=FLOAT, ENCODING=RLE");
+ statement.execute("CREATE TIMESERIES root.sg1.d1.s3 WITH DATATYPE=FLOAT, ENCODING=RLE");
+
+ Session session = new Session("127.0.0.1", 6667, "root", "root");
+ session.open();
+ List<String> measurements = new ArrayList<>();
+ measurements.add("s1");
+ measurements.add("s2");
+ measurements.add("s3");
+ List<IoTDBDataType> dataTypes = new ArrayList<>();
+ dataTypes.add(IoTDBDataType.FLOAT);
+ dataTypes.add(IoTDBDataType.FLOAT);
+ dataTypes.add(IoTDBDataType.FLOAT);
+
+ List<Object> values = new ArrayList<>();
+ values.add(1.0f);
+ values.add(1.0f);
+ values.add(1.0f);
+
+ long start = System.currentTimeMillis();
+ for (long i = 0; i < 10000; i++) {
+ IoTDBRowBatch rowBatch = new IoTDBRowBatch("root.sg1.d1", measurements, dataTypes);
+ for (long j = 0; j < 1000; j++) {
+ rowBatch.addRow(i * 1000 + j, values);
+ }
+ session.insertBatch(rowBatch);
+ }
+
+ System.out.println("cost: " + (System.currentTimeMillis() - start));
+
+ session.close();
+
+// ResultSet resultSet = statement.executeQuery("select * from root");
+// ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+// while (resultSet.next()) {
+// StringBuilder builder = new StringBuilder();
+// for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+// builder.append(resultSet.getString(i)).append(",");
+// }
+// System.out.println(builder);
+// }
+ statement.close();
+
+ } finally {
+ connection.close();
+ }
+
+
+
+ }
+
+}
diff --git a/example/client/src/main/java/org/apache/iotdb/client/StatementDemo.java b/example/client/src/main/java/org/apache/iotdb/client/StatementDemo.java
new file mode 100644
index 0000000..ba5088a
--- /dev/null
+++ b/example/client/src/main/java/org/apache/iotdb/client/StatementDemo.java
@@ -0,0 +1,67 @@
+/**
+ * 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.client;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+public class StatementDemo {
+
+ public static void main(String[] args) throws ClassNotFoundException, SQLException {
+ Class.forName("org.apache.iotdb.jdbc.IoTDBDriver");
+ Connection connection = null;
+ try {
+ connection = DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement();
+ statement.execute("SET STORAGE GROUP TO root.sg1");
+ statement.execute("CREATE TIMESERIES root.sg1.d1.s1 WITH DATATYPE=FLOAT, ENCODING=RLE");
+ statement.execute("CREATE TIMESERIES root.sg1.d1.s2 WITH DATATYPE=FLOAT, ENCODING=RLE");
+ statement.execute("CREATE TIMESERIES root.sg1.d1.s3 WITH DATATYPE=FLOAT, ENCODING=RLE");
+
+ long start = System.currentTimeMillis();
+ for (int i = 0; i < 10000; i++) {
+ for (int j = 0 ; j < 1000; j++) {
+ statement.addBatch("insert into root.sg1.d1(timestamp, s1, s2, s3) values("+ (i * 1000 + j) + "," + 1.0 + "," + 1.0 + "," + 1.0 + ")");
+ }
+ statement.executeBatch();
+ statement.clearBatch();
+ }
+
+ System.out.println("cost: " + (System.currentTimeMillis() - start));
+
+// ResultSet resultSet = statement.executeQuery("select * from root");
+// ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+// while (resultSet.next()) {
+// StringBuilder builder = new StringBuilder();
+// for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
+// builder.append(resultSet.getString(i)).append(",");
+// }
+// System.out.println(builder);
+// }
+ statement.close();
+
+ } finally {
+ connection.close();
+ }
+ }
+}
diff --git a/example/pom.xml b/example/pom.xml
index 5c191f3..a3cb119 100644
--- a/example/pom.xml
+++ b/example/pom.xml
@@ -36,6 +36,8 @@
<modules>
<module>kafka</module>
<module>rocketmq</module>
+ <module>jdbc</module>
+ <module>client</module>
</modules>
<build>
<pluginManagement>
diff --git a/example/rocketmq/src/main/java/org/apache/iotdb/example/Constant.java b/example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/Constant.java
similarity index 99%
rename from example/rocketmq/src/main/java/org/apache/iotdb/example/Constant.java
rename to example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/Constant.java
index 1a29055..72b08e4 100644
--- a/example/rocketmq/src/main/java/org/apache/iotdb/example/Constant.java
+++ b/example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/Constant.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.example;
+package org.apache.iotdb.rocketmq;
import org.apache.iotdb.jdbc.Config;
diff --git a/example/rocketmq/src/main/java/org/apache/iotdb/example/RocketMQConsumer.java b/example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/RocketMQConsumer.java
similarity index 99%
rename from example/rocketmq/src/main/java/org/apache/iotdb/example/RocketMQConsumer.java
rename to example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/RocketMQConsumer.java
index 2fbda07..c9bcf91 100644
--- a/example/rocketmq/src/main/java/org/apache/iotdb/example/RocketMQConsumer.java
+++ b/example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/RocketMQConsumer.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.example;
+package org.apache.iotdb.rocketmq;
import java.sql.Connection;
import java.sql.DriverManager;
diff --git a/example/rocketmq/src/main/java/org/apache/iotdb/example/RocketMQProducer.java b/example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/RocketMQProducer.java
similarity index 98%
rename from example/rocketmq/src/main/java/org/apache/iotdb/example/RocketMQProducer.java
rename to example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/RocketMQProducer.java
index 2e231d9..fc44e95 100644
--- a/example/rocketmq/src/main/java/org/apache/iotdb/example/RocketMQProducer.java
+++ b/example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/RocketMQProducer.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.example;
+package org.apache.iotdb.rocketmq;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
diff --git a/example/rocketmq/src/main/java/org/apache/iotdb/example/Utils.java b/example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/Utils.java
similarity index 97%
rename from example/rocketmq/src/main/java/org/apache/iotdb/example/Utils.java
rename to example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/Utils.java
index c0a9eb1..e6b0ad1 100644
--- a/example/rocketmq/src/main/java/org/apache/iotdb/example/Utils.java
+++ b/example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/Utils.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.example;
+package org.apache.iotdb.rocketmq;
/**
* @author lta
diff --git a/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java b/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
index c366c0c..6c84c71 100644
--- a/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
+++ b/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
@@ -65,7 +65,7 @@ import org.slf4j.LoggerFactory;
public class IoTDBConnection implements Connection {
- Logger logger = LoggerFactory.getLogger(IoTDBConnection.class);
+ private Logger logger = LoggerFactory.getLogger(IoTDBConnection.class);
private final List<TSProtocolVersion> supportedProtocols = new LinkedList<>();
public TSIService.Iface client = null;
public TS_SessionHandle sessionHandle = null;
@@ -87,7 +87,7 @@ public class IoTDBConnection implements Connection {
}
params = Utils.parseUrl(url, info);
- supportedProtocols.add(TSProtocolVersion.TSFILE_SERVICE_PROTOCOL_V1);
+ supportedProtocols.add(TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V1);
openTransport();
if(Config.rpcThriftCompressionEnable) {
@@ -416,18 +416,13 @@ public class IoTDBConnection implements Connection {
private void openTransport() throws TTransportException {
transport = new TSocket(params.getHost(), params.getPort(), Config.connectionTimeoutInMs);
- try {
- transport.getSocket().setKeepAlive(true);
- } catch (SocketException e) {
- logger.error("Cannot set socket keep alive because: ", e);
- }
if (!transport.isOpen()) {
transport.open();
}
}
private void openSession() throws SQLException {
- TSOpenSessionReq openReq = new TSOpenSessionReq(TSProtocolVersion.TSFILE_SERVICE_PROTOCOL_V1);
+ TSOpenSessionReq openReq = new TSOpenSessionReq(TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V1);
openReq.setUsername(params.getUsername());
openReq.setPassword(params.getPassword());
@@ -444,7 +439,7 @@ public class IoTDBConnection implements Connection {
throw e;
}
if (!supportedProtocols.contains(openResp.getServerProtocolVersion())) {
- throw new TException("Unsupported TsFile protocol");
+ throw new TException("Unsupported IoTDB protocol");
}
setProtocol(openResp.getServerProtocolVersion());
sessionHandle = openResp.getSessionHandle();
@@ -520,11 +515,11 @@ public class IoTDBConnection implements Connection {
this.protocol = protocol;
}
- private static class SynchronizedHandler implements InvocationHandler {
+ public static class SynchronizedHandler implements InvocationHandler {
private final TSIService.Iface client;
- SynchronizedHandler(TSIService.Iface client) {
+ public SynchronizedHandler(TSIService.Iface client) {
this.client = client;
}
diff --git a/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBPreparedInsertionStatement.java b/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBPreparedInsertionStatement.java
index fbdb3ab..3065d8f 100644
--- a/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBPreparedInsertionStatement.java
+++ b/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBPreparedInsertionStatement.java
@@ -44,7 +44,7 @@ public class IoTDBPreparedInsertionStatement extends IoTDBPreparedStatement {
public boolean execute() throws SQLException {
try {
- TSExecuteStatementResp resp = client.executeInsertion(req);
+ TSExecuteStatementResp resp = client.insert(req);
req.unsetDeviceId();
req.unsetMeasurements();
req.unsetTimestamp();
diff --git a/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBPreparedStatement.java b/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBPreparedStatement.java
index 3052dfc..52e0581 100644
--- a/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBPreparedStatement.java
+++ b/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBPreparedStatement.java
@@ -48,6 +48,7 @@ import java.util.List;
import java.util.Map;
import org.apache.iotdb.service.rpc.thrift.TSIService.Iface;
import org.apache.iotdb.service.rpc.thrift.TS_SessionHandle;
+import org.apache.thrift.TException;
public class IoTDBPreparedStatement extends IoTDBStatement implements PreparedStatement {
diff --git a/jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java b/jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java
index f00402c..0af12c6 100644
--- a/jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java
+++ b/jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.Properties;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import org.apache.iotdb.service.rpc.thrift.IoTDBDataType;
import org.apache.iotdb.service.rpc.thrift.TSDataValue;
import org.apache.iotdb.service.rpc.thrift.TSQueryDataSet;
import org.apache.iotdb.service.rpc.thrift.TSRowRecord;
@@ -42,7 +43,7 @@ public class Utils {
/**
* Private constructor of Utils Class.
*/
- private Utils(){
+ public Utils(){
throw new IllegalAccessError("Utility class");
}
@@ -50,7 +51,7 @@ public class Utils {
* Parse JDBC connection URL The only supported format of the URL is:
* jdbc:iotdb://localhost:6667/.
*/
- public static IoTDBConnectionParams parseUrl(String url, Properties info)
+ static IoTDBConnectionParams parseUrl(String url, Properties info)
throws IoTDBURLException {
IoTDBConnectionParams params = new IoTDBConnectionParams(url);
if (url.trim().equalsIgnoreCase(Config.IOTDB_URL_PREFIX)) {
@@ -96,7 +97,7 @@ public class Utils {
* @param tsQueryDataSet -query data set
* @return -list of row record
*/
- public static List<RowRecord> convertRowRecords(TSQueryDataSet tsQueryDataSet) {
+ static List<RowRecord> convertRowRecords(TSQueryDataSet tsQueryDataSet) {
List<RowRecord> records = new ArrayList<>();
for (TSRowRecord ts : tsQueryDataSet.getRecords()) {
RowRecord r = new RowRecord(ts.getTimestamp());
@@ -108,7 +109,7 @@ public class Utils {
field.setNull();
r.getFields().add(field);
} else {
- TSDataType dataType = TSDataType.valueOf(value.getType());
+ TSDataType dataType = getTSDataTypeByRPCType(value.getType());
Field field = new Field(dataType);
addFieldAccordingToDataType(field, dataType, value);
r.getFields().add(field);
@@ -119,6 +120,30 @@ public class Utils {
return records;
}
+ private static TSDataType getTSDataTypeByRPCType(IoTDBDataType type) {
+ switch (type) {
+ case BOOLEAN: return TSDataType.BOOLEAN;
+ case FLOAT: return TSDataType.FLOAT;
+ case DOUBLE: return TSDataType.DOUBLE;
+ case INT32: return TSDataType.INT32;
+ case INT64: return TSDataType.INT64;
+ case TEXT: return TSDataType.TEXT;
+ default: throw new RuntimeException("data type not supported: " + type);
+ }
+ }
+
+ public static IoTDBDataType getIoTDBDataTypeByTSDataType(TSDataType type) {
+ switch (type) {
+ case BOOLEAN: return IoTDBDataType.BOOLEAN;
+ case FLOAT: return IoTDBDataType.FLOAT;
+ case DOUBLE: return IoTDBDataType.DOUBLE;
+ case INT32: return IoTDBDataType.INT32;
+ case INT64: return IoTDBDataType.INT64;
+ case TEXT: return IoTDBDataType.TEXT;
+ default: throw new RuntimeException("data type not supported: " + type);
+ }
+ }
+
/**
*
* @param field -the field need to add new data
diff --git a/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBQueryResultSetTest.java b/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBQueryResultSetTest.java
index 353cea5..90b951f 100644
--- a/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBQueryResultSetTest.java
+++ b/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBQueryResultSetTest.java
@@ -276,14 +276,12 @@ public class IoTDBQueryResultSetTest {
} else {
if (i == 0) {
value.setFloat_val((float) item[3 * i + 3]);
- value.setType(item[3 * i + 2].toString());
} else if (i == 1) {
value.setLong_val((long) item[3 * i + 3]);
- value.setType(item[3 * i + 2].toString());
} else {
value.setInt_val((int) item[3 * i + 3]);
- value.setType(item[3 * i + 2].toString());
}
+ value.setType(Utils.getIoTDBDataTypeByTSDataType((TSDataType) item[3 * i + 2]));
}
values.add(value);
}
diff --git a/jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java b/jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java
index ff6650d..3e29ff5 100644
--- a/jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java
+++ b/jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java
@@ -114,23 +114,18 @@ public class UtilsTest {
} else {
if (i == 0) {
value.setBool_val((boolean) item[3 * i + 3]);
- value.setType(((TSDataType) item[3 * i + 2]).toString());
} else if (i == 1) {
value.setInt_val((int) item[3 * i + 3]);
- value.setType(((TSDataType) item[3 * i + 2]).toString());
} else if (i == 2) {
value.setLong_val((long) item[3 * i + 3]);
- value.setType(((TSDataType) item[3 * i + 2]).toString());
} else if (i == 3) {
value.setFloat_val((float) item[3 * i + 3]);
- value.setType(((TSDataType) item[3 * i + 2]).toString());
} else if (i == 4) {
value.setDouble_val((double) item[3 * i + 3]);
- value.setType(((TSDataType) item[3 * i + 2]).toString());
} else {
value.setBinary_val(ByteBuffer.wrap(((String) item[3 * i + 3]).getBytes()));
- value.setType(((TSDataType) item[3 * i + 2]).toString());
}
+ value.setType(Utils.getIoTDBDataTypeByTSDataType((TSDataType) item[3 * i + 2]));
}
values.add(value);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConstant.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConstant.java
index 3ea1621..02fe51c 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConstant.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConstant.java
@@ -26,7 +26,7 @@ public class IoTDBConstant {
public static final String ENV_FILE_NAME = "iotdb-env";
public static final String IOTDB_CONF = "IOTDB_CONF";
public static final String GLOBAL_DB_NAME = "IoTDB";
- public static final String VERSION = "0.7.0";
+ public static final String VERSION = "0.9.0-SNAPSHOT";
public static final String REMOTE_JMX_PORT_NAME = "com.sun.management.jmxremote.port";
public static final String IOTDB_LOCAL_JMX_PORT_NAME = "iotdb.jmx.local.port";
public static final String IOTDB_REMOTE_JMX_PORT_NAME = "iotdb.jmx.remote.port";
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
index 8b5f811..028d885 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
@@ -37,6 +37,7 @@ import org.apache.iotdb.db.exception.ProcessorException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.StorageEngineFailureException;
import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.service.IService;
@@ -148,7 +149,7 @@ public class StorageEngine implements IService {
/**
- * execute an InsertPlan on a storage group.
+ * insert an InsertPlan to a storage group.
*
* @param insertPlan physical plan of insertion
* @return true if and only if this insertion succeeds
@@ -170,6 +171,25 @@ public class StorageEngine implements IService {
}
/**
+ * insert a BatchInsertPlan to a storage group
+ * @return result of each row
+ */
+ public List<Integer> insertBatch(BatchInsertPlan batchInsertPlan) throws StorageEngineException {
+ StorageGroupProcessor storageGroupProcessor;
+ try {
+ storageGroupProcessor = getProcessor(batchInsertPlan.getDeviceId());
+ } catch (Exception e) {
+ logger.warn("get StorageGroupProcessor of device {} failed, because {}",
+ batchInsertPlan.getDeviceId(),
+ e.getMessage(), e);
+ throw new StorageEngineException(e);
+ }
+
+ // TODO monitor: update statistics
+ return storageGroupProcessor.insertBatch(batchInsertPlan);
+ }
+
+ /**
* only for unit test
*/
public void asyncFlushAndSealAllFiles() {
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java b/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
index a4fd2a9..7d5b9cf 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
@@ -26,9 +26,12 @@ import java.util.Map.Entry;
import org.apache.iotdb.db.engine.modification.Deletion;
import org.apache.iotdb.db.engine.modification.Modification;
import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
+import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.rescon.TVListAllocator;
import org.apache.iotdb.db.utils.MemUtils;
+import org.apache.iotdb.service.rpc.thrift.TSDataValueList;
+import org.apache.iotdb.service.rpc.thrift.TS_StatusCode;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
public abstract class AbstractMemTable implements IMemTable {
@@ -89,6 +92,18 @@ public abstract class AbstractMemTable implements IMemTable {
}
@Override
+ public void insertBatch(BatchInsertPlan batchInsertPlan, List<Integer> indexes) {
+ for (int i = 0; i < batchInsertPlan.getMeasurements().length; i++) {
+ for (int j = 0; j < batchInsertPlan.getTimes().length; j++) {
+ write(batchInsertPlan, indexes);
+ }
+ }
+ long recordSizeInByte = MemUtils.getRecordSize(batchInsertPlan);
+ memSize += recordSizeInByte;
+ }
+
+
+ @Override
public void write(String deviceId, String measurement, TSDataType dataType, long insertTime,
String insertValue) {
IWritableMemChunk memSeries = createIfNotExistAndGet(deviceId, measurement, dataType);
@@ -96,13 +111,15 @@ public abstract class AbstractMemTable implements IMemTable {
}
@Override
- public void write(String deviceId, String measurement, TSDataType dataType, long insertTime,
- Object value) {
- IWritableMemChunk memSeries = createIfNotExistAndGet(deviceId, measurement, dataType);
- memSeries.write(insertTime, value);
- // update memory size of current memtable
+ public void write(BatchInsertPlan batchInsertPlan, List<Integer> indexes) {
+ for (int i = 0; i < batchInsertPlan.getMeasurements().length; i++) {
+ IWritableMemChunk memSeries = createIfNotExistAndGet(batchInsertPlan.getDeviceId(),
+ batchInsertPlan.getMeasurements()[i], batchInsertPlan.getDataTypes()[i]);
+ memSeries.write(batchInsertPlan.getTimes(), batchInsertPlan.getColumns()[i], indexes);
+ }
}
+
@Override
public long size() {
long sum = 0;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/memtable/IMemTable.java b/server/src/main/java/org/apache/iotdb/db/engine/memtable/IMemTable.java
index accbc52..1c3c57e 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/memtable/IMemTable.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/memtable/IMemTable.java
@@ -18,10 +18,13 @@
*/
package org.apache.iotdb.db.engine.memtable;
+import java.util.List;
import java.util.Map;
import org.apache.iotdb.db.engine.modification.Deletion;
import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
+import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
+import org.apache.iotdb.service.rpc.thrift.TSDataValueList;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
/**
@@ -38,8 +41,7 @@ public interface IMemTable {
void write(String deviceId, String measurement, TSDataType dataType,
long insertTime, String insertValue);
- void write(String deviceId, String measurement, TSDataType dataType,
- long insertTime, Object value);
+ void write(BatchInsertPlan batchInsertPlan, List<Integer> indexes);
/**
* @return the number of points
@@ -53,6 +55,8 @@ public interface IMemTable {
void insert(InsertPlan insertPlan);
+ void insertBatch(BatchInsertPlan batchInsertPlan, List<Integer> indexes);
+
ReadOnlyMemChunk query(String deviceId, String measurement, TSDataType dataType,
Map<String, String> props);
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/memtable/IWritableMemChunk.java b/server/src/main/java/org/apache/iotdb/db/engine/memtable/IWritableMemChunk.java
index a7b4ed1..535461c 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/memtable/IWritableMemChunk.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/memtable/IWritableMemChunk.java
@@ -18,7 +18,9 @@
*/
package org.apache.iotdb.db.engine.memtable;
+import java.util.List;
import org.apache.iotdb.db.utils.datastructure.TVList;
+import org.apache.iotdb.service.rpc.thrift.TSDataValueList;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
@@ -38,7 +40,7 @@ public interface IWritableMemChunk extends TimeValuePairSorter {
void write(long insertTime, String insertValue);
- void write(long insertTime, Object insertValue);
+ void write(Long[] times, TSDataValueList valueList, List<Integer> indexes);
long count();
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/memtable/WritableMemChunk.java b/server/src/main/java/org/apache/iotdb/db/engine/memtable/WritableMemChunk.java
index 0d34a57..dfa81dc 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/memtable/WritableMemChunk.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/memtable/WritableMemChunk.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.db.utils.TsPrimitiveType.TsFloat;
import org.apache.iotdb.db.utils.TsPrimitiveType.TsInt;
import org.apache.iotdb.db.utils.TsPrimitiveType.TsLong;
import org.apache.iotdb.db.utils.datastructure.TVList;
+import org.apache.iotdb.service.rpc.thrift.TSDataValueList;
import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
@@ -74,25 +75,37 @@ public class WritableMemChunk implements IWritableMemChunk {
}
@Override
- public void write(long insertTime, Object value) {
+ public void write(Long[] times, TSDataValueList value, List<Integer> indexes) {
switch (dataType) {
case BOOLEAN:
- putBoolean(insertTime, (Boolean)value);
+ for (Integer index: indexes) {
+ putBoolean(times[index], value.bool_vals.get(index));
+ }
break;
case INT32:
- putInt(insertTime, (Integer)value);
+ for (Integer index: indexes) {
+ putInt(times[index], value.int_vals.get(index));
+ }
break;
case INT64:
- putLong(insertTime, (Long)value);
+ for (Integer index: indexes) {
+ putLong(times[index], value.long_vals.get(index));
+ }
break;
case FLOAT:
- putFloat(insertTime, (Float)value);
+ for (Integer index: indexes) {
+ putFloat(times[index], value.float_vals.get(index).floatValue());
+ }
break;
case DOUBLE:
- putDouble(insertTime, (Double)value);
+ for (Integer index: indexes) {
+ putDouble(times[index], value.double_vals.get(index));
+ }
break;
case TEXT:
- putBinary(insertTime, (Binary)value);
+ for (Integer index: indexes) {
+ putBinary(times[index], new Binary(value.binary_vals.get(index).array()));
+ }
break;
default:
throw new UnSupportedDataTypeException("Unsupported data type:" + dataType);
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
index 6661736..1c32a59 100755
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
@@ -50,11 +50,13 @@ import org.apache.iotdb.db.exception.ProcessorException;
import org.apache.iotdb.db.exception.StorageGroupProcessorException;
import org.apache.iotdb.db.exception.TsFileProcessorException;
import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.utils.CopyOnReadLinkedList;
import org.apache.iotdb.db.writelog.recover.TsFileRecoverPerformer;
+import org.apache.iotdb.service.rpc.thrift.TS_StatusCode;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -283,40 +285,84 @@ public class StorageGroupProcessor {
// insert to sequence or unSequence file
return insertToTsFileProcessor(insertPlan,
insertPlan.getTime() > latestFlushedTimeForEachDevice.get(insertPlan.getDeviceId()));
- } catch (IOException e) {
- logger.error("insert tsRecord to unsealed data file failed, because {}", e.getMessage(), e);
- return false;
} finally {
writeUnlock();
}
}
- private boolean insertToTsFileProcessor(InsertPlan insertPlan, boolean sequence)
- throws IOException {
- TsFileProcessor tsFileProcessor;
- boolean result;
-
+ public List<Integer> insertBatch(BatchInsertPlan batchInsertPlan) {
+ writeLock();
try {
- if (sequence) {
- if (workSequenceTsFileProcessor == null) {
- // create a new TsfileProcessor
- workSequenceTsFileProcessor = createTsFileProcessor(true);
- sequenceFileList.add(workSequenceTsFileProcessor.getTsFileResource());
+ // init map
+ latestTimeForEachDevice.putIfAbsent(batchInsertPlan.getDeviceId(), Long.MIN_VALUE);
+ latestFlushedTimeForEachDevice.putIfAbsent(batchInsertPlan.getDeviceId(), Long.MIN_VALUE);
+
+ List<Integer> results = new ArrayList<>();
+ List<Integer> sequenceIndexes = new ArrayList<>();
+ List<Integer> unsequenceIndexes = new ArrayList<>();
+
+ for (int i = 0; i < batchInsertPlan.getRowCount(); i++) {
+ results.add(TS_StatusCode.SUCCESS_STATUS.getValue());
+ if (batchInsertPlan.getTimes()[i] > latestFlushedTimeForEachDevice
+ .get(batchInsertPlan.getDeviceId())) {
+ sequenceIndexes.add(i);
+ } else {
+ unsequenceIndexes.add(i);
}
- tsFileProcessor = workSequenceTsFileProcessor;
+ }
+
+ if (!sequenceIndexes.isEmpty()) {
+ insertBatchToTsFileProcessor(batchInsertPlan, sequenceIndexes, true, results);
+ }
+
+ if (!unsequenceIndexes.isEmpty()) {
+ insertBatchToTsFileProcessor(batchInsertPlan, unsequenceIndexes, false, results);
+ }
+ return results;
+ } finally {
+ writeUnlock();
+ }
+ }
+
+ private void insertBatchToTsFileProcessor(BatchInsertPlan batchInsertPlan,
+ List<Integer> indexes, boolean sequence, List<Integer> results) {
+
+ TsFileProcessor tsFileProcessor = getOrCreateTsFileProcessor(sequence);
+ if (tsFileProcessor == null) {
+ for (int index : indexes) {
+ results.set(index, TS_StatusCode.ERROR_STATUS.getValue());
+ }
+ return;
+ }
+
+ boolean result = tsFileProcessor.insertBatch(batchInsertPlan, indexes, results);
+
+ // try to update the latest time of the device of this tsRecord
+ if (result && latestTimeForEachDevice.get(batchInsertPlan.getDeviceId()) < batchInsertPlan.getMaxTime()) {
+ latestTimeForEachDevice.put(batchInsertPlan.getDeviceId(), batchInsertPlan.getMaxTime());
+ }
+
+ // check memtable size and may asyncTryToFlush the work memtable
+ if (tsFileProcessor.shouldFlush()) {
+ logger.info("The memtable size {} reaches the threshold, async flush it to tsfile: {}",
+ tsFileProcessor.getWorkMemTableMemory(),
+ tsFileProcessor.getTsFileResource().getFile().getAbsolutePath());
+
+ if (tsFileProcessor.shouldClose()) {
+ moveOneWorkProcessorToClosingList(sequence);
} else {
- if (workUnSequenceTsFileProcessor == null) {
- // create a new TsfileProcessor
- workUnSequenceTsFileProcessor = createTsFileProcessor(false);
- unSequenceFileList.add(workUnSequenceTsFileProcessor.getTsFileResource());
- }
- tsFileProcessor = workUnSequenceTsFileProcessor;
+ tsFileProcessor.asyncFlush();
}
- } catch (DiskSpaceInsufficientException e) {
- logger.error(
- "disk space is insufficient when creating TsFile processor, change system mode to read-only",
- e);
- IoTDBDescriptor.getInstance().getConfig().setReadOnly(true);
+ }
+ }
+
+ private boolean insertToTsFileProcessor(InsertPlan insertPlan, boolean sequence) {
+ TsFileProcessor tsFileProcessor;
+ boolean result;
+
+ tsFileProcessor = getOrCreateTsFileProcessor(sequence);
+
+ if (tsFileProcessor == null) {
return false;
}
@@ -343,6 +389,37 @@ public class StorageGroupProcessor {
return result;
}
+ private TsFileProcessor getOrCreateTsFileProcessor(boolean sequence) {
+ TsFileProcessor tsFileProcessor = null;
+ try {
+ if (sequence) {
+ if (workSequenceTsFileProcessor == null) {
+ // create a new TsfileProcessor
+ workSequenceTsFileProcessor = createTsFileProcessor(true);
+ sequenceFileList.add(workSequenceTsFileProcessor.getTsFileResource());
+ }
+ tsFileProcessor = workSequenceTsFileProcessor;
+ } else {
+ if (workUnSequenceTsFileProcessor == null) {
+ // create a new TsfileProcessor
+ workUnSequenceTsFileProcessor = createTsFileProcessor(false);
+ unSequenceFileList.add(workUnSequenceTsFileProcessor.getTsFileResource());
+ }
+ tsFileProcessor = workUnSequenceTsFileProcessor;
+ }
+ } catch (DiskSpaceInsufficientException e) {
+ logger.error(
+ "disk space is insufficient when creating TsFile processor, change system mode to read-only",
+ e);
+ IoTDBDescriptor.getInstance().getConfig().setReadOnly(true);
+ } catch (IOException e) {
+ logger.error("meet IOException when creating TsFileProcessor, change system mode to read-only",
+ e);
+ IoTDBDescriptor.getInstance().getConfig().setReadOnly(true);
+ }
+ return tsFileProcessor;
+ }
+
private TsFileProcessor createTsFileProcessor(boolean sequence)
throws IOException, DiskSpaceInsufficientException {
String baseDir;
@@ -368,6 +445,7 @@ public class StorageGroupProcessor {
}
}
+
/**
* only called by insert(), thread-safety should be ensured by caller
*/
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index 34dcbac..8ddc6e6 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -44,12 +44,14 @@ import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.CloseTsFile
import org.apache.iotdb.db.engine.version.VersionController;
import org.apache.iotdb.db.exception.TsFileProcessorException;
import org.apache.iotdb.db.qp.constant.DatetimeUtils;
+import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.rescon.MemTablePool;
import org.apache.iotdb.db.utils.QueryUtils;
import org.apache.iotdb.db.writelog.manager.MultiFileLogNodeManager;
import org.apache.iotdb.db.writelog.node.WriteLogNode;
+import org.apache.iotdb.service.rpc.thrift.TS_StatusCode;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Pair;
@@ -134,10 +136,9 @@ public class TsFileProcessor {
public boolean insert(InsertPlan insertPlan) {
if (workMemTable == null) {
- // TODO change the impl of getAvailableMemTable to non-blocking
workMemTable = MemTablePool.getInstance().getAvailableMemTable(this);
- // no empty memtable, return failure
+ // no empty memtable, return failure, currently ,this will not happen
if (workMemTable == null) {
return false;
}
@@ -165,6 +166,56 @@ public class TsFileProcessor {
return true;
}
+ public boolean insertBatch(BatchInsertPlan batchInsertPlan, List<Integer> indexes,
+ List<Integer> results) {
+ if (workMemTable == null) {
+ workMemTable = MemTablePool.getInstance().getAvailableMemTable(this);
+
+ // no empty memtable, return, currently ,this will not happen
+ if (workMemTable == null) {
+ for (int index: indexes) {
+ results.set(index, TS_StatusCode.ERROR_STATUS.getValue());
+ }
+ return false;
+ }
+ }
+
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
+ try {
+ getLogNode().write(batchInsertPlan);
+ } catch (IOException e) {
+ logger.error("write WAL failed", e);
+ return false;
+ }
+ }
+
+ // update start time of this memtable
+ long startTime = Long.MAX_VALUE;
+ for (int index: indexes) {
+ if (batchInsertPlan.getTimes()[index] < startTime) {
+ startTime = batchInsertPlan.getTimes()[index];
+ }
+ }
+
+ tsFileResource.updateStartTime(batchInsertPlan.getDeviceId(), startTime);
+
+ //for sequence tsfile, we update the endTime only when the file is prepared to be closed.
+ //for unsequence tsfile, we have to update the endTime for each insertion.
+ if (!sequence) {
+ long endTime = Long.MIN_VALUE;
+ for (int index: indexes) {
+ if (batchInsertPlan.getTimes()[index] > endTime) {
+ endTime = batchInsertPlan.getTimes()[index];
+ }
+ }
+ tsFileResource.updateEndTime(batchInsertPlan.getDeviceId(), endTime);
+ }
+
+ // insert insertPlan to the work memtable
+ workMemTable.insertBatch(batchInsertPlan, indexes);
+ return true;
+ }
+
/**
* Delete data which belongs to the timeseries `deviceId.measurementId` and the timestamp of which
* <= 'timestamp' in the deletion. <br/>
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MGraph.java b/server/src/main/java/org/apache/iotdb/db/metadata/MGraph.java
index ac71b97..64d10d5 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/MGraph.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/MGraph.java
@@ -376,14 +376,12 @@ public class MGraph implements Serializable {
}
/**
- *
* @return storage group name -> the series number
*/
Map<String, Integer> countSeriesNumberInEachStorageGroup() throws PathErrorException {
Map<String, Integer> res = new HashMap<>();
Set<String> storageGroups = this.getAllStorageGroup();
for (String sg : storageGroups) {
- this.getNumSchemaMapForOneFileNode(sg);
MNode node = mtree.getNodeByPath(sg);
res.put(sg, node.getLeafCount());
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/IQueryProcessExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/IQueryProcessExecutor.java
index a6aefd6..c5f8473 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/IQueryProcessExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/IQueryProcessExecutor.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.exception.MetadataErrorException;
import org.apache.iotdb.db.exception.PathErrorException;
import org.apache.iotdb.db.exception.ProcessorException;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.query.context.QueryContext;
@@ -114,6 +115,12 @@ public interface IQueryProcessExecutor {
*/
boolean insert(InsertPlan insertPlan) throws ProcessorException;
+ /**
+ * execute batch insert plan
+ * @return result of each row
+ */
+ List<Integer> insertBatch(BatchInsertPlan batchInsertPlan) throws ProcessorException;
+
boolean judgePathExists(Path fullPath);
/**
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/QueryProcessExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/QueryProcessExecutor.java
index 6d5ee59..5439c12 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/QueryProcessExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/QueryProcessExecutor.java
@@ -46,6 +46,7 @@ import org.apache.iotdb.db.qp.logical.sys.AuthorOperator.AuthorType;
import org.apache.iotdb.db.qp.logical.sys.MetadataOperator;
import org.apache.iotdb.db.qp.logical.sys.PropertyOperator;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.qp.physical.crud.UpdatePlan;
@@ -56,6 +57,7 @@ import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.dataset.AuthDataSet;
import org.apache.iotdb.db.query.fill.IFill;
import org.apache.iotdb.db.utils.AuthUtils;
+import org.apache.iotdb.service.rpc.thrift.TSDataValueList;
import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -200,13 +202,11 @@ public class QueryProcessExecutor extends AbstractQueryProcessExecutor {
@Override
- public boolean insert(InsertPlan insertPlan)
- throws ProcessorException {
-
+ public boolean insert(InsertPlan insertPlan) throws ProcessorException {
try {
String[] measurementList = insertPlan.getMeasurements();
String deviceId = insertPlan.getDeviceId();
- MNode node = mManager.getNodeByDeviceIdFromCache(insertPlan.getDeviceId());
+ MNode node = mManager.getNodeByDeviceIdFromCache(deviceId);
String[] values = insertPlan.getValues();
TSDataType[] dataTypes = new TSDataType[measurementList.length];
@@ -235,6 +235,32 @@ public class QueryProcessExecutor extends AbstractQueryProcessExecutor {
}
@Override
+ public List<Integer> insertBatch(BatchInsertPlan batchInsertPlan) throws ProcessorException {
+ try {
+ String[] measurementList = batchInsertPlan.getMeasurements();
+ String deviceId = batchInsertPlan.getDeviceId();
+ MNode node = mManager.getNodeByDeviceIdFromCache(deviceId);
+
+ for (String s : measurementList) {
+ if (!node.hasChild(s)) {
+ throw new ProcessorException(
+ String.format("Current deviceId[%s] does not contains measurement:%s",
+ deviceId, s));
+ }
+ MNode measurementNode = node.getChild(s);
+ if (!measurementNode.isLeaf()) {
+ throw new ProcessorException(
+ String.format("Current Path is not leaf node. %s.%s", deviceId, s));
+ }
+ }
+ return storageEngine.insertBatch(batchInsertPlan);
+
+ } catch (PathErrorException | StorageEngineException e) {
+ throw new ProcessorException(e);
+ }
+ }
+
+ @Override
public List<String> getAllPaths(String originPath) throws MetadataErrorException {
return MManager.getInstance().getPaths(originPath);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
index 498b429..f18df87 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
@@ -67,7 +67,7 @@ public abstract class Operator {
*/
public enum OperatorType {
SFW, JOIN, UNION, FILTER, GROUPBY, ORDERBY, LIMIT, SELECT, SEQTABLESCAN, HASHTABLESCAN,
- MERGEJOIN, FILEREAD, NULL, TABLESCAN, UPDATE, INSERT, DELETE, BASIC_FUNC, QUERY, MERGEQUERY,
+ MERGEJOIN, FILEREAD, NULL, TABLESCAN, UPDATE, INSERT, BATCHINSERT, DELETE, BASIC_FUNC, QUERY, MERGEQUERY,
AGGREGATION, AUTHOR, FROM, FUNC, LOADDATA, METADATA, PROPERTY, INDEX, INDEXQUERY, FILL,
SET_STORAGE_GROUP, CREATE_TIMESERIES, DELETE_TIMESERIES, CREATE_USER, DELETE_USER, MODIFY_PASSWORD,
GRANT_USER_PRIVILEGE, REVOKE_USER_PRIVILEGE, GRANT_USER_ROLE, REVOKE_USER_ROLE, CREATE_ROLE,
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
index b60599f..4288539 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
@@ -23,6 +23,7 @@ import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.List;
import org.apache.iotdb.db.qp.logical.Operator;
+import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.tsfile.read.common.Path;
@@ -118,6 +119,10 @@ public abstract class PhysicalPlan {
plan = new DeletePlan();
plan.deserializeFrom(buffer);
break;
+ case BATCHINSERT:
+ plan = new BatchInsertPlan();
+ plan.deserializeFrom(buffer);
+ break;
default:
throw new IOException("unrecognized log type " + type);
}
@@ -126,7 +131,7 @@ public abstract class PhysicalPlan {
}
public enum PhysicalPlanType {
- INSERT, DELETE
+ INSERT, DELETE, BATCHINSERT
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
new file mode 100644
index 0000000..af20272
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
@@ -0,0 +1,233 @@
+package org.apache.iotdb.db.qp.physical.crud;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.utils.QueryDataSetUtils;
+import org.apache.iotdb.service.rpc.thrift.IoTDBDataType;
+import org.apache.iotdb.service.rpc.thrift.TSDataValueList;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.transport.TIOStreamTransport;
+import org.apache.thrift.transport.TTransport;
+
+public class BatchInsertPlan extends PhysicalPlan {
+
+ private String deviceId;
+ private String[] measurements;
+ private TSDataType[] dataTypes;
+ private Long[] times;
+ private TSDataValueList[] columns;
+ private int rowCount = 0;
+ private Long maxTime = null;
+ private Long minTime = null;
+
+ public BatchInsertPlan() {
+ super(false, OperatorType.BATCHINSERT);
+ }
+
+ public BatchInsertPlan(String deviceId, String[] measurements, List<IoTDBDataType> dataTypes) {
+ super(false, OperatorType.BATCHINSERT);
+ this.deviceId = deviceId;
+ this.measurements = measurements;
+ setDataTypes(dataTypes);
+ }
+
+ public BatchInsertPlan(String deviceId, String[] measurements, IoTDBDataType[] dataTypes) {
+ super(false, OperatorType.BATCHINSERT);
+ this.deviceId = deviceId;
+ this.measurements = measurements;
+ setDataTypes(dataTypes);
+ }
+
+
+ @Override
+ public List<Path> getPaths() {
+ List<Path> ret = new ArrayList<>();
+
+ for (String m : measurements) {
+ ret.add(new Path(deviceId, m));
+ }
+ return ret;
+ }
+
+ @Override
+ public void serializeTo(ByteBuffer buffer) {
+ int type = PhysicalPlanType.BATCHINSERT.ordinal();
+ buffer.put((byte) type);
+
+ putString(buffer, deviceId);
+
+ buffer.putInt(measurements.length);
+ for (String m : measurements) {
+ putString(buffer, m);
+ }
+
+ for (TSDataType dataType: dataTypes) {
+ buffer.putShort(dataType.serialize());
+ }
+
+ buffer.putInt(times.length);
+ for (Long time: times) {
+ buffer.putLong(time);
+ }
+
+ for (int i = 0; i < measurements.length; i++) {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ TTransport transport = new TIOStreamTransport(out);
+ TBinaryProtocol tp = new TBinaryProtocol(transport);
+ try {
+ columns[i].write(tp);
+ } catch (TException e) {
+ throw new RuntimeException("meet error when serializing BatchInsertPlan", e);
+ }
+ byte[] bytes = out.toByteArray();
+ buffer.putInt(bytes.length);
+ buffer.put(bytes);
+ }
+ }
+
+ @Override
+ public void deserializeFrom(ByteBuffer buffer) {
+ this.deviceId = readString(buffer);
+
+ int measurementSize = buffer.getInt();
+ this.measurements = new String[measurementSize];
+ for (int i = 0; i < measurementSize; i++) {
+ measurements[i] = readString(buffer);
+ }
+
+ this.dataTypes = new TSDataType[measurementSize];
+ for (int i = 0; i < measurementSize; i++) {
+ dataTypes[i] = TSDataType.deserialize(buffer.getShort());
+ }
+
+ int rows = buffer.getInt();
+ this.times = new Long[rows];
+ for (int i = 0; i < rows; i++) {
+ this.times[i] = buffer.getLong();
+ }
+
+ this.columns = new TSDataValueList[measurementSize];
+ for (int i = 0; i < measurementSize; i++) {
+ int size = buffer.getInt();
+ byte[] bytes = new byte[size];
+ buffer.get(bytes);
+ ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
+ TTransport transport = new TIOStreamTransport(bis);
+ TBinaryProtocol tp = new TBinaryProtocol(transport);
+ TSDataValueList tsDataValueList = new TSDataValueList();
+ try {
+ tsDataValueList.read(tp);
+ } catch (TException e) {
+ throw new RuntimeException("meet error when deserializing BatchInsertPlan", e);
+ }
+ columns[i] = tsDataValueList;
+ }
+ }
+
+ public String getDeviceId() {
+ return deviceId;
+ }
+
+ public void setDeviceId(String deviceId) {
+ this.deviceId = deviceId;
+ }
+
+ public String[] getMeasurements() {
+ return measurements;
+ }
+
+ public void setMeasurements(List<String> measurements) {
+ this.measurements = new String[measurements.size()];
+ measurements.toArray(this.measurements);
+ }
+
+ public void setMeasurements(String[] measurements) {
+ this.measurements = measurements;
+ }
+
+ public TSDataType[] getDataTypes() {
+ return dataTypes;
+ }
+
+ public void setDataTypes(List<IoTDBDataType> dataTypes) {
+ this.dataTypes = new TSDataType[dataTypes.size()];
+ for (int i = 0; i < dataTypes.size(); i++) {
+ this.dataTypes[i] = QueryDataSetUtils.getTSDataTypeByIoTDBDataType(dataTypes.get(i));
+ }
+ }
+
+ public void setDataTypes(IoTDBDataType[] dataTypes) {
+ this.dataTypes = new TSDataType[dataTypes.length];
+ for (int i = 0; i < dataTypes.length; i++) {
+ this.dataTypes[i] = QueryDataSetUtils.getTSDataTypeByIoTDBDataType(dataTypes[i]);
+ }
+ }
+
+ public TSDataValueList[] getColumns() {
+ return columns;
+ }
+
+ public void setColumns(List<TSDataValueList> columns) {
+ this.columns = new TSDataValueList[columns.size()];
+ columns.toArray(this.columns);
+ }
+
+ public long getMinTime() {
+ if (minTime != null) {
+ return minTime;
+ }
+ minTime = Long.MAX_VALUE;
+ for (Long time: times) {
+ if (time < minTime) {
+ minTime = time;
+ }
+ }
+ return minTime;
+ }
+
+ public long getMaxTime() {
+ if (maxTime != null) {
+ return maxTime;
+ }
+ long maxTime = Long.MIN_VALUE;
+ for (Long time: times) {
+ if (time > maxTime) {
+ maxTime = time;
+ }
+ }
+ return maxTime;
+ }
+
+ public Long[] getTimes() {
+ return times;
+ }
+
+ public void setTimes(List<Long> times) {
+ this.times = new Long[times.size()];
+ times.toArray(this.times);
+ }
+
+ public void setTimes(Long[] times) {
+ this.times = times;
+ }
+
+ public int getRowCount() {
+ return rowCount;
+ }
+
+ public void setRowCount(int size) {
+ this.rowCount = size;
+ }
+
+ public void setColumns(TSDataValueList[] columns) {
+ this.columns = columns;
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java b/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
index 13be7d3..f83591d 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/strategy/LogicalGenerator.java
@@ -533,10 +533,6 @@ public class LogicalGenerator {
if (filterOperator.getTokenIntType() == LESSTHAN) {
time = time - 1;
}
- // time must greater than 0 now
- if (time <= 0) {
- throw new LogicalOperatorException("delete Time:" + time + ", time must > 0");
- }
return time;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
index 29d1bc9..b41a1ef 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java
@@ -90,9 +90,6 @@ public class PhysicalGenerator {
case DELETE:
DeleteOperator delete = (DeleteOperator) operator;
paths = delete.getSelectedPaths();
- if (delete.getTime() <= 0) {
- throw new LogicalOperatorException("For Delete command, time must greater than 0.");
- }
return new DeletePlan(delete.getTime(), paths);
case INSERT:
InsertOperator insert = (InsertOperator) operator;
@@ -101,9 +98,6 @@ public class PhysicalGenerator {
throw new LogicalOperatorException(
"For Insert command, cannot specified more than one seriesPath:" + paths);
}
- if (insert.getTime() <= 0) {
- throw new LogicalOperatorException("For Insert command, time must greater than 0.");
- }
return new InsertPlan(paths.get(0).getFullPath(), insert.getTime(),
insert.getMeasurementList(),
insert.getValueList());
diff --git a/server/src/main/java/org/apache/iotdb/db/rescon/MemTablePool.java b/server/src/main/java/org/apache/iotdb/db/rescon/MemTablePool.java
index 210ee99..49188d1 100644
--- a/server/src/main/java/org/apache/iotdb/db/rescon/MemTablePool.java
+++ b/server/src/main/java/org/apache/iotdb/db/rescon/MemTablePool.java
@@ -42,6 +42,7 @@ public class MemTablePool {
private MemTablePool() {
}
+ // TODO change the impl of getAvailableMemTable to non-blocking
public IMemTable getAvailableMemTable(Object applier) {
synchronized (availableMemTables) {
if (availableMemTables.isEmpty() && size < CONFIG.getMaxMemtableNumber()) {
diff --git a/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
index 42f3f6d..4323c3a 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
@@ -80,10 +80,6 @@ public class IoTDB implements IoTDBMBean {
Runtime.getRuntime().addShutdownHook(new IoTDBShutdownHook());
setUncaughtExceptionHandler();
- boolean enableWAL = IoTDBDescriptor.getInstance().getConfig().isEnableWal();
- IoTDBDescriptor.getInstance().getConfig().setEnableWal(false);
- IoTDBDescriptor.getInstance().getConfig().setEnableWal(enableWAL);
-
// When registering statMonitor, we should start recovering some statistics
// with latest values stored
// Warn: registMonitor() method should be called after systemDataRecovery()
diff --git a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
index 531719f..1fa2103 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
@@ -61,12 +61,14 @@ import org.apache.iotdb.db.metadata.Metadata;
import org.apache.iotdb.db.qp.QueryProcessor;
import org.apache.iotdb.db.qp.executor.QueryProcessExecutor;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.QueryResourceManager;
import org.apache.iotdb.db.utils.QueryDataSetUtils;
import org.apache.iotdb.service.rpc.thrift.ServerProperties;
+import org.apache.iotdb.service.rpc.thrift.TSBatchInsertionReq;
import org.apache.iotdb.service.rpc.thrift.TSCancelOperationReq;
import org.apache.iotdb.service.rpc.thrift.TSCancelOperationResp;
import org.apache.iotdb.service.rpc.thrift.TSCloseOperationReq;
@@ -164,7 +166,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
tsStatus.setErrorMessage("login failed. Username or password is wrong.");
}
TSOpenSessionResp resp = new TSOpenSessionResp(tsStatus,
- TSProtocolVersion.TSFILE_SERVICE_PROTOCOL_V1);
+ TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V1);
resp.setSessionHandle(
new TS_SessionHandle(new TSHandleIdentifier(ByteBuffer.wrap(req.getUsername().getBytes()),
ByteBuffer.wrap(req.getPassword().getBytes()))));
@@ -973,7 +975,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
}
@Override
- public TSExecuteStatementResp executeInsertion(TSInsertionReq req) {
+ public TSExecuteStatementResp insert(TSInsertionReq req) {
if (!checkLogin()) {
logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME);
return getTSExecuteStatementResp(TS_StatusCode.ERROR_STATUS, ERROR_NOT_LOGIN);
@@ -1005,6 +1007,61 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
}
@Override
+ public TSExecuteBatchStatementResp insertBatch(TSBatchInsertionReq req) {
+ long t1 = System.currentTimeMillis();
+ List<Integer> results;
+ try {
+ if (!checkLogin()) {
+ logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME);
+ return getTSBathExecuteStatementResp(TS_StatusCode.ERROR_STATUS, ERROR_NOT_LOGIN, null);
+ }
+
+ BatchInsertPlan batchInsertPlan = new BatchInsertPlan();
+ batchInsertPlan.setDeviceId(req.deviceId);
+ batchInsertPlan.setMeasurements(req.measurements);
+ batchInsertPlan.setTimes(req.timestamps);
+ batchInsertPlan.setColumns(req.columns);
+ batchInsertPlan.setDataTypes(req.types);
+ batchInsertPlan.setRowCount(req.timestamps.size());
+
+ boolean isAllSuccessful = true;
+
+ List<Path> paths = batchInsertPlan.getPaths();
+ try {
+ if (!checkAuthorization(paths, batchInsertPlan)) {
+ return getTSBathExecuteStatementResp(TS_StatusCode.ERROR_STATUS,
+ "No permissions for this operation " + batchInsertPlan.getOperatorType(), null);
+ }
+ } catch (AuthException e) {
+ logger.error("meet error while checking authorization.", e);
+ return getTSBathExecuteStatementResp(TS_StatusCode.ERROR_STATUS,
+ "Uninitialized authorizer " + e.getMessage(), null);
+ }
+
+ results = processor.getExecutor().insertBatch(batchInsertPlan);
+
+ for (Integer result : results) {
+ long t2 = System.currentTimeMillis();
+ isAllSuccessful = isAllSuccessful && (result == TS_StatusCode.SUCCESS_STATUS.getValue() ? true : false);
+ Measurement.INSTANCE.addOperationLatency(Operation.EXECUTE_ONE_SQL_IN_BATCH, t2);
+ }
+
+ if (isAllSuccessful) {
+ return getTSBathExecuteStatementResp(TS_StatusCode.SUCCESS_STATUS,
+ "Execute batch statements successfully", results);
+ } else {
+ return getTSBathExecuteStatementResp(TS_StatusCode.ERROR_STATUS,
+ "Meet error in batch", results);
+ }
+ } catch (Exception e) {
+ logger.error("{}: error occurs when executing statements", IoTDBConstant.GLOBAL_DB_NAME, e);
+ return getTSBathExecuteStatementResp(TS_StatusCode.ERROR_STATUS, e.getMessage(), null);
+ } finally {
+ Measurement.INSTANCE.addOperationLatency(Operation.EXECUTE_BATCH, t1);
+ }
+ }
+
+ @Override
public long requestStatementId() {
return globalStmtId.incrementAndGet();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/MemUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/MemUtils.java
index 381d258..c7a243b 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/MemUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/MemUtils.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.utils;
import org.apache.iotdb.db.conf.IoTDBConstant;
+import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.write.record.TSRecord;
@@ -67,6 +68,33 @@ public class MemUtils {
return memSize;
}
+ public static long getRecordSize(BatchInsertPlan batchInsertPlan) {
+ long memSize = 0;
+ for (int i = 0; i < batchInsertPlan.getMeasurements().length; i++) {
+ switch (batchInsertPlan.getDataTypes()[i]) {
+ case INT32:
+ memSize += batchInsertPlan.getRowCount() * (8L + 4L); break;
+ case INT64:
+ memSize += batchInsertPlan.getRowCount() * (8L + 8L); break;
+ case FLOAT:
+ memSize += batchInsertPlan.getRowCount() * (8L + 4L); break;
+ case DOUBLE:
+ memSize += batchInsertPlan.getRowCount() * (8L + 8L); break;
+ case BOOLEAN:
+ memSize += batchInsertPlan.getRowCount() * (8L + 1L); break;
+ case TEXT:
+ memSize += batchInsertPlan.getRowCount() * 8L;
+ for (int j = 0; j < batchInsertPlan.getRowCount(); j++) {
+ memSize += batchInsertPlan.getColumns()[i].getBinary_vals().get(j).capacity();
+ }
+ break;
+ default:
+ memSize += batchInsertPlan.getRowCount() * (8L + 8L);
+ }
+ }
+ return memSize;
+ }
+
/**
* Calculate how much memory will be used if the given record is written to sequence file.
*/
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/QueryDataSetUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/QueryDataSetUtils.java
index 0cd742c..9d1569e 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/QueryDataSetUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/QueryDataSetUtils.java
@@ -18,10 +18,12 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
+import org.apache.iotdb.service.rpc.thrift.IoTDBDataType;
import org.apache.iotdb.service.rpc.thrift.TSDataValue;
import org.apache.iotdb.service.rpc.thrift.TSQueryDataSet;
import org.apache.iotdb.service.rpc.thrift.TSRowRecord;
import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
@@ -95,10 +97,34 @@ public class QueryDataSetUtils {
"data type %s is not supported when convert data at server",
f.getDataType().toString()));
}
- value.setType(f.getDataType().toString());
+ value.setType(getIoTDBDataTypeByTSDataType(f.getDataType()));
}
tsRowRecord.getValues().add(value);
}
return tsRowRecord;
}
+
+ public static IoTDBDataType getIoTDBDataTypeByTSDataType(TSDataType type) {
+ switch (type) {
+ case BOOLEAN: return IoTDBDataType.BOOLEAN;
+ case FLOAT: return IoTDBDataType.FLOAT;
+ case DOUBLE: return IoTDBDataType.DOUBLE;
+ case INT32: return IoTDBDataType.INT32;
+ case INT64: return IoTDBDataType.INT64;
+ case TEXT: return IoTDBDataType.TEXT;
+ default: throw new RuntimeException("data type not supported: " + type);
+ }
+ }
+
+ public static TSDataType getTSDataTypeByIoTDBDataType(IoTDBDataType type) {
+ switch (type) {
+ case BOOLEAN: return TSDataType.BOOLEAN;
+ case FLOAT: return TSDataType.FLOAT;
+ case DOUBLE: return TSDataType.DOUBLE;
+ case INT32: return TSDataType.INT32;
+ case INT64: return TSDataType.INT64;
+ case TEXT: return TSDataType.TEXT;
+ default: throw new RuntimeException("data type not supported: " + type);
+ }
+ }
}
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java
index f0ee8f0..c372b49 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java
@@ -21,9 +21,12 @@ package org.apache.iotdb.db.engine.storagegroup;
import org.apache.iotdb.db.engine.MetadataManagerHelper;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
import org.apache.iotdb.db.exception.StorageGroupProcessorException;
+import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.service.rpc.thrift.IoTDBDataType;
+import org.apache.iotdb.service.rpc.thrift.TSDataValueList;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.write.record.TSRecord;
import org.apache.iotdb.tsfile.write.record.datapoint.DataPoint;
@@ -74,6 +77,68 @@ public class StorageGroupProcessorTest {
}
}
+ @Test
+ public void testIoTDBRowBatchWriteAndSyncClose() {
+
+ String[] measurements = new String[2];
+ measurements[0] = "s0";
+ measurements[1] = "s1";
+ IoTDBDataType[] dataTypes = new IoTDBDataType[2];
+ dataTypes[0] = IoTDBDataType.INT32;
+ dataTypes[1] = IoTDBDataType.INT64;
+
+ BatchInsertPlan batchInsertPlan1 = new BatchInsertPlan("root.vehicle.d0", measurements, dataTypes);
+
+ Long[] times = new Long[100];
+ TSDataValueList[] dataValueLists = new TSDataValueList[2];
+ dataValueLists[0] = new TSDataValueList();
+ dataValueLists[0].setType(IoTDBDataType.INT32);
+ dataValueLists[1] = new TSDataValueList();
+ dataValueLists[1].setType(IoTDBDataType.INT64);
+ for (long i = 1; i <= 100; i++) {
+ times[(int) i-1] = i;
+ dataValueLists[0].addToInt_vals((int) i);
+ dataValueLists[1].addToLong_vals(i);
+ }
+
+ batchInsertPlan1.setTimes(times);
+ batchInsertPlan1.setColumns(dataValueLists);
+ batchInsertPlan1.setRowCount(times.length);
+
+ processor.insertBatch(batchInsertPlan1);
+ processor.putAllWorkingTsFileProcessorIntoClosingList();
+
+ BatchInsertPlan batchInsertPlan2 = new BatchInsertPlan("root.vehicle.d0", measurements, dataTypes);
+
+ times = new Long[100];
+ dataValueLists = new TSDataValueList[2];
+ dataValueLists[0] = new TSDataValueList();
+ dataValueLists[0].setType(IoTDBDataType.INT32);
+ dataValueLists[1] = new TSDataValueList();
+ dataValueLists[1].setType(IoTDBDataType.INT64);
+ for (long i = 50; i <= 149; i++) {
+ times[(int) i-50] = i;
+ dataValueLists[0].addToInt_vals((int) i);
+ dataValueLists[1].addToLong_vals(i);
+ }
+
+ batchInsertPlan2.setTimes(times);
+ batchInsertPlan2.setColumns(dataValueLists);
+ batchInsertPlan2.setRowCount(times.length);
+
+ processor.insertBatch(batchInsertPlan2);
+ processor.putAllWorkingTsFileProcessorIntoClosingList();
+ processor.waitForAllCurrentTsFileProcessorsClosed();
+
+ QueryDataSource queryDataSource = processor.query(deviceId, measurementId, context);
+
+ Assert.assertEquals(2, queryDataSource.getSeqResources().size());
+ Assert.assertEquals(1, queryDataSource.getUnseqResources().size());
+ for (TsFileResource resource : queryDataSource.getSeqResources()) {
+ Assert.assertTrue(resource.isClosed());
+ }
+ }
+
@Test
public void testSeqAndUnSeqSyncClose() throws StorageGroupProcessorException {
diff --git a/server/src/test/java/org/apache/iotdb/db/qp/utils/MemIntQpExecutor.java b/server/src/test/java/org/apache/iotdb/db/qp/utils/MemIntQpExecutor.java
index 13591b7..ba76baf 100644
--- a/server/src/test/java/org/apache/iotdb/db/qp/utils/MemIntQpExecutor.java
+++ b/server/src/test/java/org/apache/iotdb/db/qp/utils/MemIntQpExecutor.java
@@ -33,6 +33,7 @@ import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.qp.constant.SQLConstant;
import org.apache.iotdb.db.qp.executor.AbstractQueryProcessExecutor;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.qp.physical.crud.UpdatePlan;
@@ -194,6 +195,11 @@ public class MemIntQpExecutor extends AbstractQueryProcessExecutor {
}
@Override
+ public List<Integer> insertBatch(BatchInsertPlan batchInsertPlan) throws ProcessorException {
+ return null;
+ }
+
+ @Override
protected QueryDataSet processAuthorQuery(AuthorPlan plan, QueryContext context) {
return null;
}
diff --git a/service-rpc/src/main/thrift/rpc.thrift b/service-rpc/src/main/thrift/rpc.thrift
index 43d405e..b8d3d10 100644
--- a/service-rpc/src/main/thrift/rpc.thrift
+++ b/service-rpc/src/main/thrift/rpc.thrift
@@ -28,6 +28,10 @@ enum TS_StatusCode {
INVALID_HANDLE_STATUS
}
+enum IoTDBDataType {
+ BOOLEAN, INT32, INT64, FLOAT, DOUBLE, TEXT
+}
+
// The return status of a remote request
struct TS_Status {
1: required TS_StatusCode statusCode
@@ -79,7 +83,7 @@ struct TSExecuteStatementResp {
}
enum TSProtocolVersion {
- TSFILE_SERVICE_PROTOCOL_V1,
+ IOTDB_SERVICE_PROTOCOL_V1,
}
// Client-side handle to persistent session information on the server-side.
@@ -93,7 +97,7 @@ struct TSOpenSessionResp {
1: required TS_Status status
// The protocol version that the server is using.
- 2: required TSProtocolVersion serverProtocolVersion = TSProtocolVersion.TSFILE_SERVICE_PROTOCOL_V1
+ 2: required TSProtocolVersion serverProtocolVersion = TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V1
// Session Handle
3: optional TS_SessionHandle sessionHandle
@@ -105,7 +109,7 @@ struct TSOpenSessionResp {
// OpenSession()
// Open a session (connection) on the server against which operations may be executed.
struct TSOpenSessionReq {
- 1: required TSProtocolVersion client_protocol = TSProtocolVersion.TSFILE_SERVICE_PROTOCOL_V1
+ 1: required TSProtocolVersion client_protocol = TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V1
2: optional string username
3: optional string password
4: optional map<string, string> configuration
@@ -184,6 +188,7 @@ struct TSCloseOperationResp {
1: required TS_Status status
}
+
struct TSDataValue{
1: required bool is_empty
2: optional bool bool_val
@@ -192,7 +197,18 @@ struct TSDataValue{
5: optional double float_val
6: optional double double_val
7: optional binary binary_val
- 8: optional string type;
+ 8: optional IoTDBDataType type
+}
+
+
+struct TSDataValueList{
+ 1: required IoTDBDataType type
+ 2: optional list<bool> bool_vals
+ 3: optional list<i32> int_vals
+ 4: optional list<i64> long_vals
+ 5: optional list<double> float_vals
+ 6: optional list<double> double_vals
+ 7: optional list<binary> binary_vals
}
struct TSRowRecord{
@@ -259,6 +275,14 @@ struct TSInsertionReq {
5: required i64 stmtId
}
+struct TSBatchInsertionReq {
+ 1: required string deviceId
+ 2: required list<string> measurements
+ 3: required list<TSDataValueList> columns
+ 4: required list<i64> timestamps
+ 5: required list<IoTDBDataType> types
+}
+
struct ServerProperties {
1: required string version;
2: required list<string> supportedTimeAggregationOperations;
@@ -292,7 +316,9 @@ service TSIService {
ServerProperties getProperties();
- TSExecuteStatementResp executeInsertion(1:TSInsertionReq req);
+ TSExecuteStatementResp insert(1:TSInsertionReq req);
+
+ TSExecuteBatchStatementResp insertBatch(1:TSBatchInsertionReq req);
i64 requestStatementId();
}