You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2020/05/08 04:28:24 UTC
[incubator-iotdb] 01/01: premerge for the distributed version
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch cluster_premerge
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 9410ce29f84ad6ffb3923979783294e389384717
Author: jt2594838 <jt...@163.com>
AuthorDate: Fri May 8 12:27:47 2020 +0800
premerge for the distributed version
---
.../main/java/org/apache/iotdb/jdbc/Config.java | 2 +-
.../org/apache/iotdb/jdbc/IoTDBConnection.java | 7 +-
.../java/org/apache/iotdb/jdbc/IoTDBStatement.java | 9 +-
.../org/apache/iotdb/db/qp/strategy/SqlBase.g4 | 7 +-
.../iotdb/db/auth/authorizer/BasicAuthorizer.java | 60 ++-
.../iotdb/db/auth/authorizer/IAuthorizer.java | 186 ++++----
.../apache/iotdb/db/auth/entity/PathPrivilege.java | 26 +-
.../java/org/apache/iotdb/db/auth/entity/Role.java | 35 +-
.../java/org/apache/iotdb/db/auth/entity/User.java | 51 ++-
.../iotdb/db/auth/role/BasicRoleManager.java | 18 +
.../apache/iotdb/db/auth/role/IRoleManager.java | 28 +-
.../iotdb/db/auth/user/BasicUserManager.java | 20 +
.../apache/iotdb/db/auth/user/IUserManager.java | 94 ++--
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 17 +-
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 38 +-
.../org/apache/iotdb/db/engine/StorageEngine.java | 92 +++-
.../iotdb/db/engine/merge/task/MergeFileTask.java | 8 +-
.../db/engine/modification/ModificationFile.java | 40 +-
.../engine/storagegroup/StorageGroupProcessor.java | 119 +++--
.../iotdb/db/engine/storagegroup/TsFileLock.java | 109 +++++
.../db/engine/storagegroup/TsFileResource.java | 105 ++++-
.../iotdb/db/engine/upgrade/UpgradeTask.java | 6 +-
.../version/SimpleFileVersionController.java | 6 +
.../engine/version/SysTimeVersionController.java | 2 +
.../iotdb/db/engine/version/VersionController.java | 3 +
.../org/apache/iotdb/db/metadata/MManager.java | 159 +++++--
.../java/org/apache/iotdb/db/metadata/MTree.java | 2 +-
.../org/apache/iotdb/db/mqtt/PublishHandler.java | 2 +-
.../apache/iotdb/db/qp/constant/SQLConstant.java | 2 +
.../apache/iotdb/db/qp/executor/IPlanExecutor.java | 3 +-
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 394 ++++++++++-------
.../qp/logical/sys/LoadConfigurationOperator.java | 13 +-
.../apache/iotdb/db/qp/physical/PhysicalPlan.java | 127 +++++-
.../iotdb/db/qp/physical/crud/DeletePlan.java | 21 +-
.../iotdb/db/qp/physical/crud/InsertPlan.java | 18 +-
.../db/qp/physical/crud/InsertTabletPlan.java | 6 +-
.../iotdb/db/qp/physical/sys/AuthorPlan.java | 187 +++++++-
.../db/qp/physical/sys/CreateTimeSeriesPlan.java | 8 +-
.../iotdb/db/qp/physical/sys/DataAuthPlan.java | 54 ++-
.../db/qp/physical/sys/DeleteStorageGroupPlan.java | 36 ++
.../db/qp/physical/sys/DeleteTimeSeriesPlan.java | 48 +-
.../db/qp/physical/sys/LoadConfigurationPlan.java | 105 ++++-
.../db/qp/physical/sys/SetStorageGroupPlan.java | 4 +-
.../iotdb/db/qp/physical/sys/SetTTLPlan.java | 29 ++
.../db/qp/physical/sys/ShowChildPathsPlan.java | 2 +-
.../db/qp/physical/sys/ShowTimeSeriesPlan.java | 31 ++
.../iotdb/db/qp/strategy/LogicalGenerator.java | 8 +-
.../iotdb/db/qp/strategy/PhysicalGenerator.java | 28 +-
.../db/query/aggregation/AggregateResult.java | 7 +-
.../iotdb/db/query/control/FileReaderManager.java | 4 +-
.../db/query/dataset/AlignByDeviceDataSet.java | 18 +-
.../db/query/dataset/ShowTimeSeriesResult.java | 61 ++-
.../dataset/groupby/LocalGroupByExecutor.java | 10 +-
.../iotdb/db/query/executor/FillQueryExecutor.java | 10 +-
.../apache/iotdb/db/query/executor/fill/IFill.java | 11 +-
.../db/query/executor/fill/LastPointReader.java | 17 +-
.../iotdb/db/query/executor/fill/LinearFill.java | 39 +-
.../iotdb/db/query/reader/series/SeriesReader.java | 8 +-
.../org/apache/iotdb/db/service/JMXService.java | 2 +-
.../org/apache/iotdb/db/service/RPCService.java | 2 +
.../org/apache/iotdb/db/service/TSServiceImpl.java | 2 +-
.../org/apache/iotdb/db/utils/OpenFileNumUtil.java | 2 +-
.../org/apache/iotdb/db/utils/SerializeUtils.java | 492 +++++++++++++++++++++
.../org/apache/iotdb/db/utils/UpgradeUtils.java | 4 +-
.../writelog/manager/MultiFileLogNodeManager.java | 4 +-
.../db/writelog/node/ExclusiveWriteLogNode.java | 4 +-
.../iotdb/db/engine/storagegroup/TTLTest.java | 3 +-
.../iotdb/db/metadata/MManagerAdvancedTest.java | 48 +-
.../apache/iotdb/db/qp/plan/PhysicalPlanTest.java | 65 ++-
.../org/apache/iotdb/db/tools/WalCheckerTest.java | 4 +-
.../iotdb/db/writelog/io/LogWriterReaderTest.java | 2 +-
.../db/writelog/io/MultiFileLogReaderTest.java | 2 +-
.../java/org/apache/iotdb/rpc/TSStatusCode.java | 8 +
service-rpc/src/main/thrift/cluster.thrift | 386 ++++++++++++++++
.../java/org/apache/iotdb/session/Session.java | 9 +-
.../org/apache/iotdb/session/pool/SessionPool.java | 2 +-
.../apache/iotdb/session/pool/SessionPoolTest.java | 4 +-
.../apache/iotdb/tsfile/common/cache/LRUCache.java | 6 +-
.../apache/iotdb/tsfile/read/TimeValuePair.java | 1 -
.../iotdb/tsfile/utils/ReadWriteIOUtils.java | 9 +
80 files changed, 3041 insertions(+), 580 deletions(-)
diff --git a/jdbc/src/main/java/org/apache/iotdb/jdbc/Config.java b/jdbc/src/main/java/org/apache/iotdb/jdbc/Config.java
index 2cd4ff8..21f6197 100644
--- a/jdbc/src/main/java/org/apache/iotdb/jdbc/Config.java
+++ b/jdbc/src/main/java/org/apache/iotdb/jdbc/Config.java
@@ -52,6 +52,6 @@ public class Config {
public static final String JDBC_DRIVER_NAME = "org.apache.iotdb.jdbc.IoTDBDriver";
- public static boolean rpcThriftCompressionEnable = false;
+ public static boolean rpcThriftCompressionEnable = true;
}
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 0f5982c..ebf98e0 100644
--- a/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
+++ b/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
@@ -45,7 +45,9 @@ import org.apache.iotdb.service.rpc.thrift.*;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TCompactProtocol;
+import org.apache.thrift.transport.TFastFramedTransport;
import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,7 +62,7 @@ public class IoTDBConnection implements Connection {
private IoTDBConnectionParams params;
private boolean isClosed = true;
private SQLWarning warningChain = null;
- private TSocket transport;
+ private TTransport transport;
private ZoneId zoneId;
private boolean autoCommit;
@@ -404,7 +406,8 @@ public class IoTDBConnection implements Connection {
}
private void openTransport() throws TTransportException {
- transport = new TSocket(params.getHost(), params.getPort(), Config.connectionTimeoutInMs);
+ transport = new TFastFramedTransport(new TSocket(params.getHost(), params.getPort(),
+ Config.connectionTimeoutInMs));
if (!transport.isOpen()) {
transport.open();
}
diff --git a/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBStatement.java b/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBStatement.java
index cdd9a82..c26e03b 100644
--- a/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBStatement.java
+++ b/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBStatement.java
@@ -22,7 +22,14 @@ package org.apache.iotdb.jdbc;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.rpc.TSStatusCode;
-import org.apache.iotdb.service.rpc.thrift.*;
+import org.apache.iotdb.service.rpc.thrift.TSCancelOperationReq;
+import org.apache.iotdb.service.rpc.thrift.TSCloseOperationReq;
+import org.apache.iotdb.service.rpc.thrift.TSExecuteBatchStatementReq;
+import org.apache.iotdb.service.rpc.thrift.TSExecuteBatchStatementResp;
+import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementReq;
+import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp;
+import org.apache.iotdb.service.rpc.thrift.TSIService;
+import org.apache.iotdb.service.rpc.thrift.TSStatus;
import org.apache.thrift.TException;
import java.sql.*;
import java.time.ZoneId;
diff --git a/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4 b/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
index e75f4d1..d3feaab 100644
--- a/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
+++ b/server/src/main/antlr4/org/apache/iotdb/db/qp/strategy/SqlBase.g4
@@ -74,7 +74,7 @@ statement
| SHOW DEVICES prefixPath? #showDevices
| COUNT TIMESERIES prefixPath (GROUP BY LEVEL OPERATOR_EQ INT)? #countTimeseries
| COUNT NODES prefixPath LEVEL OPERATOR_EQ INT #countNodes
- | LOAD CONFIGURATION #loadConfigurationStatement
+ | LOAD CONFIGURATION (MINUS GLOBAL)? #loadConfigurationStatement
| LOAD FILE autoCreateSchema? #loadFiles
| REMOVE FILE #removeFile
| MOVE FILE FILE #moveFile
@@ -819,6 +819,11 @@ RENAME
: R E N A M E
;
+GLOBAL
+ : G L O B A L
+ | G
+ ;
+
FULL
: F U L L
;
diff --git a/server/src/main/java/org/apache/iotdb/db/auth/authorizer/BasicAuthorizer.java b/server/src/main/java/org/apache/iotdb/db/auth/authorizer/BasicAuthorizer.java
index 3659d2d..95ce406 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/authorizer/BasicAuthorizer.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/authorizer/BasicAuthorizer.java
@@ -18,8 +18,10 @@
*/
package org.apache.iotdb.db.auth.authorizer;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import org.apache.iotdb.db.auth.AuthException;
import org.apache.iotdb.db.auth.entity.PrivilegeType;
@@ -152,7 +154,7 @@ public abstract class BasicAuthorizer implements IAuthorizer, IService {
if (!PrivilegeType.isPathRelevant(privilegeId)) {
p = IoTDBConstant.PATH_ROOT;
}
- if(!roleManager.grantPrivilegeToRole(roleName, p, privilegeId)) {
+ if (!roleManager.grantPrivilegeToRole(roleName, p, privilegeId)) {
throw new AuthException(String.format("Role %s already has %s on %s", roleName,
PrivilegeType.values()[privilegeId], path));
}
@@ -186,7 +188,7 @@ public abstract class BasicAuthorizer implements IAuthorizer, IService {
}
} else {
throw new AuthException(String.format("User %s already has role %s",
- username, roleName));
+ username, roleName));
}
}
@@ -255,6 +257,50 @@ public abstract class BasicAuthorizer implements IAuthorizer, IService {
}
@Override
+ public Map<String, Boolean> getAllUserWaterMarkStatus() {
+ Map<String, Boolean> userWaterMarkStatus = new HashMap<>();
+ List<String> allUsers = listAllUsers();
+ for (String user : allUsers) {
+ try {
+ userWaterMarkStatus.put(user, isUserUseWaterMark(user));
+ } catch (AuthException e) {
+ logger.error(String.format("No such user : %s", user));
+ }
+ }
+ return userWaterMarkStatus;
+ }
+
+ @Override
+ public Map<String, User> getAllUsers() {
+ Map<String, User> allUsers = new HashMap<>();
+ List<String> userNames = listAllUsers();
+ for (String userName : userNames) {
+ try {
+ allUsers.put(userName, getUser(userName));
+ } catch (AuthException e) {
+ logger.error(String.format("get all users failed, No such user : %s", userName));
+ }
+ }
+ return allUsers;
+ }
+
+
+ @Override
+ public Map<String, Role> getAllRoles() {
+ Map<String, Role> allRoles = new HashMap<>();
+ List<String> roleNames = listAllRoles();
+ for (String roleName : roleNames) {
+ try {
+ allRoles.put(roleName, getRole(roleName));
+ } catch (AuthException e) {
+ logger.error(String.format("get all roles failed, No such role : %s", roleName));
+ }
+ }
+ return allRoles;
+ }
+
+
+ @Override
public void reset() throws AuthException {
init();
}
@@ -307,4 +353,14 @@ public abstract class BasicAuthorizer implements IAuthorizer, IService {
public void setUserUseWaterMark(String userName, boolean useWaterMark) throws AuthException {
userManager.setUserUseWaterMark(userName, useWaterMark);
}
+
+ @Override
+ public void replaceAllUsers(Map<String, User> users) throws AuthException {
+ userManager.replaceAllUsers(users);
+ }
+
+ @Override
+ public void replaceAllRoles(Map<String, Role> roles) throws AuthException {
+ roleManager.replaceAllRoles(roles);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/auth/authorizer/IAuthorizer.java b/server/src/main/java/org/apache/iotdb/db/auth/authorizer/IAuthorizer.java
index 331a79d..d60b96a 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/authorizer/IAuthorizer.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/authorizer/IAuthorizer.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.auth.authorizer;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import org.apache.iotdb.db.auth.AuthException;
import org.apache.iotdb.db.auth.entity.Role;
@@ -32,10 +33,8 @@ public interface IAuthorizer {
/**
* Login for a user.
*
- * @param username
- * The username of the user.
- * @param password
- * The password of the user.
+ * @param username The username of the user.
+ * @param password The password of the user.
* @return True if such user exists and the given password is correct, else return false.
* @throws AuthException if exception raised when searching for the user.
*/
@@ -44,53 +43,42 @@ public interface IAuthorizer {
/**
* Create a user with given username and password. New users will only be granted no privileges.
*
- * @param username
- * is not null or empty
- * @param password
- * is not null or empty
- * @throws AuthException
- * if the given username or password is illegal or the user already exists.
+ * @param username is not null or empty
+ * @param password is not null or empty
+ * @throws AuthException if the given username or password is illegal or the user already exists.
*/
void createUser(String username, String password) throws AuthException;
/**
* Delete a user.
*
- * @param username
- * the username of the user.
- * @throws AuthException
- * When attempting to delete the default administrator or the user does not exists.
+ * @param username the username of the user.
+ * @throws AuthException When attempting to delete the default administrator or the user does not
+ * exists.
*/
void deleteUser(String username) throws AuthException;
/**
* Grant a privilege on a seriesPath to a user.
*
- * @param username
- * The username of the user to which the privilege should be added.
- * @param path
- * The seriesPath on which the privilege takes effect. If the privilege is a
- * seriesPath-free privilege, this should be "root".
- * @param privilegeId
- * An integer that represents a privilege.
- * @throws AuthException
- * If the user does not exist or the privilege or the seriesPath is illegal or the permission already exists.
+ * @param username The username of the user to which the privilege should be added.
+ * @param path The seriesPath on which the privilege takes effect. If the privilege is a
+ * seriesPath-free privilege, this should be "root".
+ * @param privilegeId An integer that represents a privilege.
+ * @throws AuthException If the user does not exist or the privilege or the seriesPath is illegal
+ * or the permission already exists.
*/
void grantPrivilegeToUser(String username, String path, int privilegeId) throws AuthException;
/**
* Revoke a privilege on seriesPath from a user.
*
- * @param username
- * The username of the user from which the privilege should be removed.
- * @param path
- * The seriesPath on which the privilege takes effect. If the privilege is a
- * seriesPath-free privilege, this should be "root".
- * @param privilegeId
- * An integer that represents a privilege.
- * @throws AuthException
- * If the user does not exist or the privilege or the seriesPath is illegal or if
- * the permission does not exist.
+ * @param username The username of the user from which the privilege should be removed.
+ * @param path The seriesPath on which the privilege takes effect. If the privilege is a
+ * seriesPath-free privilege, this should be "root".
+ * @param privilegeId An integer that represents a privilege.
+ * @throws AuthException If the user does not exist or the privilege or the seriesPath is illegal
+ * or if the permission does not exist.
*/
void revokePrivilegeFromUser(String username, String path,
int privilegeId) throws AuthException;
@@ -98,8 +86,7 @@ public interface IAuthorizer {
/**
* Add a role.
*
- * @param roleName
- * the name of the role to be added.
+ * @param roleName the name of the role to be added.
* @throws AuthException if exception raised when adding the role or the role already exists.
*/
void createRole(String roleName) throws AuthException;
@@ -107,8 +94,7 @@ public interface IAuthorizer {
/**
* Delete a role.
*
- * @param roleName
- * the name of the role tobe deleted.
+ * @param roleName the name of the role tobe deleted.
* @throws AuthException if exception raised when deleting the role or the role does not exists.
*/
void deleteRole(String roleName) throws AuthException;
@@ -116,32 +102,24 @@ public interface IAuthorizer {
/**
* Add a privilege on a seriesPath to a role.
*
- * @param roleName
- * The name of the role to which the privilege is added.
- * @param path
- * The seriesPath on which the privilege takes effect. If the privilege is
- * a seriesPath-free privilege, this should be "root".
- * @param privilegeId
- * An integer that represents a privilege.
- * @throws AuthException
- * If the role does not exist or the privilege or the seriesPath is illegal or
- * the privilege already exists.
+ * @param roleName The name of the role to which the privilege is added.
+ * @param path The seriesPath on which the privilege takes effect. If the privilege is a
+ * seriesPath-free privilege, this should be "root".
+ * @param privilegeId An integer that represents a privilege.
+ * @throws AuthException If the role does not exist or the privilege or the seriesPath is illegal
+ * or the privilege already exists.
*/
void grantPrivilegeToRole(String roleName, String path, int privilegeId) throws AuthException;
/**
* Remove a privilege on a seriesPath from a role.
*
- * @param roleName
- * The name of the role from which the privilege is removed.
- * @param path
- * The seriesPath on which the privilege takes effect. If the privilege is a
- * seriesPath-free privilege, this should be "root".
- * @param privilegeId
- * An integer that represents a privilege.
- * @throws AuthException
- * If the role does not exist or the privilege or the seriesPath is illegal or the
- * privilege does not exists.
+ * @param roleName The name of the role from which the privilege is removed.
+ * @param path The seriesPath on which the privilege takes effect. If the privilege is a
+ * seriesPath-free privilege, this should be "root".
+ * @param privilegeId An integer that represents a privilege.
+ * @throws AuthException If the role does not exist or the privilege or the seriesPath is illegal
+ * or the privilege does not exists.
*/
void revokePrivilegeFromRole(String roleName, String path,
int privilegeId) throws AuthException;
@@ -149,35 +127,29 @@ public interface IAuthorizer {
/**
* Add a role to a user.
*
- * @param roleName
- * The name of the role to be added.
- * @param username
- * The name of the user to which the role is added.
- * @throws AuthException
- * If either the role or the user does not exist or the role already exists.
+ * @param roleName The name of the role to be added.
+ * @param username The name of the user to which the role is added.
+ * @throws AuthException If either the role or the user does not exist or the role already
+ * exists.
*/
void grantRoleToUser(String roleName, String username) throws AuthException;
/**
* Revoke a role from a user.
*
- * @param roleName
- * The name of the role to be removed.
- * @param username
- * The name of the user from which the role is removed.
- * @throws AuthException
- * If either the role or the user does not exist or the role already exists.
+ * @param roleName The name of the role to be removed.
+ * @param username The name of the user from which the role is removed.
+ * @throws AuthException If either the role or the user does not exist or the role already
+ * exists.
*/
void revokeRoleFromUser(String roleName, String username) throws AuthException;
/**
* Get the all the privileges of a user on a seriesPath.
*
- * @param username
- * The user whose privileges are to be queried.
- * @param path
- * The seriesPath on which the privileges take effect. If the privilege is
- * a seriesPath-free privilege, this should be "root".
+ * @param username The user whose privileges are to be queried.
+ * @param path The seriesPath on which the privileges take effect. If the privilege is a
+ * seriesPath-free privilege, this should be "root".
* @return A set of integers each present a privilege.
* @throws AuthException if exception raised when finding the privileges.
*/
@@ -186,28 +158,21 @@ public interface IAuthorizer {
/**
* Modify the password of a user.
*
- * @param username
- * The user whose password is to be modified.
- * @param newPassword
- * The new password.
- * @throws AuthException
- * If the user does not exists or the new password is illegal.
+ * @param username The user whose password is to be modified.
+ * @param newPassword The new password.
+ * @throws AuthException If the user does not exists or the new password is illegal.
*/
void updateUserPassword(String username, String newPassword) throws AuthException;
/**
* Check if the user have the privilege on the seriesPath.
*
- * @param username
- * The name of the user whose privileges are checked.
- * @param path
- * The seriesPath on which the privilege takes effect. If the privilege
- * is a seriesPath-free privilege, this should be "root".
- * @param privilegeId
- * An integer that represents a privilege.
+ * @param username The name of the user whose privileges are checked.
+ * @param path The seriesPath on which the privilege takes effect. If the privilege is a
+ * seriesPath-free privilege, this should be "root".
+ * @param privilegeId An integer that represents a privilege.
* @return True if the user has such privilege, false if the user does not have such privilege.
- * @throws AuthException
- * If the seriesPath or the privilege is illegal.
+ * @throws AuthException If the seriesPath or the privilege is illegal.
*/
boolean checkUserPrivileges(String username, String path, int privilegeId) throws AuthException;
@@ -218,18 +183,21 @@ public interface IAuthorizer {
/**
* List existing users in the database.
+ *
* @return A list contains all usernames.
*/
List<String> listAllUsers();
/**
* List existing roles in the database.
+ *
* @return A list contains all roleNames.
*/
List<String> listAllRoles();
/**
* Find a role by its name.
+ *
* @param roleName the name of the role.
* @return A role whose name is roleName or null if such role does not exist.
*/
@@ -237,6 +205,7 @@ public interface IAuthorizer {
/**
* Find a user by its name.
+ *
* @param username the name of the user.
* @return A user whose name is username or null if such user does not exist.
*/
@@ -244,6 +213,7 @@ public interface IAuthorizer {
/**
* Whether data water-mark is enabled for user 'userName'.
+ *
* @param userName
* @return
* @throws AuthException if the user does not exist
@@ -252,9 +222,47 @@ public interface IAuthorizer {
/**
* Enable or disable data water-mark for user 'userName'.
+ *
* @param userName
* @param useWaterMark
* @throws AuthException if the user does not exist.
*/
void setUserUseWaterMark(String userName, boolean useWaterMark) throws AuthException;
+
+ /**
+ * get all user water mark status
+ *
+ * @return key->userName, value->useWaterMark or not
+ */
+ Map<String, Boolean> getAllUserWaterMarkStatus();
+
+ /**
+ * get all user
+ *
+ * @return key-> userName, value->user
+ */
+ Map<String, User> getAllUsers();
+
+ /**
+ * get all role
+ *
+ * @return key->userName, value->role
+ */
+ Map<String, Role> getAllRoles();
+
+ /**
+ * clear all old users info, replace the old users with the new one
+ *
+ * @param users new users info
+ * @throws AuthException
+ */
+ void replaceAllUsers(Map<String, User> users) throws AuthException;
+
+ /**
+ * clear all old roles info, replace the old roles with the new one
+ *
+ * @param roles new roles info
+ * @throws AuthException
+ */
+ void replaceAllRoles(Map<String, Role> roles) throws AuthException;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/auth/entity/PathPrivilege.java b/server/src/main/java/org/apache/iotdb/db/auth/entity/PathPrivilege.java
index df51582..daf7206 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/entity/PathPrivilege.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/entity/PathPrivilege.java
@@ -18,11 +18,15 @@
*/
package org.apache.iotdb.db.auth.entity;
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.nio.ByteBuffer;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.iotdb.db.utils.SerializeUtils;
/**
* This class represents a privilege on a specific seriesPath. If the privilege is seriesPath-free,
@@ -46,7 +50,11 @@ public class PathPrivilege {
* Sort PathPrivilege by referenceCnt in descent order.
*/
public static final Comparator<PathPrivilege> REFERENCE_DESCENT_SORTER = (o1, o2) -> -Integer.
- compare(o1.referenceCnt.get(), o2.referenceCnt.get());
+ compare(o1.referenceCnt.get(), o2.referenceCnt.get());
+
+ public PathPrivilege() {
+
+ }
public PathPrivilege(String path) {
this.path = path;
@@ -103,4 +111,20 @@ public class PathPrivilege {
}
return builder.toString();
}
+
+ public ByteBuffer serialize() {
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
+
+ SerializeUtils.serialize(privileges, dataOutputStream);
+ SerializeUtils.serialize(path, dataOutputStream);
+
+ return ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
+ }
+
+ public void deserialize(ByteBuffer buffer) {
+ privileges = new HashSet<>();
+ SerializeUtils.deserialize(privileges, buffer);
+ path = SerializeUtils.deserializeString(buffer);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/auth/entity/Role.java b/server/src/main/java/org/apache/iotdb/db/auth/entity/Role.java
index 5ee6a5b..8d49959 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/entity/Role.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/entity/Role.java
@@ -18,11 +18,16 @@
*/
package org.apache.iotdb.db.auth.entity;
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import org.apache.iotdb.db.utils.AuthUtils;
+import org.apache.iotdb.db.utils.SerializeUtils;
/**
* This class contains all information of a role.
@@ -102,7 +107,35 @@ public class Role {
@Override
public int hashCode() {
-
return Objects.hash(name, privilegeList);
}
+
+ public ByteBuffer serialize() {
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
+
+ SerializeUtils.serialize(name, dataOutputStream);
+
+ try {
+ dataOutputStream.writeInt(privilegeList.size());
+ for (PathPrivilege pathPrivilege : privilegeList) {
+ dataOutputStream.write(pathPrivilege.serialize().array());
+ }
+ } catch (IOException e) {
+ // unreachable
+ }
+
+ return ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
+ }
+
+ public void deserialize(ByteBuffer buffer) {
+ name = SerializeUtils.deserializeString(buffer);
+ int privilegeListSize = buffer.getInt();
+ privilegeList = new ArrayList<>(privilegeListSize);
+ for (int i = 0; i < privilegeListSize; i++) {
+ PathPrivilege pathPrivilege = new PathPrivilege();
+ pathPrivilege.deserialize(buffer);
+ privilegeList.add(pathPrivilege);
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/auth/entity/User.java b/server/src/main/java/org/apache/iotdb/db/auth/entity/User.java
index 48ade95..b0c4b29 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/entity/User.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/entity/User.java
@@ -18,11 +18,16 @@
*/
package org.apache.iotdb.db.auth.entity;
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import org.apache.iotdb.db.utils.AuthUtils;
+import org.apache.iotdb.db.utils.SerializeUtils;
/**
* This class contains all information of a User.
@@ -49,7 +54,7 @@ public class User {
/**
* construct function for User.
*
- * @param name -user name
+ * @param name -user name
* @param password -user password
*/
public User(String name, String password) {
@@ -114,7 +119,7 @@ public class User {
/**
* set the privilege.
*
- * @param path -path
+ * @param path -path
* @param privileges -set of integer to determine privilege
*/
public void setPrivileges(String path, Set<Integer> privileges) {
@@ -152,12 +157,11 @@ public class User {
private boolean contentEquals(User user) {
return Objects.equals(name, user.name)
- && Objects.equals(password, user.password)
- && Objects.equals(privilegeList, user.privilegeList)
- && Objects.equals(roleList, user.roleList);
+ && Objects.equals(password, user.password)
+ && Objects.equals(privilegeList, user.privilegeList)
+ && Objects.equals(roleList, user.roleList);
}
-
@Override
public int hashCode() {
return Objects.hash(name, password, privilegeList, roleList, lastActiveTime);
@@ -170,4 +174,39 @@ public class User {
public void setUseWaterMark(boolean useWaterMark) {
this.useWaterMark = useWaterMark;
}
+
+ public ByteBuffer serialize() {
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
+
+ SerializeUtils.serialize(name, dataOutputStream);
+ SerializeUtils.serialize(password, dataOutputStream);
+
+ try {
+ dataOutputStream.writeInt(privilegeList.size());
+ for (PathPrivilege pathPrivilege : privilegeList) {
+ dataOutputStream.write(pathPrivilege.serialize().array());
+ }
+ dataOutputStream.writeBoolean(useWaterMark);
+ } catch (IOException e) {
+ // unreachable
+ }
+ SerializeUtils.serializeStringList(roleList, dataOutputStream);
+
+ return ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
+ }
+
+ public void deserialize(ByteBuffer buffer) {
+ name = SerializeUtils.deserializeString(buffer);
+ password = SerializeUtils.deserializeString(buffer);
+ int privilegeListSize = buffer.getInt();
+ privilegeList = new ArrayList<>(privilegeListSize);
+ for (int i = 0; i < privilegeListSize; i++) {
+ PathPrivilege pathPrivilege = new PathPrivilege();
+ pathPrivilege.deserialize(buffer);
+ privilegeList.add(pathPrivilege);
+ }
+ useWaterMark = buffer.get() == 1;
+ roleList = SerializeUtils.deserializeStringList(buffer);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/auth/role/BasicRoleManager.java b/server/src/main/java/org/apache/iotdb/db/auth/role/BasicRoleManager.java
index b7343a3..7da6d0f 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/role/BasicRoleManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/role/BasicRoleManager.java
@@ -23,6 +23,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Set;
import org.apache.iotdb.db.auth.AuthException;
import org.apache.iotdb.db.auth.entity.Role;
@@ -167,4 +168,21 @@ public abstract class BasicRoleManager implements IRoleManager {
rtlist.sort(null);
return rtlist;
}
+
+ @Override
+ public void replaceAllRoles(Map<String, Role> roles) throws AuthException {
+ synchronized (this) {
+ reset();
+ roleMap = roles;
+
+ for (Entry<String, Role> entry : roleMap.entrySet()) {
+ Role role = entry.getValue();
+ try {
+ accessor.saveRole(role);
+ } catch (IOException e) {
+ throw new AuthException(e);
+ }
+ }
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/auth/role/IRoleManager.java b/server/src/main/java/org/apache/iotdb/db/auth/role/IRoleManager.java
index 83c957c..8b34326 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/role/IRoleManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/role/IRoleManager.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.auth.role;
import java.util.List;
+import java.util.Map;
import org.apache.iotdb.db.auth.AuthException;
import org.apache.iotdb.db.auth.entity.Role;
@@ -58,11 +59,12 @@ public interface IRoleManager {
* Grant a privilege on a seriesPath to a role.
*
* @param rolename The rolename of the role to which the privilege should be added.
- * @param path The seriesPath on which the privilege takes effect. If the privilege
- * is a seriesPath-free privilege, this should be "root".
+ * @param path The seriesPath on which the privilege takes effect. If the privilege is a
+ * seriesPath-free privilege, this should be "root".
* @param privilegeId An integer that represents a privilege.
* @return True if the permission is successfully added, false if the permission already exists.
- * @throws AuthException If the role does not exist or the privilege or the seriesPath is illegal.
+ * @throws AuthException If the role does not exist or the privilege or the seriesPath is
+ * illegal.
*/
boolean grantPrivilegeToRole(String rolename, String path, int privilegeId) throws AuthException;
@@ -70,12 +72,13 @@ public interface IRoleManager {
* Revoke a privilege on seriesPath from a role.
*
* @param rolename The rolename of the role from which the privilege should be removed.
- * @param path The seriesPath on which the privilege takes effect. If the privilege
- * is a seriesPath-free privilege like 'CREATE_USER', this should be "root".
+ * @param path The seriesPath on which the privilege takes effect. If the privilege is a
+ * seriesPath-free privilege like 'CREATE_USER', this should be "root".
* @param privilegeId An integer that represents a privilege.
- * @return True if the permission is successfully revoked,
- * false if the permission does not exists.
- * @throws AuthException If the role does not exist or the privilege or the seriesPath is illegal.
+ * @return True if the permission is successfully revoked, false if the permission does not
+ * exists.
+ * @throws AuthException If the role does not exist or the privilege or the seriesPath is
+ * illegal.
*/
boolean revokePrivilegeFromRole(String rolename, String path,
int privilegeId) throws AuthException;
@@ -87,7 +90,16 @@ public interface IRoleManager {
/**
* List all roles in the database.
+ *
* @return A list that contains names of all roles.
*/
List<String> listAllRoles();
+
+ /**
+ * clear all old roles info, replace the old roles with the new one
+ *
+ * @param roles new roles info
+ * @throws AuthException
+ */
+ void replaceAllRoles(Map<String, Role> roles) throws AuthException;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/auth/user/BasicUserManager.java b/server/src/main/java/org/apache/iotdb/db/auth/user/BasicUserManager.java
index 3145ac7..299e9e4 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/user/BasicUserManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/user/BasicUserManager.java
@@ -23,6 +23,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Set;
import org.apache.iotdb.db.auth.AuthException;
import org.apache.iotdb.db.auth.entity.User;
@@ -310,4 +311,23 @@ public abstract class BasicUserManager implements IUserManager {
throw new AuthException(e);
}
}
+
+
+ @Override
+ public void replaceAllUsers(Map<String, User> users) throws AuthException {
+ synchronized (this) {
+ reset();
+ userMap = users;
+
+ for (Entry<String, User> entry : userMap.entrySet()) {
+ User user = entry.getValue();
+ try {
+ accessor.saveUser(user);
+ } catch (IOException e) {
+ throw new AuthException(e);
+ }
+ }
+ }
+ }
+
}
diff --git a/server/src/main/java/org/apache/iotdb/db/auth/user/IUserManager.java b/server/src/main/java/org/apache/iotdb/db/auth/user/IUserManager.java
index 4d7a3f3..5541bb3 100644
--- a/server/src/main/java/org/apache/iotdb/db/auth/user/IUserManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/auth/user/IUserManager.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.auth.user;
import java.util.List;
+import java.util.Map;
import org.apache.iotdb.db.auth.AuthException;
import org.apache.iotdb.db.auth.entity.User;
@@ -30,8 +31,7 @@ public interface IUserManager {
/**
* Get a user object.
*
- * @param username
- * The name of the user.
+ * @param username The name of the user.
* @return A user object whose name is username or null if such user does not exist.
* @throws AuthException if an exception is raised when interacting with the lower storage.
*/
@@ -40,57 +40,46 @@ public interface IUserManager {
/**
* Create a user with given username and password. New users will only be granted no privileges.
*
- * @param username
- * is not null or empty
- * @param password
- * is not null or empty
+ * @param username is not null or empty
+ * @param password is not null or empty
* @return True if the user is successfully created, false when the user already exists.
- * @throws AuthException
- * if the given username or password is illegal.
+ * @throws AuthException if the given username or password is illegal.
*/
boolean createUser(String username, String password) throws AuthException;
/**
* Delete a user.
*
- * @param username
- * the username of the user.
+ * @param username the username of the user.
* @return True if the user is successfully deleted, false if the user does not exists.
- * @throws AuthException
- * .
+ * @throws AuthException .
*/
boolean deleteUser(String username) throws AuthException;
/**
* Grant a privilege on a seriesPath to a user.
*
- * @param username
- * The username of the user to which the privilege should be added.
- * @param path
- * The seriesPath on which the privilege takes effect. If the privilege
- * is a seriesPath-free privilege, this should be "root".
- * @param privilegeId
- * An integer that represents a privilege.
+ * @param username The username of the user to which the privilege should be added.
+ * @param path The seriesPath on which the privilege takes effect. If the privilege is a
+ * seriesPath-free privilege, this should be "root".
+ * @param privilegeId An integer that represents a privilege.
* @return True if the permission is successfully added, false if the permission already exists.
- * @throws AuthException
- * If the user does not exist or the privilege or the seriesPath is illegal.
+ * @throws AuthException If the user does not exist or the privilege or the seriesPath is
+ * illegal.
*/
boolean grantPrivilegeToUser(String username, String path, int privilegeId) throws AuthException;
/**
* Revoke a privilege on seriesPath from a user.
*
- * @param username
- * The username of the user from which the privilege should be removed.
- * @param path
- * The seriesPath on which the privilege takes effect. If the privilege
- * is a seriesPath-free privilege, this should be "root".
- * @param privilegeId
- * An integer that represents a privilege.
- * @return True if the permission is successfully revoked,
- * false if the permission does not exists.
- * @throws AuthException
- * If the user does not exist or the privilege or the seriesPath is illegal.
+ * @param username The username of the user from which the privilege should be removed.
+ * @param path The seriesPath on which the privilege takes effect. If the privilege is a
+ * seriesPath-free privilege, this should be "root".
+ * @param privilegeId An integer that represents a privilege.
+ * @return True if the permission is successfully revoked, false if the permission does not
+ * exists.
+ * @throws AuthException If the user does not exist or the privilege or the seriesPath is
+ * illegal.
*/
boolean revokePrivilegeFromUser(String username, String path,
int privilegeId) throws AuthException;
@@ -98,39 +87,30 @@ public interface IUserManager {
/**
* Modify the password of a user.
*
- * @param username
- * The user whose password is to be modified.
- * @param newPassword
- * The new password.
+ * @param username The user whose password is to be modified.
+ * @param newPassword The new password.
* @return True if the password is successfully modified, false if the new password is illegal.
- * @throws AuthException
- * If the user does not exists.
+ * @throws AuthException If the user does not exists.
*/
boolean updateUserPassword(String username, String newPassword) throws AuthException;
/**
* Add a role to a user.
*
- * @param roleName
- * The name of the role to be added.
- * @param username
- * The name of the user to which the role is added.
+ * @param roleName The name of the role to be added.
+ * @param username The name of the user to which the role is added.
* @return True if the role is successfully added, false if the role already exists.
- * @throws AuthException
- * If the user does not exist.
+ * @throws AuthException If the user does not exist.
*/
boolean grantRoleToUser(String roleName, String username) throws AuthException;
/**
* Revoke a role from a user.
*
- * @param roleName
- * The name of the role to be removed.
- * @param username
- * The name of the user from which the role is removed.
+ * @param roleName The name of the role to be removed.
+ * @param username The name of the user from which the role is removed.
* @return True if the role is successfully removed, false if the role does not exist.
- * @throws AuthException
- * If the user does not exist.
+ * @throws AuthException If the user does not exist.
*/
boolean revokeRoleFromUser(String roleName, String username) throws AuthException;
@@ -141,12 +121,14 @@ public interface IUserManager {
/**
* List all users in the database.
+ *
* @return A list that contains all users'name.
*/
List<String> listAllUsers();
/**
* Whether data water-mark is enabled for user 'userName'.
+ *
* @param userName
* @return
* @throws AuthException if the user does not exist
@@ -155,9 +137,21 @@ public interface IUserManager {
/**
* Enable or disable data water-mark for user 'userName'.
+ *
* @param userName
* @param useWaterMark
* @throws AuthException if the user does not exist.
*/
void setUserUseWaterMark(String userName, boolean useWaterMark) throws AuthException;
+
+
+ /**
+ * clear all old users info, replace the old users with the new one
+ *
+ * @param users new users info
+ * @throws AuthException
+ */
+ void replaceAllUsers(Map<String, User> users) throws AuthException;
+
+
}
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 51bb3fc..bd1b66d 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -96,7 +96,7 @@ public class IoTDBConfig {
/**
* whether to use thrift compression.
*/
- private boolean rpcThriftCompressionEnable = false;
+ private boolean rpcThriftCompressionEnable = true;
/**
* Port which the JDBC server listens to.
@@ -280,6 +280,11 @@ public class IoTDBConfig {
private int mManagerCacheSize = 400000;
/**
+ * Cache size of {@code checkAndGetDataTypeCache} in {@link MManager}.
+ */
+ private int mRemoteSchemaCacheSize = 100000;
+
+ /**
* Is external sort enable.
*/
private boolean enableExternalSort = true;
@@ -887,11 +892,19 @@ public class IoTDBConfig {
this.mManagerCacheSize = mManagerCacheSize;
}
+ public int getmRemoteSchemaCacheSize() {
+ return mRemoteSchemaCacheSize;
+ }
+
+ public void setmRemoteSchemaCacheSize(int mRemoteSchemaCacheSize) {
+ this.mRemoteSchemaCacheSize = mRemoteSchemaCacheSize;
+ }
+
public boolean isSyncEnable() {
return isSyncEnable;
}
- void setSyncEnable(boolean syncEnable) {
+ public void setSyncEnable(boolean syncEnable) {
isSyncEnable = syncEnable;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 4389fa8..ce7b55d 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -69,7 +69,6 @@ public class IoTDBDescriptor {
boolean ok = parseCommandLine(options, params);
if (!ok) {
logger.error("replaces properties failed, use default conf params");
- return;
} else {
if (commandLine.hasOption(RPC_PORT)) {
conf.setRpcPort(Integer.parseInt(commandLine.getOptionValue(RPC_PORT)));
@@ -89,7 +88,7 @@ public class IoTDBDescriptor {
return true;
}
- private String getPropsUrl() {
+ public String getPropsUrl() {
String url = System.getProperty(IoTDBConstant.IOTDB_CONF, null);
if (url == null) {
url = System.getProperty(IoTDBConstant.IOTDB_HOME, null);
@@ -270,6 +269,10 @@ public class IoTDBDescriptor {
.parseInt(properties.getProperty("metadata_node_cache_size",
Integer.toString(conf.getmManagerCacheSize())).trim()));
+ conf.setmRemoteSchemaCacheSize(Integer
+ .parseInt(properties.getProperty("remote_schema_cache_size",
+ Integer.toString(conf.getmRemoteSchemaCacheSize())).trim()));
+
conf.setLanguageVersion(properties.getProperty("language_version",
conf.getLanguageVersion()).trim());
@@ -538,17 +541,9 @@ public class IoTDBDescriptor {
.toString(TSFileDescriptor.getInstance().getConfig().getMaxDegreeOfIndexNode()))));
}
- public void loadHotModifiedProps() throws QueryProcessException {
- String url = getPropsUrl();
- if (url == null) {
- return;
- }
-
- try (InputStream inputStream = new FileInputStream(new File(url))) {
- logger.info("Start to reload config file {}", url);
- Properties properties = new Properties();
- properties.load(inputStream);
-
+ public void loadHotModifiedProps(Properties properties)
+ throws QueryProcessException {
+ try {
// update data dirs
String dataDirs = properties.getProperty("data_dirs", null);
if (dataDirs != null) {
@@ -592,6 +587,23 @@ public class IoTDBDescriptor {
loadTsFileProps(properties);
} catch (Exception e) {
+ logger.warn("Fail to reload configuration ", e);
+ throw new QueryProcessException(
+ String.format("Fail to reload configuration because %s", e.getMessage()));
+ }
+ }
+
+ public void loadHotModifiedProps() throws QueryProcessException {
+ String url = getPropsUrl();
+ if (url == null) {
+ return;
+ }
+ try (InputStream inputStream = new FileInputStream(new File(url))) {
+ logger.info("Start to reload config file {}", url);
+ Properties properties = new Properties();
+ properties.load(inputStream);
+ loadHotModifiedProps(properties);
+ } catch (Exception e) {
logger.warn("Fail to reload config file {}", url, e);
throw new QueryProcessException(
String.format("Fail to reload config file %s because %s", url, e.getMessage()));
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 8fbf967..3c7a532 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
@@ -111,27 +111,13 @@ public class StorageEngine implements IService {
* TimestampPrecision
*/
@ServerConfigConsistent
- private static long timePartitionInterval;
+ private static long timePartitionInterval = -1;
private StorageEngine() {
logger = LoggerFactory.getLogger(StorageEngine.class);
systemDir = FilePathUtils.regularizePath(config.getSystemDir()) + "storage_groups";
// build time Interval to divide time partition
- String timePrecision = IoTDBDescriptor.getInstance().getConfig().getTimestampPrecision();
- switch (timePrecision) {
- case "ns":
- timePartitionInterval = IoTDBDescriptor.getInstance().
- getConfig().getPartitionInterval() * 1000_000_000L;
- break;
- case "us":
- timePartitionInterval = IoTDBDescriptor.getInstance().
- getConfig().getPartitionInterval() * 1000_000L;
- break;
- default:
- timePartitionInterval = IoTDBDescriptor.getInstance().
- getConfig().getPartitionInterval() * 1000;
- break;
- }
+ initTimePartition();
// create systemDir
try {
FileUtils.forceMkdir(SystemFileFactory.INSTANCE.getFile(systemDir));
@@ -166,6 +152,27 @@ public class StorageEngine implements IService {
}
}
+ private static void initTimePartition() {
+ timePartitionInterval = convertMilliWithPrecision(IoTDBDescriptor.getInstance().
+ getConfig().getPartitionInterval() * 1000L);
+ }
+
+ public static long convertMilliWithPrecision(long milliTime) {
+ long result = milliTime;
+ String timePrecision = IoTDBDescriptor.getInstance().getConfig().getTimestampPrecision();
+ switch (timePrecision) {
+ case "ns":
+ result = milliTime * 1000_000L;
+ break;
+ case "us":
+ result = milliTime * 1000L;
+ break;
+ default:
+ break;
+ }
+ return result;
+ }
+
@Override
public void start() {
ttlCheckThread = Executors.newSingleThreadScheduledExecutor();
@@ -259,7 +266,7 @@ public class StorageEngine implements IService {
}
/**
- * insert a BatchInsertPlan to a storage group
+ * insert a InsertTabletPlan to a storage group
*
* @return result of each row
*/
@@ -321,6 +328,41 @@ public class StorageEngine implements IService {
}
}
+ public void asyncCloseProcessor(String storageGroupName, long partitionId, boolean isSeq)
+ throws StorageGroupNotSetException {
+ StorageGroupProcessor processor = processorMap.get(storageGroupName);
+ if (processor != null) {
+ logger.info("async closing sg processor is called for closing {}, seq = {}, partitionId = {}",
+ storageGroupName, isSeq, partitionId);
+ processor.writeLock();
+ try {
+ if (isSeq) {
+ // to avoid concurrent modification problem, we need a new array list
+ for (TsFileProcessor tsfileProcessor : new ArrayList<>(
+ processor.getWorkSequenceTsFileProcessors())) {
+ if (tsfileProcessor.getTimeRangeId() == partitionId) {
+ processor.asyncCloseOneTsFileProcessor(true, tsfileProcessor);
+ break;
+ }
+ }
+ } else {
+ // to avoid concurrent modification problem, we need a new array list
+ for (TsFileProcessor tsfileProcessor : new ArrayList<>(
+ processor.getWorkUnsequenceTsFileProcessor())) {
+ if (tsfileProcessor.getTimeRangeId() == partitionId) {
+ processor.asyncCloseOneTsFileProcessor(false, tsfileProcessor);
+ break;
+ }
+ }
+ }
+ } finally {
+ processor.writeUnlock();
+ }
+ } else {
+ throw new StorageGroupNotSetException(storageGroupName);
+ }
+ }
+
/**
* update data.
*/
@@ -512,10 +554,24 @@ public class StorageEngine implements IService {
}
public static long getTimePartitionInterval() {
+ if (timePartitionInterval == -1) {
+ initTimePartition();
+ }
return timePartitionInterval;
}
public static long getTimePartition(long time) {
- return time / timePartitionInterval;
+ return time / getTimePartitionInterval();
+ }
+
+ /**
+ * Set the version of given partition to newMaxVersion if it is larger than the current version.
+ * @param storageGroup
+ * @param partitionId
+ * @param newMaxVersion
+ */
+ public void setPartitionVersionToMax(String storageGroup, long partitionId, long newMaxVersion)
+ throws StorageEngineException {
+ getProcessor(storageGroup).setPartitionFileVersionToMax(partitionId, newMaxVersion);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
index 006ec5c..15cfa6e 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
@@ -121,7 +121,7 @@ class MergeFileTask {
return;
}
- seqFile.getWriteQueryLock().writeLock().lock();
+ seqFile.writeLock();
try {
ChunkMetadataCache.getInstance().remove(seqFile);
FileReaderManager.getInstance().closeFileAndRemoveReader(seqFile.getPath());
@@ -184,7 +184,7 @@ class MergeFileTask {
}
throw e;
} finally {
- seqFile.getWriteQueryLock().writeLock().unlock();
+ seqFile.writeUnlock();
}
}
@@ -257,7 +257,7 @@ class MergeFileTask {
mergeLogger.logFileMergeEnd();
logger.debug("{} moved unmerged chunks of {} to the new file", taskName, seqFile);
- seqFile.getWriteQueryLock().writeLock().lock();
+ seqFile.writeLock();
try {
resource.removeFileReader(seqFile);
ChunkMetadataCache.getInstance().remove(seqFile);
@@ -274,7 +274,7 @@ class MergeFileTask {
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
- seqFile.getWriteQueryLock().writeLock().unlock();
+ seqFile.writeUnlock();
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/modification/ModificationFile.java b/server/src/main/java/org/apache/iotdb/db/engine/modification/ModificationFile.java
index 43bf6b0..b21d1aa 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/modification/ModificationFile.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/modification/ModificationFile.java
@@ -19,15 +19,21 @@
package org.apache.iotdb.db.engine.modification;
+import java.io.File;
import java.io.IOException;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
-
+import java.util.Random;
import org.apache.iotdb.db.engine.modification.io.LocalTextModificationAccessor;
import org.apache.iotdb.db.engine.modification.io.ModificationReader;
import org.apache.iotdb.db.engine.modification.io.ModificationWriter;
import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* ModificationFile stores the Modifications of a TsFile or unseq file in another file in the same
@@ -35,6 +41,7 @@ import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
*/
public class ModificationFile implements AutoCloseable {
+ private static final Logger logger = LoggerFactory.getLogger(ModificationFile.class);
public static final String FILE_SUFFIX = ".mods";
private List<Modification> modifications;
@@ -125,4 +132,35 @@ public class ModificationFile implements AutoCloseable {
FSFactoryProducer.getFSFactory().getFile(filePath).delete();
}
+ public boolean exists() {
+ return new File(filePath).exists();
+ }
+
+ /**
+ * Create a hardlink for the modification file.
+ * The hardlink with have a suffix like ".{sysTime}_{randomLong}"
+ * @return a new ModificationFile with its path changed to the hardlink, or null if the origin
+ * file does not exist or the hardlink cannot be created.
+ */
+ public ModificationFile createHardlink() {
+ if (!exists()) {
+ return null;
+ }
+
+ Random random = new Random();
+ while (true) {
+ String hardlinkSuffix = "." + System.currentTimeMillis() + "_" + random.nextLong();
+ File hardlink = new File(filePath + hardlinkSuffix);
+
+ try {
+ Files.createLink(Paths.get(hardlink.getAbsolutePath()), Paths.get(filePath));
+ return new ModificationFile(hardlink.getAbsolutePath());
+ } catch (FileAlreadyExistsException e) {
+ // retry a different name if the file is already created
+ } catch (IOException e) {
+ logger.error("Cannot create hardlink for {}", filePath, e);
+ return null;
+ }
+ }
+ }
}
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 d559482..2e42b50 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
@@ -574,8 +574,7 @@ public class StorageGroupProcessor {
* @param timePartitionId time partition id
*/
private void insertTabletToTsFileProcessor(InsertTabletPlan insertTabletPlan,
- int start, int end, boolean sequence, TSStatus[] results, long timePartitionId)
- throws WriteProcessException {
+ int start, int end, boolean sequence, TSStatus[] results, long timePartitionId) {
// return when start >= end
if (start >= end) {
return;
@@ -673,9 +672,10 @@ public class StorageGroupProcessor {
for (int i = 0; i < measurementList.length; i++) {
// Update cached last value with high priority
MNode measurementNode = node.getChild(measurementList[i]);
-
- ((LeafMNode) measurementNode)
- .updateCachedLast(plan.composeTimeValuePair(i), true, latestFlushedTime);
+ if (measurementNode != null) {
+ ((LeafMNode) measurementNode)
+ .updateCachedLast(plan.composeTimeValuePair(i), true, latestFlushedTime);
+ }
}
} catch (MetadataException | QueryProcessException e) {
throw new WriteProcessException(e);
@@ -963,7 +963,7 @@ public class StorageGroupProcessor {
}
// ensure that the file is not used by any queries
- if (resource.getWriteQueryLock().writeLock().tryLock()) {
+ if (resource.tryWriteLock()) {
try {
// physical removal
resource.remove();
@@ -977,7 +977,7 @@ public class StorageGroupProcessor {
unSequenceFileList.remove(resource);
}
} finally {
- resource.getWriteQueryLock().writeLock().unlock();
+ resource.writeUnlock();
}
}
} finally {
@@ -1422,11 +1422,11 @@ public class StorageGroupProcessor {
}
for (TsFileResource unseqFile : unseqFiles) {
- unseqFile.getWriteQueryLock().writeLock().lock();
+ unseqFile.writeLock();
try {
unseqFile.remove();
} finally {
- unseqFile.getWriteQueryLock().writeLock().unlock();
+ unseqFile.writeUnlock();
}
}
}
@@ -1479,13 +1479,23 @@ public class StorageGroupProcessor {
for (int i = 0; i < seqFiles.size(); i++) {
TsFileResource seqFile = seqFiles.get(i);
- while (!seqFile.getWriteQueryLock().writeLock().tryLock() || !mergeLock.writeLock()
- .tryLock()) {
- if (seqFile.getWriteQueryLock().writeLock().isHeldByCurrentThread()) {
- seqFile.getWriteQueryLock().writeLock().unlock();
- }
- if(mergeLock.writeLock().isHeldByCurrentThread()) {
- mergeLock.writeLock().unlock();
+ // get both seqFile lock and merge lock
+ boolean fileLockGot;
+ boolean mergeLockGot;
+ while (true) {
+ fileLockGot = seqFile.tryWriteLock();
+ mergeLockGot = mergeLock.writeLock().tryLock();
+
+ if (fileLockGot && mergeLockGot) {
+ break;
+ } else {
+ // did not get all of them, release the gotten one and retry
+ if (fileLockGot) {
+ seqFile.writeUnlock();
+ }
+ if(mergeLockGot) {
+ mergeLock.writeLock().unlock();
+ }
}
}
try {
@@ -1498,7 +1508,7 @@ public class StorageGroupProcessor {
}
} finally {
mergeLock.writeLock().unlock();
- seqFile.getWriteQueryLock().writeLock().unlock();
+ seqFile.writeUnlock();
}
}
logger.info("{} a merge task ends", storageGroupName);
@@ -1604,6 +1614,14 @@ public class StorageGroupProcessor {
}
/**
+ * Set the version in "partition" to "version" if "version" is larger than the current version.
+ * @param partition
+ * @param version
+ */
+ public void setPartitionFileVersionToMax(long partition, long version) {
+ }
+
+ /**
* Find the position of "newTsFileResource" in the sequence files if it can be inserted into them.
* @param newTsFileResource
* @param newFilePartitionId
@@ -1702,30 +1720,42 @@ public class StorageGroupProcessor {
closeQueryLock.writeLock().lock();
try {
Iterator<TsFileResource> iterator = sequenceFileTreeSet.iterator();
- removeFullyOverlapFiles(resource, iterator);
+ removeFullyOverlapFiles(resource, iterator, true);
iterator = unSequenceFileList.iterator();
- removeFullyOverlapFiles(resource, iterator);
+ removeFullyOverlapFiles(resource, iterator, false);
} finally {
closeQueryLock.writeLock().unlock();
writeUnlock();
}
}
- private void removeFullyOverlapFiles(TsFileResource resource, Iterator<TsFileResource> iterator) {
+ private void removeFullyOverlapFiles(TsFileResource resource, Iterator<TsFileResource> iterator
+ , boolean isSeq) {
while (iterator.hasNext()) {
TsFileResource seqFile = iterator.next();
if (resource.getHistoricalVersions().containsAll(seqFile.getHistoricalVersions())
&& !resource.getHistoricalVersions().equals(seqFile.getHistoricalVersions())
- && seqFile.getWriteQueryLock().writeLock().tryLock()) {
+ && seqFile.tryWriteLock()) {
try {
+ if (!seqFile.isClosed()) {
+ // also remove the TsFileProcessor if the overlapped file is not closed
+ long timePartition = seqFile.getTimePartition();
+ Map<Long, TsFileProcessor> fileProcessorMap = isSeq ? workSequenceTsFileProcessors :
+ workUnsequenceTsFileProcessors;
+ TsFileProcessor tsFileProcessor = fileProcessorMap.get(timePartition);
+ if (tsFileProcessor != null && tsFileProcessor.getTsFileResource() == seqFile) {
+ tsFileProcessor.syncClose();
+ fileProcessorMap.remove(timePartition);
+ }
+ }
iterator.remove();
seqFile.remove();
} catch (Exception e) {
logger.error("Something gets wrong while removing FullyOverlapFiles ", e);
throw e;
} finally {
- seqFile.getWriteQueryLock().writeLock().unlock();
+ seqFile.writeUnlock();
}
}
}
@@ -1940,12 +1970,12 @@ public class StorageGroupProcessor {
if (tsFileResourceToBeDeleted == null) {
return false;
}
- tsFileResourceToBeDeleted.getWriteQueryLock().writeLock().lock();
+ tsFileResourceToBeDeleted.writeLock();
try {
tsFileResourceToBeDeleted.remove();
logger.info("Delete tsfile {} successfully.", tsFileResourceToBeDeleted.getFile());
} finally {
- tsFileResourceToBeDeleted.getWriteQueryLock().writeLock().unlock();
+ tsFileResourceToBeDeleted.writeUnlock();
}
return true;
}
@@ -1998,14 +2028,14 @@ public class StorageGroupProcessor {
if (tsFileResourceToBeMoved == null) {
return false;
}
- tsFileResourceToBeMoved.getWriteQueryLock().writeLock().lock();
+ tsFileResourceToBeMoved.writeLock();
try {
tsFileResourceToBeMoved.moveTo(targetDir);
logger
.info("Move tsfile {} to target dir {} successfully.", tsFileResourceToBeMoved.getFile(),
targetDir.getPath());
} finally {
- tsFileResourceToBeMoved.getWriteQueryLock().writeLock().unlock();
+ tsFileResourceToBeMoved.writeUnlock();
}
return true;
}
@@ -2042,9 +2072,42 @@ public class StorageGroupProcessor {
return storageGroupName;
}
+ /**
+ * Check if the data of "tsFileResource" all exist locally by comparing the historical versions
+ * in the partition of "partitionNumber". This is available only when the IoTDB which generated
+ * "tsFileResource" has the same close file policy as the local one.
+ * If one of the version in "tsFileResource" equals to a version of a working file, false is
+ * also returned because "tsFileResource" may have unwritten data of that file.
+ * @param tsFileResource
+ * @param partitionNum
+ * @return true if the historicalVersions of "tsFileResource" is a subset of
+ * partitionDirectFileVersions, or false if it is not a subset and it does not contain any
+ * version of a working file
+ */
public boolean isFileAlreadyExist(TsFileResource tsFileResource, long partitionNum) {
- return partitionDirectFileVersions.getOrDefault(partitionNum, Collections.emptySet())
- .containsAll(tsFileResource.getHistoricalVersions());
+ // consider the case: The local node crashes when it is writing TsFile no.5.
+ // when it restarts, the leader has proceeded to no.6. When the leader sends no.5 to this
+ // node, the file should be accepted as local no.5 is not closed which means there may be
+ // unreceived data in no.5
+ // So if the incoming file contains the version of an unclosed file, it should be accepted
+ for (TsFileProcessor workSequenceTsFileProcessor : getWorkSequenceTsFileProcessors()) {
+ long workingFileVersion = workSequenceTsFileProcessor.getTsFileResource().getMaxVersion();
+ if (tsFileResource.getHistoricalVersions().contains(workingFileVersion)) {
+ return false;
+ }
+ }
+ for (TsFileProcessor workUnsequenceTsFileProcessor : getWorkUnsequenceTsFileProcessor()) {
+ long workingFileVersion = workUnsequenceTsFileProcessor.getTsFileResource().getMaxVersion();
+ if (tsFileResource.getHistoricalVersions().contains(workingFileVersion)) {
+ return false;
+ }
+ }
+ Set<Long> partitionFileVersions = partitionDirectFileVersions
+ .getOrDefault(partitionNum, Collections.emptySet());
+ // TODO-Cluter: change to debug
+ logger.info("FileVersions/PartitionVersions: {}/{}", tsFileResource.getHistoricalVersions(),
+ partitionFileVersions);
+ return partitionFileVersions.containsAll(tsFileResource.getHistoricalVersions());
}
@FunctionalInterface
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileLock.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileLock.java
new file mode 100644
index 0000000..5b22c3c
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileLock.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.engine.storagegroup;
+
+/**
+ * TsFileLock is a special read-write lock that can be locked by one thread but unlocked by
+ * another thread.
+ *
+ * When the query thread crashes, the locks it used can no longer be unlocked by itself and
+ * another thread must do the unlock for it. Unfortunately, normal Java ReentrantLock can only be
+ * unlocked by the thread which locked it, so we have to use a different implementation.
+ * It is also interesting to use a third-party implementation and if you do find a better one,
+ * please submit an issue on our website.
+ *
+ * WARNING: as the lock holder is not recorded, the caller must assure that lock() and unlock()
+ * match, i.e., if you only call lock() once then do not call unlock() more than once and vice
+ * versa.
+ */
+public class TsFileLock {
+
+ private volatile int readCnt;
+ private volatile int writeCnt;
+
+ public void readLock() {
+ synchronized (this) {
+ while (writeCnt > 0) {
+ try {
+ this.wait(1);
+ } catch (InterruptedException e) {
+ // ignore
+ }
+ }
+
+ readCnt ++;
+ }
+ }
+
+ public void readUnlock() {
+ synchronized (this) {
+ if (readCnt > 0) {
+ readCnt --;
+ this.notifyAll();
+ }
+ }
+ }
+
+ public void writeLock() {
+ synchronized (this) {
+ while (writeCnt > 0 || readCnt > 0) {
+ try {
+ this.wait(1);
+ } catch (InterruptedException e) {
+ // ignore
+ }
+ }
+
+ writeCnt ++;
+ }
+ }
+
+ public void writUnlock() {
+ synchronized (this) {
+ if (writeCnt > 0) {
+ writeCnt --;
+ this.notifyAll();
+ }
+ }
+ }
+
+ public boolean tryWriteLock() {
+ synchronized (this) {
+ if (writeCnt > 0 || readCnt > 0) {
+ return false;
+ }
+
+ writeCnt ++;
+ return true;
+ }
+ }
+
+ public boolean tryReadLock() {
+ synchronized (this) {
+ if (writeCnt > 0) {
+ return false;
+ }
+
+ readCnt ++;
+ return true;
+ }
+ }
+
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
index 4b96339..d60cf38 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
@@ -22,6 +22,9 @@ import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -30,9 +33,10 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
+import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
+
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.modification.ModificationFile;
@@ -92,6 +96,8 @@ public class TsFileResource {
// same file generation policy but have their own merge policies.
private Set<Long> historicalVersions;
+ private TsFileLock tsFileLock = new TsFileLock();
+
/**
* Chunk metadata list of unsealed tsfile. Only be set in a temporal TsFileResource in a query
* process.
@@ -108,8 +114,6 @@ public class TsFileResource {
*/
private TimeseriesMetadata timeSeriesMetadata;
- private ReentrantReadWriteLock writeQueryLock = new ReentrantReadWriteLock();
-
private FSFactory fsFactory = FSFactoryProducer.getFSFactory();
public TsFileResource() {
@@ -127,7 +131,7 @@ public class TsFileResource {
this.chunkMetadataList = other.chunkMetadataList;
this.readOnlyMemChunk = other.readOnlyMemChunk;
generateTimeSeriesMetadata();
- this.writeQueryLock = other.writeQueryLock;
+ this.tsFileLock = other.tsFileLock;
this.fsFactory = other.fsFactory;
this.historicalVersions = other.historicalVersions;
}
@@ -168,18 +172,19 @@ public class TsFileResource {
}
private void generateTimeSeriesMetadata() throws IOException {
- if (chunkMetadataList.isEmpty() && readOnlyMemChunk.isEmpty()) {
+ if ((chunkMetadataList == null ||chunkMetadataList.isEmpty())
+ && (readOnlyMemChunk == null || readOnlyMemChunk.isEmpty())) {
timeSeriesMetadata = null;
}
timeSeriesMetadata = new TimeseriesMetadata();
timeSeriesMetadata.setOffsetOfChunkMetaDataList(-1);
timeSeriesMetadata.setDataSizeOfChunkMetaDataList(-1);
- if (!chunkMetadataList.isEmpty()) {
+ if (!(chunkMetadataList == null ||chunkMetadataList.isEmpty())) {
timeSeriesMetadata.setMeasurementId(chunkMetadataList.get(0).getMeasurementUid());
TSDataType dataType = chunkMetadataList.get(0).getDataType();
timeSeriesMetadata.setTSDataType(dataType);
- } else if (!readOnlyMemChunk.isEmpty()) {
+ } else if (!(readOnlyMemChunk == null || readOnlyMemChunk.isEmpty())) {
timeSeriesMetadata.setMeasurementId(readOnlyMemChunk.get(0).getMeasurementUid());
TSDataType dataType = readOnlyMemChunk.get(0).getDataType();
timeSeriesMetadata.setTSDataType(dataType);
@@ -222,6 +227,10 @@ public class TsFileResource {
ReadWriteIOUtils.write(historicalVersion, outputStream);
}
}
+
+ if (modFile != null && modFile.exists()) {
+ ReadWriteIOUtils.write(modFile.getFilePath(), outputStream);
+ }
}
File src = fsFactory.getFile(file + RESOURCE_SUFFIX + TEMP_SUFFIX);
File dest = fsFactory.getFile(file + RESOURCE_SUFFIX);
@@ -260,6 +269,10 @@ public class TsFileResource {
long version = Long.parseLong(file.getName().split(IoTDBConstant.TSFILE_NAME_SEPARATOR)[1]);
historicalVersions = Collections.singleton(version);
}
+
+ if (inputStream.available() > 0) {
+ modFile = new ModificationFile(ReadWriteIOUtils.readString(inputStream));
+ }
}
}
@@ -346,8 +359,29 @@ public class TsFileResource {
return processor;
}
- public ReentrantReadWriteLock getWriteQueryLock() {
- return writeQueryLock;
+
+ public void writeLock() {
+ tsFileLock.writeLock();
+ }
+
+ public void writeUnlock() {
+ tsFileLock.writUnlock();
+ }
+
+ public void readLock() {
+ tsFileLock.readLock();
+ }
+
+ public void readUnlock() {
+ tsFileLock.readUnlock();
+ }
+
+ public boolean tryReadLock() {
+ return tsFileLock.tryReadLock();
+ }
+
+ public boolean tryWriteLock() {
+ return tsFileLock.tryWriteLock();
}
void doUpgrade() {
@@ -528,4 +562,57 @@ public class TsFileResource {
}
return partitionId;
}
+
+ /**
+ * Create a hardlink for the TsFile and modification file (if exists)
+ * The hardlink with have a suffix like ".{sysTime}_{randomLong}"
+ * @return a new TsFileResource with its file changed to the hardlink or null the hardlink
+ * cannot be created.
+ */
+ public TsFileResource createHardlink() {
+ if (!file.exists()) {
+ return null;
+ }
+
+ TsFileResource newResource;
+ try {
+ newResource = new TsFileResource(this);
+ } catch (IOException e) {
+ logger.error("Cannot create hardlink for {}", file, e);
+ return null;
+ }
+
+ Random random = new Random();
+ while (true) {
+ String hardlinkSuffix = "." + System.currentTimeMillis() + "_" + random.nextLong();
+ File hardlink = new File(file.getAbsolutePath() + hardlinkSuffix);
+
+ try {
+ Files.createLink(Paths.get(hardlink.getAbsolutePath()), Paths.get(file.getAbsolutePath()));
+ newResource.setFile(hardlink);
+ if (modFile != null && modFile.exists()) {
+ newResource.setModFile(modFile.createHardlink());
+ }
+ break;
+ } catch (FileAlreadyExistsException e) {
+ // retry a different name if the file is already created
+ } catch (IOException e) {
+ logger.error("Cannot create hardlink for {}", file, e);
+ return null;
+ }
+ }
+ return newResource;
+ }
+
+ public void setModFile(ModificationFile modFile) {
+ this.modFile = modFile;
+ }
+
+ public long getMaxVersion() {
+ long maxVersion = 0;
+ if (historicalVersions != null) {
+ maxVersion = Collections.max(historicalVersions);
+ }
+ return maxVersion;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/upgrade/UpgradeTask.java b/server/src/main/java/org/apache/iotdb/db/engine/upgrade/UpgradeTask.java
index d13e982..3430b75 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/upgrade/UpgradeTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/upgrade/UpgradeTask.java
@@ -40,13 +40,13 @@ public class UpgradeTask extends WrappedRunnable {
@Override
public void runMayThrow() {
try {
- upgradeResource.getWriteQueryLock().readLock().lock();
+ upgradeResource.readLock();
String tsfilePathBefore = upgradeResource.getFile().getAbsolutePath();
String tsfilePathAfter = UpgradeUtils.getUpgradeFileName(upgradeResource.getFile());
UpgradeLog.writeUpgradeLogFile(
tsfilePathBefore + COMMA_SEPERATOR + UpgradeCheckStatus.BEGIN_UPGRADE_FILE);
- upgradeResource.getWriteQueryLock().writeLock().lock();
+ upgradeResource.writeLock();
try {
FSFactoryProducer.getFSFactory().getFile(tsfilePathBefore).delete();
FSFactoryProducer.getFSFactory()
@@ -56,7 +56,7 @@ public class UpgradeTask extends WrappedRunnable {
tsfilePathBefore + COMMA_SEPERATOR + UpgradeCheckStatus.UPGRADE_SUCCESS);
FSFactoryProducer.getFSFactory().getFile(tsfilePathAfter).getParentFile().delete();
} finally {
- upgradeResource.getWriteQueryLock().writeLock().unlock();
+ upgradeResource.writeUnlock();
}
UpgradeSevice.setCntUpgradeFileNum(UpgradeSevice.getCntUpgradeFileNum() - 1);
logger.info("Upgrade completes, file path:{} , the remaining upgraded file num: {}",
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/version/SimpleFileVersionController.java b/server/src/main/java/org/apache/iotdb/db/engine/version/SimpleFileVersionController.java
index 95c14b8..edbde9e 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/version/SimpleFileVersionController.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/version/SimpleFileVersionController.java
@@ -54,6 +54,12 @@ public class SimpleFileVersionController implements VersionController {
restore();
}
+ public SimpleFileVersionController(String directoryPath)
+ throws IOException {
+ this.directoryPath = directoryPath;
+ restore();
+ }
+
public static long getSaveInterval() {
return saveInterval;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/version/SysTimeVersionController.java b/server/src/main/java/org/apache/iotdb/db/engine/version/SysTimeVersionController.java
index 3e47cd9..233c3a13 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/version/SysTimeVersionController.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/version/SysTimeVersionController.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.db.engine.version;
+import java.io.IOException;
+
/**
* SysTimeVersionController uses system timestamp as the version number.
*/
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/version/VersionController.java b/server/src/main/java/org/apache/iotdb/db/engine/version/VersionController.java
index 30c93c6..c982299 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/version/VersionController.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/version/VersionController.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.db.engine.version;
+import java.io.IOException;
+
/**
* VersionController controls the version(a monotonic increasing long) of a FileNode.
*/
@@ -34,4 +36,5 @@ public interface VersionController {
* @return the current version number.
*/
long currVersion();
+
}
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
index f378135..c44778d 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
@@ -18,7 +18,29 @@
*/
package org.apache.iotdb.db.metadata;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.conf.adapter.ActiveTimeSeriesCounter;
import org.apache.iotdb.db.conf.adapter.IoTDBConfigDynamicAdapter;
@@ -37,6 +59,7 @@ import org.apache.iotdb.db.qp.physical.sys.ShowTimeSeriesPlan;
import org.apache.iotdb.db.query.dataset.ShowTimeSeriesResult;
import org.apache.iotdb.db.utils.RandomDeleteCache;
import org.apache.iotdb.db.utils.TestOnly;
+import org.apache.iotdb.tsfile.common.cache.LRUCache;
import org.apache.iotdb.tsfile.exception.cache.CacheException;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -46,13 +69,6 @@ import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.FileReader;
-import java.io.IOException;
-import java.util.*;
-import java.util.Map.Entry;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
/**
* This class takes the responsibility of serialization of all the metadata info and persistent it
@@ -74,6 +90,7 @@ public class MManager {
private boolean writeToLog;
// device -> DeviceMNode
private RandomDeleteCache<String, MNode> mNodeCache;
+ private LRUCache<String, MeasurementSchema> mRemoteSchemaCache;
// tag key -> tag value -> LeafMNode
private Map<String, Map<String, Set<LeafMNode>>> tagIndex = new HashMap<>();
@@ -83,6 +100,7 @@ public class MManager {
private long maxSeriesNumberAmongStorageGroup;
private boolean initialized;
private IoTDBConfig config;
+ private ByteBuffer partitionTableBuffer;
private static class MManagerHolder {
@@ -110,21 +128,33 @@ public class MManager {
writeToLog = false;
int cacheSize = config.getmManagerCacheSize();
- mNodeCache =
- new RandomDeleteCache<String, MNode>(cacheSize) {
-
- @Override
- public MNode loadObjectByKey(String key) throws CacheException {
- lock.readLock().lock();
- try {
- return mtree.getNodeByPathWithStorageGroupCheck(key);
- } catch (MetadataException e) {
- throw new CacheException(e);
- } finally {
- lock.readLock().unlock();
- }
- }
- };
+ mNodeCache = new RandomDeleteCache<String, MNode>(cacheSize) {
+
+ @Override
+ public MNode loadObjectByKey(String key) throws CacheException {
+ lock.readLock().lock();
+ try {
+ return mtree.getNodeByPathWithStorageGroupCheck(key);
+ } catch (MetadataException e) {
+ throw new CacheException(e);
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+ };
+
+ int remoteCacheSize = config.getmRemoteSchemaCacheSize();
+ mRemoteSchemaCache = new LRUCache<String, MeasurementSchema>(remoteCacheSize) {
+ @Override
+ protected MeasurementSchema loadObjectByKey(String key) throws IOException {
+ throw new IOException();
+ }
+
+ @Override
+ public synchronized void removeItem(String key) {
+ cache.keySet().removeIf(s -> s.startsWith(key));
+ }
+ };
}
public static MManager getInstance() {
@@ -366,6 +396,10 @@ public class MManager {
*/
public Pair<Set<String>, String> deleteTimeseries(String prefixPath) throws MetadataException {
lock.writeLock().lock();
+
+ // clear cached schema
+ mRemoteSchemaCache.removeItem(prefixPath);
+
if (isStorageGroup(prefixPath)) {
if (config.isEnableParameterAdapter()) {
@@ -507,6 +541,9 @@ public class MManager {
lock.writeLock().lock();
try {
for (String storageGroup : storageGroups) {
+ // clear cached schema
+ mRemoteSchemaCache.removeItem(storageGroup);
+
// try to delete storage group
List<LeafMNode> leafMNodes = mtree.deleteStorageGroup(storageGroup);
for (LeafMNode leafMNode : leafMNodes) {
@@ -565,6 +602,14 @@ public class MManager {
if (path.equals(SQLConstant.RESERVED_TIME)) {
return TSDataType.INT64;
}
+
+ try {
+ MeasurementSchema schema = mRemoteSchemaCache.get(path);
+ return schema.getType();
+ } catch (IOException e) {
+ // ignore
+ }
+
return mtree.getSchema(path).getType();
} finally {
lock.readLock().unlock();
@@ -630,7 +675,7 @@ public class MManager {
*
* @return storage group in the given path
*/
- public String getStorageGroupName(String path) throws MetadataException {
+ public String getStorageGroupName(String path) throws StorageGroupNotSetException {
lock.readLock().lock();
try {
return mtree.getStorageGroupName(path);
@@ -799,12 +844,34 @@ public class MManager {
}
}
- public MeasurementSchema getSeriesSchema(String device, String measuremnet)
+ public MeasurementSchema getSeriesSchema(String device, String measurement)
throws MetadataException {
lock.readLock().lock();
try {
InternalMNode node = (InternalMNode) mtree.getNodeByPath(device);
- return ((LeafMNode) node.getChild(measuremnet)).getSchema();
+ MNode leaf = node.getChild(measurement);
+ if (leaf != null) {
+ return ((LeafMNode) leaf).getSchema();
+ } else {
+ try {
+ return mRemoteSchemaCache
+ .get(device + IoTDBConstant.PATH_SEPARATOR + measurement);
+ } catch (IOException e) {
+ throw new PathNotExistException(device + IoTDBConstant.PATH_SEPARATOR + measurement);
+ }
+ }
+ } catch (PathNotExistException e) {
+ try {
+ MeasurementSchema measurementSchema = mRemoteSchemaCache
+ .get(device + IoTDBConstant.PATH_SEPARATOR + measurement);
+ if (measurementSchema != null) {
+ return measurementSchema;
+ } else {
+ throw e;
+ }
+ } catch (IOException ex) {
+ throw e;
+ }
} finally {
lock.readLock().unlock();
}
@@ -961,7 +1028,27 @@ public class MManager {
}
/**
- * change or set the new offset of a timeseries
+ * get all storageGroups ttl
+ *
+ * @return key-> storageGroupName, value->ttl
+ */
+ public Map<String, Long> getStorageGroupsTTL() {
+ Map<String, Long> storageGroupsTTL = new HashMap<>();
+ try {
+ List<String> storageGroups = this.getAllStorageGroupNames();
+ for (String storageGroup : storageGroups) {
+ long ttl = getStorageGroupNode(storageGroup).getDataTTL();
+ storageGroupsTTL.put(storageGroup, ttl);
+ }
+ } catch (MetadataException e) {
+ logger.error("get storage groups ttl failed.", e);
+ }
+ return storageGroupsTTL;
+ }
+
+ /**
+ * Check whether the given path contains a storage group change or set the new offset of a
+ * timeseries
*
* @param path timeseries
* @param offset offset in the tag file
@@ -1349,6 +1436,13 @@ public class MManager {
}
}
+ /**
+ * Collect the timeseries schemas under "startingPath". Notice the measurements in the collected
+ * MeasurementSchemas are the full path here.
+ *
+ * @param startingPath
+ * @param timeseriesSchemas
+ */
public void collectSeries(String startingPath, List<MeasurementSchema> timeseriesSchemas) {
MNode mNode;
try {
@@ -1391,4 +1485,17 @@ public class MManager {
lock.readLock().unlock();
}
}
+
+ public void cacheSchema(String path, MeasurementSchema schema) {
+ // check schema is in local
+ try {
+ ShowTimeSeriesPlan tempPlan = new ShowTimeSeriesPlan(new Path(path), false, null, null, 0, 0);
+ List<String[]> schemas = mtree.getAllMeasurementSchema(tempPlan);
+ if (schemas.isEmpty()) {
+ mRemoteSchemaCache.put(path, schema);
+ }
+ } catch (MetadataException e) {
+ mRemoteSchemaCache.put(path, schema);
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java b/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java
index ffa7bbb..3dbaeea 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java
@@ -469,7 +469,7 @@ public class MTree implements Serializable {
*
* @return storage group in the given path
*/
- String getStorageGroupName(String path) throws MetadataException {
+ String getStorageGroupName(String path) throws StorageGroupNotSetException {
String[] nodes = MetaUtils.getNodeNames(path);
MNode cur = root;
for (int i = 1; i < nodes.length; i++) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mqtt/PublishHandler.java b/server/src/main/java/org/apache/iotdb/db/mqtt/PublishHandler.java
index fcbaef2..4020901 100644
--- a/server/src/main/java/org/apache/iotdb/db/mqtt/PublishHandler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mqtt/PublishHandler.java
@@ -79,7 +79,7 @@ public class PublishHandler extends AbstractInterceptHandler {
return;
}
- // since device ids from messages maybe different, so we use the InsertPlan not BatchInsertPlan.
+ // since device ids from messages maybe different, so we use the InsertPlan not InsertTabletPlan.
for (Message event : events) {
if (event == null) {
continue;
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java b/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
index e7258a8..ea2ca86 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
@@ -138,6 +138,8 @@ public class SQLConstant {
public static final int TOK_COUNT_TIMESERIES = 77;
public static final int TOK_COUNT_NODE_TIMESERIES = 78;
public static final int TOK_COUNT_NODES = 79;
+ public static final int TOK_LOAD_CONFIGURATION_GLOBAL = 80;
+ public static final int TOK_LOAD_CONFIGURATION_LOCAL = 81;
public static final int TOK_METADATA_ALTER = 80;
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
index 7e70ed2..24a2629 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
@@ -33,6 +33,7 @@ import org.apache.iotdb.service.rpc.thrift.TSStatus;
import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.thrift.TException;
public interface IPlanExecutor {
@@ -44,7 +45,7 @@ public interface IPlanExecutor {
*/
QueryDataSet processQuery(PhysicalPlan queryPlan, QueryContext context)
throws IOException, StorageEngineException,
- QueryFilterOptimizationException, QueryProcessException, MetadataException, SQLException;
+ QueryFilterOptimizationException, QueryProcessException, MetadataException, SQLException, TException, InterruptedException;
/**
* Process Non-Query Physical plan, including insert/update/delete operation of
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
index 92b71e5..57b6d8f 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
@@ -50,6 +50,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
+
import org.apache.iotdb.db.auth.AuthException;
import org.apache.iotdb.db.auth.authorizer.IAuthorizer;
import org.apache.iotdb.db.auth.authorizer.LocalFileAuthorizer;
@@ -99,6 +100,7 @@ import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.DataAuthPlan;
import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan;
import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.LoadConfigurationPlan;
import org.apache.iotdb.db.qp.physical.sys.FlushPlan;
import org.apache.iotdb.db.qp.physical.sys.MergePlan;
import org.apache.iotdb.db.qp.physical.sys.OperateFilePlan;
@@ -163,7 +165,7 @@ public class PlanExecutor implements IPlanExecutor {
@Override
public QueryDataSet processQuery(PhysicalPlan queryPlan, QueryContext context)
- throws IOException, StorageEngineException, QueryFilterOptimizationException,
+ throws IOException, StorageEngineException, QueryFilterOptimizationException,
QueryProcessException, MetadataException {
if (queryPlan instanceof QueryPlan) {
return processDataQuery((QueryPlan) queryPlan, context);
@@ -223,7 +225,7 @@ public class PlanExecutor implements IPlanExecutor {
operateTTL((SetTTLPlan) plan);
return true;
case LOAD_CONFIGURATION:
- IoTDBDescriptor.getInstance().loadHotModifiedProps();
+ loadConfiguration((LoadConfigurationPlan) plan);
return true;
case LOAD_FILES:
operateLoadFiles((OperateFilePlan) plan);
@@ -245,7 +247,7 @@ public class PlanExecutor implements IPlanExecutor {
return true;
default:
throw new UnsupportedOperationException(
- String.format("operation %s is not supported", plan.getOperatorType()));
+ String.format("operation %s is not supported", plan.getOperatorType()));
}
}
@@ -276,13 +278,12 @@ public class PlanExecutor implements IPlanExecutor {
}
protected QueryDataSet processDataQuery(QueryPlan queryPlan, QueryContext context)
- throws StorageEngineException, QueryFilterOptimizationException, QueryProcessException,
+ throws StorageEngineException, QueryFilterOptimizationException, QueryProcessException,
IOException {
QueryDataSet queryDataSet;
if (queryPlan instanceof AlignByDevicePlan) {
- queryDataSet = new AlignByDeviceDataSet((AlignByDevicePlan) queryPlan, context, queryRouter);
+ queryDataSet = getAlignByDeviceDataSet((AlignByDevicePlan) queryPlan, context, queryRouter);
} else {
-
if (queryPlan.getPaths() == null || queryPlan.getPaths().isEmpty()) {
// no time series are selected, return EmptyDataSet
return new EmptyDataSet();
@@ -309,8 +310,13 @@ public class PlanExecutor implements IPlanExecutor {
return queryDataSet;
}
- private QueryDataSet processShowQuery(ShowPlan showPlan)
- throws QueryProcessException, MetadataException {
+ protected AlignByDeviceDataSet getAlignByDeviceDataSet(AlignByDevicePlan plan,
+ QueryContext context, IQueryRouter router) {
+ return new AlignByDeviceDataSet(plan, context, queryRouter);
+ }
+
+ protected QueryDataSet processShowQuery(ShowPlan showPlan)
+ throws QueryProcessException, MetadataException {
switch (showPlan.getShowContentType()) {
case TTL:
return processShowTTLQuery((ShowTTLPlan) showPlan);
@@ -344,13 +350,14 @@ public class PlanExecutor implements IPlanExecutor {
}
}
- private QueryDataSet processCountNodes(CountPlan countPlan) throws MetadataException {
+ private QueryDataSet processCountNodes(CountPlan countPlan)
+ throws MetadataException {
List<String> nodes = getNodesList(countPlan.getPath().toString(), countPlan.getLevel());
int num = nodes.size();
SingleDataSet singleDataSet =
- new SingleDataSet(
- Collections.singletonList(new Path(COLUMN_COUNT)),
- Collections.singletonList(TSDataType.INT32));
+ new SingleDataSet(
+ Collections.singletonList(new Path(COLUMN_COUNT)),
+ Collections.singletonList(TSDataType.INT32));
Field field = new Field(TSDataType.INT32);
field.setIntV(num);
RowRecord record = new RowRecord(0);
@@ -359,12 +366,13 @@ public class PlanExecutor implements IPlanExecutor {
return singleDataSet;
}
- private QueryDataSet processCountNodeTimeSeries(CountPlan countPlan) throws MetadataException {
+ private QueryDataSet processCountNodeTimeSeries(CountPlan countPlan)
+ throws MetadataException {
List<String> nodes = getNodesList(countPlan.getPath().toString(), countPlan.getLevel());
ListDataSet listDataSet =
- new ListDataSet(
- Arrays.asList(new Path(COLUMN_COLUMN), new Path(COLUMN_COUNT)),
- Arrays.asList(TSDataType.TEXT, TSDataType.TEXT));
+ new ListDataSet(
+ Arrays.asList(new Path(COLUMN_COLUMN), new Path(COLUMN_COUNT)),
+ Arrays.asList(TSDataType.TEXT, TSDataType.TEXT));
for (String columnPath : nodes) {
RowRecord record = new RowRecord(0);
Field field = new Field(TSDataType.TEXT);
@@ -389,9 +397,9 @@ public class PlanExecutor implements IPlanExecutor {
private QueryDataSet processCountTimeSeries(CountPlan countPlan) throws MetadataException {
int num = getPaths(countPlan.getPath().toString()).size();
SingleDataSet singleDataSet =
- new SingleDataSet(
- Collections.singletonList(new Path(COLUMN_CHILD_PATHS)),
- Collections.singletonList(TSDataType.INT32));
+ new SingleDataSet(
+ Collections.singletonList(new Path(COLUMN_CHILD_PATHS)),
+ Collections.singletonList(TSDataType.INT32));
Field field = new Field(TSDataType.INT32);
field.setIntV(num);
RowRecord record = new RowRecord(0);
@@ -401,11 +409,11 @@ public class PlanExecutor implements IPlanExecutor {
}
private QueryDataSet processShowDevices(ShowDevicesPlan showDevicesPlan)
- throws MetadataException {
+ throws MetadataException {
ListDataSet listDataSet =
- new ListDataSet(
- Collections.singletonList(new Path(COLUMN_DEVICES)),
- Collections.singletonList(TSDataType.TEXT));
+ new ListDataSet(
+ Collections.singletonList(new Path(COLUMN_DEVICES)),
+ Collections.singletonList(TSDataType.TEXT));
Set<String> devices = getDevices(showDevicesPlan.getPath().toString());
for (String s : devices) {
RowRecord record = new RowRecord(0);
@@ -422,12 +430,12 @@ public class PlanExecutor implements IPlanExecutor {
}
private QueryDataSet processShowChildPaths(ShowChildPathsPlan showChildPathsPlan)
- throws MetadataException {
+ throws MetadataException {
Set<String> childPathsList = getPathNextChildren(showChildPathsPlan.getPath().toString());
ListDataSet listDataSet =
- new ListDataSet(
- Collections.singletonList(new Path(COLUMN_CHILD_PATHS)),
- Collections.singletonList(TSDataType.TEXT));
+ new ListDataSet(
+ Collections.singletonList(new Path(COLUMN_CHILD_PATHS)),
+ Collections.singletonList(TSDataType.TEXT));
for (String s : childPathsList) {
RowRecord record = new RowRecord(0);
Field field = new Field(TSDataType.TEXT);
@@ -442,12 +450,16 @@ public class PlanExecutor implements IPlanExecutor {
return MManager.getInstance().getChildNodePathInNextLevel(path);
}
+ protected List<String> getAllStorageGroupNames() {
+ return MManager.getInstance().getAllStorageGroupNames();
+ }
+
private QueryDataSet processShowStorageGroup() {
ListDataSet listDataSet =
- new ListDataSet(
- Collections.singletonList(new Path(COLUMN_STORAGE_GROUP)),
- Collections.singletonList(TSDataType.TEXT));
- List<String> storageGroupList = MManager.getInstance().getAllStorageGroupNames();
+ new ListDataSet(
+ Collections.singletonList(new Path(COLUMN_STORAGE_GROUP)),
+ Collections.singletonList(TSDataType.TEXT));
+ List<String> storageGroupList = getAllStorageGroupNames();
for (String s : storageGroupList) {
RowRecord record = new RowRecord(0);
Field field = new Field(TSDataType.TEXT);
@@ -459,14 +471,14 @@ public class PlanExecutor implements IPlanExecutor {
}
private QueryDataSet processShowTimeseriesWithIndex(ShowTimeSeriesPlan showTimeSeriesPlan)
- throws MetadataException {
- List<ShowTimeSeriesResult> timeseriesList = mManager.getAllTimeseriesSchema(showTimeSeriesPlan);
+ throws MetadataException {
+ List<ShowTimeSeriesResult> timeseriesList = showTimeseriesWithIndex(showTimeSeriesPlan);
return getQueryDataSet(timeseriesList);
}
private QueryDataSet processShowTimeseries(ShowTimeSeriesPlan showTimeSeriesPlan)
- throws MetadataException {
- List<ShowTimeSeriesResult> timeseriesList = mManager.showTimeseries(showTimeSeriesPlan);
+ throws MetadataException {
+ List<ShowTimeSeriesResult> timeseriesList = showTimeseries(showTimeSeriesPlan);
return getQueryDataSet(timeseriesList);
}
@@ -510,8 +522,16 @@ public class PlanExecutor implements IPlanExecutor {
return listDataSet;
}
+ protected List<ShowTimeSeriesResult> showTimeseries(ShowTimeSeriesPlan plan) throws MetadataException {
+ return MManager.getInstance().showTimeseries(plan);
+ }
+
+ protected List<ShowTimeSeriesResult> showTimeseriesWithIndex(ShowTimeSeriesPlan plan) throws MetadataException {
+ return MManager.getInstance().getAllTimeseriesSchema(plan);
+ }
+
private void updateRecord(
- RowRecord record, Map<String, String> tagAndAttribute, List<Path> paths) {
+ RowRecord record, Map<String, String> tagAndAttribute, List<Path> paths) {
for (int i = 6; i < paths.size(); i++) {
updateRecord(record, tagAndAttribute.get(paths.get(i).getFullPath()));
}
@@ -527,14 +547,18 @@ public class PlanExecutor implements IPlanExecutor {
record.addField(field);
}
+ protected List<StorageGroupMNode> getAllStorageGroupNodes() {
+ return MManager.getInstance().getAllStorageGroupNodes();
+ }
+
private QueryDataSet processShowTTLQuery(ShowTTLPlan showTTLPlan) {
ListDataSet listDataSet =
- new ListDataSet(
- Arrays.asList(new Path(COLUMN_STORAGE_GROUP), new Path(COLUMN_TTL)),
- Arrays.asList(TSDataType.TEXT, TSDataType.INT64));
+ new ListDataSet(
+ Arrays.asList(new Path(COLUMN_STORAGE_GROUP), new Path(COLUMN_TTL)),
+ Arrays.asList(TSDataType.TEXT, TSDataType.INT64));
List<String> selectedSgs = showTTLPlan.getStorageGroups();
- List<StorageGroupMNode> storageGroups = MManager.getInstance().getAllStorageGroupNodes();
+ List<StorageGroupMNode> storageGroups = getAllStorageGroupNodes();
int timestamp = 0;
for (StorageGroupMNode mNode : storageGroups) {
String sgName = mNode.getFullPath();
@@ -561,9 +585,9 @@ public class PlanExecutor implements IPlanExecutor {
private QueryDataSet processShowVersion() {
SingleDataSet singleDataSet =
- new SingleDataSet(
- Collections.singletonList(new Path(IoTDBConstant.COLUMN_VERSION)),
- Collections.singletonList(TSDataType.TEXT));
+ new SingleDataSet(
+ Collections.singletonList(new Path(IoTDBConstant.COLUMN_VERSION)),
+ Collections.singletonList(TSDataType.TEXT));
Field field = new Field(TSDataType.TEXT);
field.setBinaryV(new Binary(IoTDBConstant.VERSION));
RowRecord rowRecord = new RowRecord(0);
@@ -574,76 +598,76 @@ public class PlanExecutor implements IPlanExecutor {
private QueryDataSet processShowDynamicParameterQuery() {
ListDataSet listDataSet =
- new ListDataSet(
- Arrays.asList(new Path(COLUMN_PARAMETER), new Path(COLUMN_VALUE)),
- Arrays.asList(TSDataType.TEXT, TSDataType.TEXT));
+ new ListDataSet(
+ Arrays.asList(new Path(COLUMN_PARAMETER), new Path(COLUMN_VALUE)),
+ Arrays.asList(TSDataType.TEXT, TSDataType.TEXT));
int timestamp = 0;
addRowRecordForShowQuery(
- listDataSet,
- timestamp++,
- "memtable size threshold",
- IoTDBDescriptor.getInstance().getConfig().getMemtableSizeThreshold() + "B");
+ listDataSet,
+ timestamp++,
+ "memtable size threshold",
+ IoTDBDescriptor.getInstance().getConfig().getMemtableSizeThreshold() + "B");
addRowRecordForShowQuery(
- listDataSet,
- timestamp++,
- "memtable number",
- IoTDBDescriptor.getInstance().getConfig().getMaxMemtableNumber() + "B");
+ listDataSet,
+ timestamp++,
+ "memtable number",
+ IoTDBDescriptor.getInstance().getConfig().getMaxMemtableNumber() + "B");
addRowRecordForShowQuery(
- listDataSet,
- timestamp++,
- "tsfile size threshold",
- IoTDBDescriptor.getInstance().getConfig().getTsFileSizeThreshold() + "B");
+ listDataSet,
+ timestamp++,
+ "tsfile size threshold",
+ IoTDBDescriptor.getInstance().getConfig().getTsFileSizeThreshold() + "B");
addRowRecordForShowQuery(
- listDataSet,
- timestamp++,
- "compression ratio",
- Double.toString(CompressionRatio.getInstance().getRatio()));
+ listDataSet,
+ timestamp++,
+ "compression ratio",
+ Double.toString(CompressionRatio.getInstance().getRatio()));
addRowRecordForShowQuery(
- listDataSet,
- timestamp++,
- "storage group number",
- Integer.toString(MManager.getInstance().getAllStorageGroupNames().size()));
+ listDataSet,
+ timestamp++,
+ "storage group number",
+ Integer.toString(MManager.getInstance().getAllStorageGroupNames().size()));
addRowRecordForShowQuery(
- listDataSet,
- timestamp++,
- "timeseries number",
- Integer.toString(IoTDBConfigDynamicAdapter.getInstance().getTotalTimeseries()));
+ listDataSet,
+ timestamp++,
+ "timeseries number",
+ Integer.toString(IoTDBConfigDynamicAdapter.getInstance().getTotalTimeseries()));
addRowRecordForShowQuery(
- listDataSet,
- timestamp,
- "maximal timeseries number among storage groups",
- Long.toString(MManager.getInstance().getMaximalSeriesNumberAmongStorageGroups()));
+ listDataSet,
+ timestamp,
+ "maximal timeseries number among storage groups",
+ Long.toString(MManager.getInstance().getMaximalSeriesNumberAmongStorageGroups()));
return listDataSet;
}
private QueryDataSet processShowFlushTaskInfo() {
ListDataSet listDataSet =
- new ListDataSet(
- Arrays.asList(new Path(COLUMN_ITEM), new Path(COLUMN_VALUE)),
- Arrays.asList(TSDataType.TEXT, TSDataType.TEXT));
+ new ListDataSet(
+ Arrays.asList(new Path(COLUMN_ITEM), new Path(COLUMN_VALUE)),
+ Arrays.asList(TSDataType.TEXT, TSDataType.TEXT));
int timestamp = 0;
addRowRecordForShowQuery(
- listDataSet,
- timestamp++,
- "total number of flush tasks",
- Integer.toString(FlushTaskPoolManager.getInstance().getTotalTasks()));
+ listDataSet,
+ timestamp++,
+ "total number of flush tasks",
+ Integer.toString(FlushTaskPoolManager.getInstance().getTotalTasks()));
addRowRecordForShowQuery(
- listDataSet,
- timestamp++,
- "number of working flush tasks",
- Integer.toString(FlushTaskPoolManager.getInstance().getWorkingTasksNumber()));
+ listDataSet,
+ timestamp++,
+ "number of working flush tasks",
+ Integer.toString(FlushTaskPoolManager.getInstance().getWorkingTasksNumber()));
addRowRecordForShowQuery(
- listDataSet,
- timestamp,
- "number of waiting flush tasks",
- Integer.toString(FlushTaskPoolManager.getInstance().getWaitingTasksNumber()));
+ listDataSet,
+ timestamp,
+ "number of waiting flush tasks",
+ Integer.toString(FlushTaskPoolManager.getInstance().getWaitingTasksNumber()));
return listDataSet;
}
private void addRowRecordForShowQuery(
- ListDataSet listDataSet, int timestamp, String item, String value) {
+ ListDataSet listDataSet, int timestamp, String item, String value) {
RowRecord rowRecord = new RowRecord(timestamp);
Field itemField = new Field(TSDataType.TEXT);
itemField.setBinaryV(new Binary(item));
@@ -676,7 +700,7 @@ public class PlanExecutor implements IPlanExecutor {
File file = plan.getFile();
if (!file.exists()) {
throw new QueryProcessException(
- String.format("File path %s doesn't exists.", file.getPath()));
+ String.format("File path %s doesn't exists.", file.getPath()));
}
if (file.isDirectory()) {
recursionFileDir(file, plan);
@@ -708,8 +732,8 @@ public class PlanExecutor implements IPlanExecutor {
if (restorableTsFileIOWriter.hasCrashed()) {
restorableTsFileIOWriter.close();
throw new QueryProcessException(
- String.format(
- "Cannot load file %s because the file has crashed.", file.getAbsolutePath()));
+ String.format(
+ "Cannot load file %s because the file has crashed.", file.getAbsolutePath()));
}
Map<Path, MeasurementSchema> schemaMap = new HashMap<>();
@@ -723,9 +747,9 @@ public class PlanExecutor implements IPlanExecutor {
FileLoaderUtils.checkTsFileResource(tsFileResource);
if (UpgradeUtils.isNeedUpgrade(tsFileResource)) {
throw new QueryProcessException(
- String.format(
- "Cannot load file %s because the file's version is old which needs to be upgraded.",
- file.getAbsolutePath()));
+ String.format(
+ "Cannot load file %s because the file's version is old which needs to be upgraded.",
+ file.getAbsolutePath()));
}
// create schemas if they doesn't exist
@@ -736,15 +760,15 @@ public class PlanExecutor implements IPlanExecutor {
StorageEngine.getInstance().loadNewTsFile(tsFileResource);
} catch (Exception e) {
throw new QueryProcessException(
- String.format("Cannot load file %s because %s", file.getAbsolutePath(), e.getMessage()));
+ String.format("Cannot load file %s because %s", file.getAbsolutePath(), e.getMessage()));
}
}
private void createSchemaAutomatically(
- List<ChunkGroupMetadata> chunkGroupMetadataList,
- Map<Path, MeasurementSchema> knownSchemas,
- int sgLevel)
- throws QueryProcessException, MetadataException {
+ List<ChunkGroupMetadata> chunkGroupMetadataList,
+ Map<Path, MeasurementSchema> knownSchemas,
+ int sgLevel)
+ throws QueryProcessException, MetadataException {
if (chunkGroupMetadataList.isEmpty()) {
return;
}
@@ -762,20 +786,20 @@ public class PlanExecutor implements IPlanExecutor {
MeasurementSchema schema = knownSchemas.get(series);
if (schema == null) {
throw new MetadataException(
- String.format(
- "Can not get the schema of measurement [%s]",
- chunkMetadata.getMeasurementUid()));
+ String.format(
+ "Can not get the schema of measurement [%s]",
+ chunkMetadata.getMeasurementUid()));
}
if (!node.hasChild(chunkMetadata.getMeasurementUid())) {
mManager.createTimeseries(
- series.getFullPath(),
- schema.getType(),
- schema.getEncodingType(),
- schema.getCompressor(),
- Collections.emptyMap());
+ series.getFullPath(),
+ schema.getType(),
+ schema.getEncodingType(),
+ schema.getCompressor(),
+ Collections.emptyMap());
} else if (node.getChild(chunkMetadata.getMeasurementUid()) instanceof InternalMNode) {
throw new QueryProcessException(
- String.format("Current Path is not leaf node. %s", series));
+ String.format("Current Path is not leaf node. %s", series));
}
}
}
@@ -791,29 +815,29 @@ public class PlanExecutor implements IPlanExecutor {
try {
if (!StorageEngine.getInstance().deleteTsfile(plan.getFile())) {
throw new QueryProcessException(
- String.format("File %s doesn't exist.", plan.getFile().getName()));
+ String.format("File %s doesn't exist.", plan.getFile().getName()));
}
} catch (StorageEngineException e) {
throw new QueryProcessException(
- String.format("Cannot remove file because %s", e.getMessage()));
+ String.format("Cannot remove file because %s", e.getMessage()));
}
}
private void operateMoveFile(OperateFilePlan plan) throws QueryProcessException {
if (!plan.getTargetDir().exists() || !plan.getTargetDir().isDirectory()) {
throw new QueryProcessException(
- String.format("Target dir %s is invalid.", plan.getTargetDir().getPath()));
+ String.format("Target dir %s is invalid.", plan.getTargetDir().getPath()));
}
try {
if (!StorageEngine.getInstance().moveTsfile(plan.getFile(), plan.getTargetDir())) {
throw new QueryProcessException(
- String.format("File %s doesn't exist.", plan.getFile().getName()));
+ String.format("File %s doesn't exist.", plan.getFile().getName()));
}
} catch (StorageEngineException | IOException e) {
throw new QueryProcessException(
- String.format(
- "Cannot move file %s to target directory %s because %s",
- plan.getFile().getPath(), plan.getTargetDir().getPath(), e.getMessage()));
+ String.format(
+ "Cannot move file %s to target directory %s because %s",
+ plan.getFile().getPath(), plan.getTargetDir().getPath(), e.getMessage()));
}
}
@@ -840,7 +864,7 @@ public class PlanExecutor implements IPlanExecutor {
try {
if (!mManager.isPathExist(path.getFullPath())) {
throw new QueryProcessException(
- String.format("Time series %s does not exist.", path.getFullPath()));
+ String.format("Time series %s does not exist.", path.getFullPath()));
}
mManager.getStorageGroupName(path.getFullPath());
StorageEngine.getInstance().delete(deviceId, measurementId, timestamp);
@@ -851,59 +875,88 @@ public class PlanExecutor implements IPlanExecutor {
@Override
public void insert(InsertPlan insertPlan) throws QueryProcessException {
- MNode node = null;
try {
String[] measurementList = insertPlan.getMeasurements();
String deviceId = insertPlan.getDeviceId();
- node = mManager.getDeviceNodeWithAutoCreateAndReadLock(deviceId);
String[] strValues = insertPlan.getValues();
- MeasurementSchema[] schemas = new MeasurementSchema[measurementList.length];
+ MeasurementSchema[] schemas = getSeriesSchemas(measurementList, deviceId, strValues);
+ insertPlan.setSchemas(schemas);
+ StorageEngine.getInstance().insert(insertPlan);
+ } catch (StorageEngineException | MetadataException e) {
+ throw new QueryProcessException(e);
+ }
+ }
+
+ protected MeasurementSchema[] getSeriesSchemas(String[] measurementList, String deviceId,
+ String[] strValues) throws MetadataException {
+ MeasurementSchema[] schemas = new MeasurementSchema[measurementList.length];
+ MNode node = null;
+ try {
+ node = mManager.getDeviceNodeWithAutoCreateAndReadLock(deviceId);
+ } catch (PathNotExistException e) {
+ // ignore
+ }
+ try {
for (int i = 0; i < measurementList.length; i++) {
String measurement = measurementList[i];
- if (!node.hasChild(measurement)) {
+
+ if (node != null && !node.hasChild(measurement)) {
if (!IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()) {
- throw new PathNotExistException(deviceId + PATH_SEPARATOR + measurement);
+ MeasurementSchema schema = MManager.getInstance()
+ .getSeriesSchema(deviceId, measurement);
+ if (schema != null) {
+ schemas[i] = schema;
+ continue;
+ } else {
+ throw new PathNotExistException(deviceId + PATH_SEPARATOR + measurement);
+ }
}
TSDataType dataType = TypeInferenceUtils.getPredictedDataType(strValues[i]);
Path path = new Path(deviceId, measurement);
+
internalCreateTimeseries(path.toString(), dataType);
+ LeafMNode measurementNode = (LeafMNode) node.getChild(measurement);
+ schemas[i] = measurementNode.getSchema();
+ } else if (node != null) {
+ LeafMNode measurementNode = (LeafMNode) node.getChild(measurement);
+ schemas[i] = measurementNode.getSchema();
+ } else {
+ schemas[i] = MManager.getInstance().getSeriesSchema(deviceId, measurement);
}
- LeafMNode measurementNode = (LeafMNode) node.getChild(measurement);
- schemas[i] = measurementNode.getSchema();
}
-
- insertPlan.setSchemas(schemas);
- StorageEngine.getInstance().insert(insertPlan);
- } catch (StorageEngineException | MetadataException e) {
- throw new QueryProcessException(e);
} finally {
if (node != null) {
((InternalMNode) node).readUnlock();
}
}
+ return schemas;
}
- /** create timeseries with ignore PathAlreadyExistException */
+ /**
+ * create timeseries with ignore PathAlreadyExistException
+ */
private void internalCreateTimeseries(String path, TSDataType dataType) throws MetadataException {
try {
mManager.createTimeseries(
- path,
- dataType,
- getDefaultEncoding(dataType),
- TSFileDescriptor.getInstance().getConfig().getCompressor(),
- Collections.emptyMap());
+ path,
+ dataType,
+ getDefaultEncoding(dataType),
+ TSFileDescriptor.getInstance().getConfig().getCompressor(),
+ Collections.emptyMap());
} catch (PathAlreadyExistException e) {
if (logger.isDebugEnabled()) {
logger.debug(
- "Ignore PathAlreadyExistException when Concurrent inserting"
- + " a non-exist time series {}",
- path);
+ "Ignore PathAlreadyExistException when Concurrent inserting"
+ + " a non-exist time series {}",
+ path);
}
}
}
- /** Get default encoding by dataType */
+ /**
+ * Get default encoding by dataType
+ */
private TSEncoding getDefaultEncoding(TSDataType dataType) {
IoTDBConfig conf = IoTDBDescriptor.getInstance().getConfig();
switch (dataType) {
@@ -921,7 +974,7 @@ public class PlanExecutor implements IPlanExecutor {
return conf.getDefaultTextEncoding();
default:
throw new UnSupportedDataTypeException(
- String.format("Data type %s is not supported.", dataType.toString()));
+ String.format("Data type %s is not supported.", dataType.toString()));
}
}
@@ -941,9 +994,9 @@ public class PlanExecutor implements IPlanExecutor {
if (!node.hasChild(measurementList[i])) {
if (!conf.isAutoCreateSchemaEnabled()) {
throw new QueryProcessException(
- String.format(
- "Current deviceId[%s] does not contain measurement:%s",
- deviceId, measurementList[i]));
+ String.format(
+ "Current deviceId[%s] does not contain measurement:%s",
+ deviceId, measurementList[i]));
}
Path path = new Path(deviceId, measurementList[i]);
TSDataType dataType = dataTypes[i];
@@ -954,11 +1007,11 @@ public class PlanExecutor implements IPlanExecutor {
// check data type
if (measurementNode.getSchema().getType() != insertTabletPlan.getDataTypes()[i]) {
throw new QueryProcessException(
- String.format(
- "Datatype mismatch, Insert measurement %s type %s, metadata tree type %s",
- measurementList[i],
- insertTabletPlan.getDataTypes()[i],
- measurementNode.getSchema().getType()));
+ String.format(
+ "Datatype mismatch, Insert measurement %s type %s, metadata tree type %s",
+ measurementList[i],
+ insertTabletPlan.getDataTypes()[i],
+ measurementNode.getSchema().getType()));
}
schemas[i] = measurementNode.getSchema();
}
@@ -1034,7 +1087,7 @@ public class PlanExecutor implements IPlanExecutor {
}
private boolean operateWatermarkEmbedding(List<String> users, boolean useWatermark)
- throws QueryProcessException {
+ throws QueryProcessException {
try {
for (String user : users) {
authorizer.setUserUseWaterMark(user, useWatermark);
@@ -1046,7 +1099,7 @@ public class PlanExecutor implements IPlanExecutor {
}
private boolean createTimeSeries(CreateTimeSeriesPlan createTimeSeriesPlan)
- throws QueryProcessException {
+ throws QueryProcessException {
try {
mManager.createTimeseries(createTimeSeriesPlan);
} catch (MetadataException e) {
@@ -1055,15 +1108,15 @@ public class PlanExecutor implements IPlanExecutor {
return true;
}
- private boolean deleteTimeSeries(DeleteTimeSeriesPlan deleteTimeSeriesPlan)
- throws QueryProcessException {
+ protected boolean deleteTimeSeries(DeleteTimeSeriesPlan deleteTimeSeriesPlan)
+ throws QueryProcessException {
List<Path> deletePathList = deleteTimeSeriesPlan.getPaths();
try {
deleteDataOfTimeSeries(deletePathList);
Set<String> emptyStorageGroups = new HashSet<>();
List<String> failedNames = new LinkedList<>();
for (Path path : deletePathList) {
- Pair<Set<String>, String> pair = mManager.deleteTimeseries(path.toString());
+ Pair<Set<String>, String> pair = deleteTimeseries(path.toString());
emptyStorageGroups.addAll(pair.left);
if (!pair.right.isEmpty()) {
failedNames.add(pair.right);
@@ -1082,7 +1135,7 @@ public class PlanExecutor implements IPlanExecutor {
}
private boolean alterTimeSeries(AlterTimeSeriesPlan alterTimeSeriesPlan)
- throws QueryProcessException {
+ throws QueryProcessException {
Path path = alterTimeSeriesPlan.getPath();
Map<String, String> alterMap = alterTimeSeriesPlan.getAlterMap();
try {
@@ -1106,24 +1159,24 @@ public class PlanExecutor implements IPlanExecutor {
break;
case UPSERT:
mManager.upsertTagsAndAttributes(
- alterTimeSeriesPlan.getTagsMap(),
- alterTimeSeriesPlan.getAttributesMap(),
- path.getFullPath());
+ alterTimeSeriesPlan.getTagsMap(),
+ alterTimeSeriesPlan.getAttributesMap(),
+ path.getFullPath());
break;
}
} catch (MetadataException e) {
throw new QueryProcessException(e);
} catch (IOException e) {
throw new QueryProcessException(
- String.format(
- "Something went wrong while read/write the [%s]'s tag/attribute info.",
- path.getFullPath()));
+ String.format(
+ "Something went wrong while read/write the [%s]'s tag/attribute info.",
+ path.getFullPath()));
}
return true;
}
public boolean setStorageGroup(SetStorageGroupPlan setStorageGroupPlan)
- throws QueryProcessException {
+ throws QueryProcessException {
Path path = setStorageGroupPlan.getPath();
try {
mManager.setStorageGroup(path.getFullPath());
@@ -1133,8 +1186,8 @@ public class PlanExecutor implements IPlanExecutor {
return true;
}
- private boolean deleteStorageGroups(DeleteStorageGroupPlan deleteStorageGroupPlan)
- throws QueryProcessException {
+ protected boolean deleteStorageGroups(DeleteStorageGroupPlan deleteStorageGroupPlan)
+ throws QueryProcessException {
List<String> deletePathList = new ArrayList<>();
try {
for (Path storageGroupPath : deleteStorageGroupPlan.getPaths()) {
@@ -1153,7 +1206,7 @@ public class PlanExecutor implements IPlanExecutor {
*
* @param pathList deleted paths
*/
- private void deleteDataOfTimeSeries(List<Path> pathList)
+ protected void deleteDataOfTimeSeries(List<Path> pathList)
throws QueryProcessException, StorageGroupNotSetException, StorageEngineException {
for (Path p : pathList) {
DeletePlan deletePlan = new DeletePlan();
@@ -1163,7 +1216,8 @@ public class PlanExecutor implements IPlanExecutor {
}
}
- private QueryDataSet processAuthorQuery(AuthorPlan plan) throws QueryProcessException {
+ protected QueryDataSet processAuthorQuery(AuthorPlan plan)
+ throws QueryProcessException {
AuthorType authorType = plan.getAuthorType();
String userName = plan.getUserName();
String roleName = plan.getRoleName();
@@ -1352,4 +1406,12 @@ public class PlanExecutor implements IPlanExecutor {
}
return dataSet;
}
+
+ protected Pair<Set<String>, String> deleteTimeseries(String path) throws MetadataException {
+ return mManager.deleteTimeseries(path);
+ }
+
+ protected void loadConfiguration(LoadConfigurationPlan plan) throws QueryProcessException {
+ IoTDBDescriptor.getInstance().loadHotModifiedProps();
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/LoadConfigurationOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/LoadConfigurationOperator.java
index 18ad4ae..23b827e 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/LoadConfigurationOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/LoadConfigurationOperator.java
@@ -24,8 +24,19 @@ import org.apache.iotdb.db.qp.logical.RootOperator;
public class LoadConfigurationOperator extends RootOperator {
- public LoadConfigurationOperator() {
+ LoadConfigurationOperatorType loadConfigurationOperatorType;
+
+ public LoadConfigurationOperator(LoadConfigurationOperatorType type) {
super(SQLConstant.TOK_LOAD_CONFIGURATION);
this.operatorType = OperatorType.LOAD_CONFIGURATION;
+ this.loadConfigurationOperatorType = type;
+ }
+
+ public LoadConfigurationOperatorType getLoadConfigurationOperatorType() {
+ return loadConfigurationOperatorType;
+ }
+
+ public enum LoadConfigurationOperatorType{
+ LOCAL, GLOBAL
}
}
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 0ad6470..5d30f66 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
@@ -24,11 +24,19 @@ 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.logical.Operator.OperatorType;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
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.sys.AuthorPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.DataAuthPlan;
+import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan;
+import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.LoadConfigurationPlan;
import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
+import org.apache.iotdb.db.qp.physical.sys.SetTTLPlan;
+import org.apache.iotdb.db.qp.physical.sys.ShowTimeSeriesPlan;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
@@ -44,14 +52,13 @@ public abstract class PhysicalPlan {
private static final int NULL_VALUE_LEN = -1;
//for cluster mode, whether the plan may be splitted into several sub plans
- protected boolean canbeSplit = true;
+ protected boolean canBeSplit = true;
/**
- * whether the plan can be split into more than one Plans.
- * Only used in the cluster mode.
+ * whether the plan can be split into more than one Plans. Only used in the cluster mode.
*/
- public boolean canbeSplit() {
- return canbeSplit;
+ public boolean canBeSplit() {
+ return canBeSplit;
}
protected PhysicalPlan(boolean isQuery) {
@@ -89,15 +96,33 @@ public abstract class PhysicalPlan {
isQuery = query;
}
- public void serializeTo(DataOutputStream stream) throws IOException {
+ /**
+ * Serialize the plan into the given buffer. All necessary fields will be serialized.
+ *
+ * @param stream
+ * @throws IOException
+ */
+ public void serialize(DataOutputStream stream) throws IOException {
throw new UnsupportedOperationException(SERIALIZATION_UNIMPLEMENTED);
}
- public void serializeTo(ByteBuffer buffer) {
+ /**
+ * Serialize the plan into the given buffer. This is provided for WAL, so fields that can be
+ * recovered will not be serialized.
+ *
+ * @param buffer
+ */
+ public void serialize(ByteBuffer buffer) {
throw new UnsupportedOperationException(SERIALIZATION_UNIMPLEMENTED);
}
- public void deserializeFrom(ByteBuffer buffer) {
+ /**
+ * Deserialize the plan from the given buffer. This is provided for WAL, and must be used with
+ * serializeToWAL.
+ *
+ * @param buffer
+ */
+ public void deserialize(ByteBuffer buffer) {
throw new UnsupportedOperationException(SERIALIZATION_UNIMPLEMENTED);
}
@@ -142,23 +167,95 @@ public abstract class PhysicalPlan {
switch (type) {
case INSERT:
plan = new InsertPlan();
- plan.deserializeFrom(buffer);
+ plan.deserialize(buffer);
break;
case DELETE:
plan = new DeletePlan();
- plan.deserializeFrom(buffer);
+ plan.deserialize(buffer);
break;
case BATCHINSERT:
plan = new InsertTabletPlan();
- plan.deserializeFrom(buffer);
+ plan.deserialize(buffer);
break;
case SET_STORAGE_GROUP:
plan = new SetStorageGroupPlan();
- plan.deserializeFrom(buffer);
+ plan.deserialize(buffer);
break;
case CREATE_TIMESERIES:
plan = new CreateTimeSeriesPlan();
- plan.deserializeFrom(buffer);
+ plan.deserialize(buffer);
+ break;
+ case DELETE_TIMESERIES:
+ plan = new DeleteTimeSeriesPlan();
+ plan.deserialize(buffer);
+ break;
+ case TTL:
+ plan = new SetTTLPlan();
+ plan.deserialize(buffer);
+ break;
+ case GRANT_WATERMARK_EMBEDDING:
+ plan = new DataAuthPlan(OperatorType.GRANT_WATERMARK_EMBEDDING);
+ plan.deserialize(buffer);
+ break;
+ case REVOKE_WATERMARK_EMBEDDING:
+ plan = new DataAuthPlan(OperatorType.REVOKE_WATERMARK_EMBEDDING);
+ plan.deserialize(buffer);
+ break;
+ case CREATE_ROLE:
+ plan = new AuthorPlan(OperatorType.CREATE_ROLE);
+ plan.deserialize(buffer);
+ break;
+ case DELETE_ROLE:
+ plan = new AuthorPlan(OperatorType.DELETE_ROLE);
+ plan.deserialize(buffer);
+ break;
+ case CREATE_USER:
+ plan = new AuthorPlan(OperatorType.CREATE_USER);
+ plan.deserialize(buffer);
+ break;
+ case REVOKE_USER_ROLE:
+ plan = new AuthorPlan(OperatorType.REVOKE_USER_ROLE);
+ plan.deserialize(buffer);
+ break;
+ case REVOKE_ROLE_PRIVILEGE:
+ plan = new AuthorPlan(OperatorType.REVOKE_ROLE_PRIVILEGE);
+ plan.deserialize(buffer);
+ break;
+ case REVOKE_USER_PRIVILEGE:
+ plan = new AuthorPlan(OperatorType.REVOKE_USER_PRIVILEGE);
+ plan.deserialize(buffer);
+ break;
+ case GRANT_ROLE_PRIVILEGE:
+ plan = new AuthorPlan(OperatorType.GRANT_ROLE_PRIVILEGE);
+ plan.deserialize(buffer);
+ break;
+ case GRANT_USER_PRIVILEGE:
+ plan = new AuthorPlan(OperatorType.GRANT_USER_PRIVILEGE);
+ plan.deserialize(buffer);
+ break;
+ case GRANT_USER_ROLE:
+ plan = new AuthorPlan(OperatorType.GRANT_USER_ROLE);
+ plan.deserialize(buffer);
+ break;
+ case MODIFY_PASSWORD:
+ plan = new AuthorPlan(OperatorType.MODIFY_PASSWORD);
+ plan.deserialize(buffer);
+ break;
+ case DELETE_USER:
+ plan = new AuthorPlan(OperatorType.DELETE_USER);
+ plan.deserialize(buffer);
+ break;
+ case DELETE_STORAGE_GROUP:
+ plan = new DeleteStorageGroupPlan();
+ plan.deserialize(buffer);
+ break;
+ case SHOW_TIMESERIES:
+ plan = new ShowTimeSeriesPlan();
+ plan.deserialize(buffer);
+ break;
+ case LOAD_CONFIGURATION:
+ plan = new LoadConfigurationPlan();
+ plan.deserialize(buffer);
break;
default:
throw new IOException("unrecognized log type " + type);
@@ -168,7 +265,9 @@ public abstract class PhysicalPlan {
}
public enum PhysicalPlanType {
- INSERT, DELETE, BATCHINSERT, SET_STORAGE_GROUP, CREATE_TIMESERIES
+ INSERT, DELETE, BATCHINSERT, SET_STORAGE_GROUP, CREATE_TIMESERIES, TTL, GRANT_WATERMARK_EMBEDDING, REVOKE_WATERMARK_EMBEDDING,
+ CREATE_ROLE, DELETE_ROLE, CREATE_USER, REVOKE_USER_ROLE, REVOKE_ROLE_PRIVILEGE, REVOKE_USER_PRIVILEGE, GRANT_ROLE_PRIVILEGE, GRANT_USER_PRIVILEGE, GRANT_USER_ROLE, MODIFY_PASSWORD, DELETE_USER,
+ DELETE_STORAGE_GROUP, SHOW_TIMESERIES, DELETE_TIMESERIES, LOAD_CONFIGURATION
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java
index d2f1236..381ad5c 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java
@@ -104,25 +104,34 @@ public class DeletePlan extends PhysicalPlan {
}
@Override
- public void serializeTo(DataOutputStream stream) throws IOException {
+ public void serialize(DataOutputStream stream) throws IOException {
int type = PhysicalPlanType.DELETE.ordinal();
stream.writeByte((byte) type);
stream.writeLong(deleteTime);
- putString(stream, paths.get(0).getFullPath());
+ stream.writeInt(paths.size());
+ for (Path path : paths) {
+ putString(stream, path.getFullPath());
+ }
}
@Override
- public void serializeTo(ByteBuffer buffer) {
+ public void serialize(ByteBuffer buffer) {
int type = PhysicalPlanType.DELETE.ordinal();
buffer.put((byte) type);
buffer.putLong(deleteTime);
- putString(buffer, paths.get(0).getFullPath());
+ buffer.putInt(paths.size());
+ for (Path path : paths) {
+ putString(buffer, path.getFullPath());
+ }
}
@Override
- public void deserializeFrom(ByteBuffer buffer) {
+ public void deserialize(ByteBuffer buffer) {
this.deleteTime = buffer.getLong();
+ int pathSize = buffer.getInt();
this.paths = new ArrayList();
- this.paths.add(new Path(readString(buffer)));
+ for (int i = 0; i < pathSize; i++) {
+ paths.add(new Path(readString(buffer)));
+ }
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
index 5f334af..d922f5c 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
@@ -49,7 +49,7 @@ public class InsertPlan extends PhysicalPlan {
public InsertPlan() {
super(false, OperatorType.INSERT);
- canbeSplit = false;
+ canBeSplit = false;
}
@TestOnly
@@ -59,7 +59,7 @@ public class InsertPlan extends PhysicalPlan {
this.deviceId = deviceId;
this.measurements = new String[] {measurement};
this.values = new String[] {insertValue};
- canbeSplit = false;
+ canBeSplit = false;
}
public InsertPlan(TSRecord tsRecord) {
@@ -74,7 +74,7 @@ public class InsertPlan extends PhysicalPlan {
schemas[i] = new MeasurementSchema(measurements[i], tsRecord.dataPointList.get(i).getType(), TSEncoding.PLAIN);
values[i] = tsRecord.dataPointList.get(i).getValue().toString();
}
- canbeSplit = false;
+ canBeSplit = false;
}
public InsertPlan(String deviceId, long insertTime, String[] measurementList,
@@ -84,7 +84,7 @@ public class InsertPlan extends PhysicalPlan {
this.deviceId = deviceId;
this.measurements = measurementList;
this.values = insertValues;
- canbeSplit = false;
+ canBeSplit = false;
}
public long getTime() {
@@ -157,7 +157,7 @@ public class InsertPlan extends PhysicalPlan {
}
@Override
- public void serializeTo(DataOutputStream stream) throws IOException {
+ public void serialize(DataOutputStream stream) throws IOException {
int type = PhysicalPlanType.INSERT.ordinal();
stream.writeByte((byte) type);
stream.writeLong(time);
@@ -170,17 +170,13 @@ public class InsertPlan extends PhysicalPlan {
putString(stream, m);
}
- for (MeasurementSchema schema : schemas) {
- schema.serializeTo(stream);
- }
-
for (String m : values) {
putString(stream, m);
}
}
@Override
- public void serializeTo(ByteBuffer buffer) {
+ public void serialize(ByteBuffer buffer) {
int type = PhysicalPlanType.INSERT.ordinal();
buffer.put((byte) type);
buffer.putLong(time);
@@ -199,7 +195,7 @@ public class InsertPlan extends PhysicalPlan {
}
@Override
- public void deserializeFrom(ByteBuffer buffer) {
+ public void deserialize(ByteBuffer buffer) {
this.time = buffer.getLong();
this.deviceId = readString(buffer);
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertTabletPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertTabletPlan.java
index 4fcd813..8b6a64c 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertTabletPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertTabletPlan.java
@@ -127,7 +127,7 @@ public class InsertTabletPlan extends PhysicalPlan {
}
@Override
- public void serializeTo(DataOutputStream stream) throws IOException {
+ public void serialize(DataOutputStream stream) throws IOException {
int type = PhysicalPlanType.BATCHINSERT.ordinal();
stream.writeByte((byte) type);
@@ -215,7 +215,7 @@ public class InsertTabletPlan extends PhysicalPlan {
}
@Override
- public void serializeTo(ByteBuffer buffer) {
+ public void serialize(ByteBuffer buffer) {
int type = PhysicalPlanType.BATCHINSERT.ordinal();
buffer.put((byte) type);
@@ -312,7 +312,7 @@ public class InsertTabletPlan extends PhysicalPlan {
}
@Override
- public void deserializeFrom(ByteBuffer buffer) {
+ public void deserialize(ByteBuffer buffer) {
this.deviceId = readString(buffer);
int measurementSize = buffer.getInt();
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/AuthorPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/AuthorPlan.java
index 650d3c0..4a64681 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/AuthorPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/AuthorPlan.java
@@ -18,6 +18,9 @@
*/
package org.apache.iotdb.db.qp.physical.sys;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
@@ -26,13 +29,15 @@ import java.util.Set;
import org.apache.iotdb.db.auth.AuthException;
import org.apache.iotdb.db.auth.entity.PrivilegeType;
import org.apache.iotdb.db.qp.logical.Operator;
+import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
import org.apache.iotdb.db.qp.logical.sys.AuthorOperator;
+import org.apache.iotdb.db.qp.logical.sys.AuthorOperator.AuthorType;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.tsfile.read.common.Path;
public class AuthorPlan extends PhysicalPlan {
- private final AuthorOperator.AuthorType authorType;
+ private AuthorOperator.AuthorType authorType;
private String userName;
private String roleName;
private String password;
@@ -43,13 +48,13 @@ public class AuthorPlan extends PhysicalPlan {
/**
* AuthorPlan Constructor.
*
- * @param authorType author type
- * @param userName user name
- * @param roleName role name
- * @param password password
- * @param newPassword new password
+ * @param authorType author type
+ * @param userName user name
+ * @param roleName role name
+ * @param password password
+ * @param newPassword new password
* @param authorizationList authorization list in String[] structure
- * @param nodeName node name in Path structure
+ * @param nodeName node name in Path structure
* @throws AuthException Authentication Exception
*/
public AuthorPlan(AuthorOperator.AuthorType authorType, String userName, String roleName,
@@ -125,6 +130,58 @@ public class AuthorPlan extends PhysicalPlan {
}
}
+ public AuthorPlan(OperatorType operatorType) throws IOException {
+ super(false, operatorType);
+ setAuthorType(AuthorType.values()[transformOperatorTypeToAuthorType(operatorType)]);
+ }
+
+ private int transformOperatorTypeToAuthorType(OperatorType operatorType)
+ throws IOException {
+ int type;
+ switch (operatorType) {
+ case CREATE_ROLE:
+ type = AuthorType.CREATE_ROLE.ordinal();
+ break;
+ case DELETE_ROLE:
+ type = AuthorType.DROP_ROLE.ordinal();
+ break;
+ case CREATE_USER:
+ type = AuthorType.CREATE_USER.ordinal();
+ break;
+ case REVOKE_USER_ROLE:
+ type = AuthorType.REVOKE_ROLE_FROM_USER.ordinal();
+ break;
+ case REVOKE_ROLE_PRIVILEGE:
+ type = AuthorType.REVOKE_ROLE.ordinal();
+ break;
+ case REVOKE_USER_PRIVILEGE:
+ type = AuthorType.REVOKE_USER.ordinal();
+ break;
+ case GRANT_ROLE_PRIVILEGE:
+ type = AuthorType.GRANT_ROLE_TO_USER.ordinal();
+ break;
+ case GRANT_USER_PRIVILEGE:
+ type = AuthorType.GRANT_USER.ordinal();
+ break;
+ case GRANT_USER_ROLE:
+ type = AuthorType.GRANT_ROLE.ordinal();
+ break;
+ case MODIFY_PASSWORD:
+ type = AuthorType.UPDATE_USER.ordinal();
+ break;
+ case DELETE_USER:
+ type = AuthorType.DROP_USER.ordinal();
+ break;
+ default:
+ throw new IOException("unrecognized author type " + operatorType.name());
+ }
+ return type;
+ }
+
+ public void setAuthorType(AuthorOperator.AuthorType type) {
+ this.authorType = type;
+ }
+
public AuthorOperator.AuthorType getAuthorType() {
return authorType;
}
@@ -221,4 +278,120 @@ public class AuthorPlan extends PhysicalPlan {
.hash(getAuthorType(), getUserName(), getRoleName(), getPassword(), getNewPassword(),
getPermissions(), getNodeName());
}
+
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ int type = this.getPlanType(super.getOperatorType());
+ stream.writeByte((byte) type);
+ stream.writeInt(authorType.ordinal());
+ putString(stream, userName);
+ putString(stream, roleName);
+ putString(stream, password);
+ putString(stream, newPassword);
+ if (permissions == null) {
+ stream.writeBoolean(false);
+ } else {
+ stream.writeBoolean(true);
+ stream.writeInt(permissions.size());
+ for (int permission : permissions) {
+ stream.writeInt(permission);
+ }
+ }
+ if (nodeName == null) {
+ putString(stream, null);
+ } else {
+ putString(stream, nodeName.getFullPath());
+ }
+ }
+
+
+ @Override
+ public void serialize(ByteBuffer buffer) {
+ int type = this.getPlanType(super.getOperatorType());
+ buffer.put((byte) type);
+ buffer.putInt(authorType.ordinal());
+ putString(buffer, userName);
+ putString(buffer, roleName);
+ putString(buffer, password);
+ putString(buffer, newPassword);
+ if (permissions == null) {
+ buffer.put((byte) 0);
+ } else {
+ buffer.putInt((byte) 1);
+ buffer.putInt(permissions.size());
+ for (int permission : permissions) {
+ buffer.putInt(permission);
+ }
+ }
+ if (nodeName == null) {
+ putString(buffer, null);
+ } else {
+ putString(buffer, nodeName.getFullPath());
+ }
+ }
+
+ @Override
+ public void deserialize(ByteBuffer buffer) {
+ this.authorType = AuthorType.values()[buffer.getInt()];
+ this.userName = readString(buffer);
+ this.roleName = readString(buffer);
+ this.password = readString(buffer);
+ this.newPassword = readString(buffer);
+ byte hasPermissions = buffer.get();
+ if (hasPermissions == (byte) 0) {
+ this.permissions = null;
+ } else {
+ int permissionsSize = buffer.getInt();
+ this.permissions = new HashSet<>();
+ for (int i = 0; i < permissionsSize; i++) {
+ permissions.add(buffer.getInt());
+ }
+ }
+ String nodeNameStr = readString(buffer);
+ if (nodeNameStr == null) {
+ this.nodeName = null;
+ } else {
+ this.nodeName = new Path(nodeNameStr);
+ }
+ }
+
+ private int getPlanType(OperatorType operatorType) {
+ int type = OperatorType.LAST.ordinal();
+ switch (operatorType) {
+ case CREATE_ROLE:
+ type = PhysicalPlanType.CREATE_ROLE.ordinal();
+ break;
+ case DELETE_ROLE:
+ type = PhysicalPlanType.DELETE_ROLE.ordinal();
+ break;
+ case CREATE_USER:
+ type = PhysicalPlanType.CREATE_USER.ordinal();
+ break;
+ case REVOKE_USER_ROLE:
+ type = PhysicalPlanType.REVOKE_USER_ROLE.ordinal();
+ break;
+ case REVOKE_ROLE_PRIVILEGE:
+ type = PhysicalPlanType.REVOKE_ROLE_PRIVILEGE.ordinal();
+ break;
+ case REVOKE_USER_PRIVILEGE:
+ type = PhysicalPlanType.REVOKE_USER_PRIVILEGE.ordinal();
+ break;
+ case GRANT_ROLE_PRIVILEGE:
+ type = PhysicalPlanType.GRANT_ROLE_PRIVILEGE.ordinal();
+ break;
+ case GRANT_USER_PRIVILEGE:
+ type = PhysicalPlanType.GRANT_USER_PRIVILEGE.ordinal();
+ break;
+ case GRANT_USER_ROLE:
+ type = PhysicalPlanType.GRANT_USER_ROLE.ordinal();
+ break;
+ case MODIFY_PASSWORD:
+ type = PhysicalPlanType.MODIFY_PASSWORD.ordinal();
+ break;
+ case DELETE_USER:
+ type = PhysicalPlanType.DELETE_USER.ordinal();
+ break;
+ }
+ return type;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateTimeSeriesPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateTimeSeriesPlan.java
index cd2e7c3..5d31b22 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateTimeSeriesPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateTimeSeriesPlan.java
@@ -46,7 +46,7 @@ public class CreateTimeSeriesPlan extends PhysicalPlan {
public CreateTimeSeriesPlan() {
super(false, Operator.OperatorType.CREATE_TIMESERIES);
- canbeSplit = false;
+ canBeSplit = false;
}
public CreateTimeSeriesPlan(Path path, TSDataType dataType, TSEncoding encoding,
@@ -61,7 +61,7 @@ public class CreateTimeSeriesPlan extends PhysicalPlan {
this.tags = tags;
this.attributes = attributes;
this.alias = alias;
- canbeSplit = false;
+ canBeSplit = false;
}
public Path getPath() {
@@ -140,7 +140,7 @@ public class CreateTimeSeriesPlan extends PhysicalPlan {
}
@Override
- public void serializeTo(DataOutputStream stream) throws IOException {
+ public void serialize(DataOutputStream stream) throws IOException {
stream.writeByte((byte) PhysicalPlanType.CREATE_TIMESERIES.ordinal());
byte[] bytes = path.getFullPath().getBytes();
stream.writeInt(bytes.length);
@@ -183,7 +183,7 @@ public class CreateTimeSeriesPlan extends PhysicalPlan {
}
@Override
- public void deserializeFrom(ByteBuffer buffer) {
+ public void deserialize(ByteBuffer buffer) {
int length = buffer.getInt();
byte[] bytes = new byte[length];
buffer.get(bytes);
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DataAuthPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DataAuthPlan.java
index 380c765..21bbd1b 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DataAuthPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DataAuthPlan.java
@@ -18,6 +18,10 @@
*/
package org.apache.iotdb.db.qp.physical.sys;
+import java.io.DataOutputStream;
+import java.io.IOException;
+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;
@@ -25,7 +29,11 @@ import org.apache.iotdb.tsfile.read.common.Path;
public class DataAuthPlan extends PhysicalPlan {
- private final List<String> users;
+ private List<String> users;
+
+ public DataAuthPlan(OperatorType operatorType) {
+ super(false, operatorType);
+ }
public DataAuthPlan(OperatorType operatorType, List<String> users) {
super(false, operatorType);
@@ -40,4 +48,48 @@ public class DataAuthPlan extends PhysicalPlan {
public List<Path> getPaths() {
return null;
}
+
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ int type = this.getPlanType(super.getOperatorType());
+ stream.writeByte((byte) type);
+ stream.writeInt(users.size());
+
+ for (String user : users) {
+ putString(stream, user);
+ }
+ }
+
+ @Override
+ public void serialize(ByteBuffer buffer) {
+ int type = this.getPlanType(super.getOperatorType());
+ buffer.put((byte) type);
+ buffer.putInt(users.size());
+
+ for (String user : users) {
+ putString(buffer, user);
+ }
+ }
+
+ @Override
+ public void deserialize(ByteBuffer buffer) {
+ int userSize = buffer.getInt();
+ this.users = new ArrayList<>(userSize);
+ for (int i = 0; i < userSize; i++) {
+ users.add(readString(buffer));
+ }
+ }
+
+ private int getPlanType(OperatorType operatorType) {
+ int type = OperatorType.LAST.ordinal();
+ switch (operatorType) {
+ case GRANT_WATERMARK_EMBEDDING:
+ type = PhysicalPlanType.GRANT_WATERMARK_EMBEDDING.ordinal();
+ break;
+ case REVOKE_WATERMARK_EMBEDDING:
+ type = PhysicalPlanType.REVOKE_WATERMARK_EMBEDDING.ordinal();
+ break;
+ }
+ return type;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DeleteStorageGroupPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DeleteStorageGroupPlan.java
index 600b056..bb33050 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DeleteStorageGroupPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DeleteStorageGroupPlan.java
@@ -18,6 +18,10 @@
*/
package org.apache.iotdb.db.qp.physical.sys;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.List;
import org.apache.iotdb.db.qp.logical.Operator;
@@ -32,11 +36,43 @@ public class DeleteStorageGroupPlan extends PhysicalPlan {
super(false, Operator.OperatorType.DELETE_STORAGE_GROUP);
this.deletePathList = deletePathList;
}
+
+ public DeleteStorageGroupPlan() {
+ super(false, Operator.OperatorType.DELETE_STORAGE_GROUP);
+ }
@Override
public List<Path> getPaths() {
return deletePathList;
}
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ int type = PhysicalPlan.PhysicalPlanType.DELETE_STORAGE_GROUP.ordinal();
+ stream.writeByte((byte) type);
+ stream.writeInt(this.getPaths().size());
+ for (Path path : this.getPaths()) {
+ putString(stream, path.getFullPath());
+ }
+ }
+
+ @Override
+ public void serialize(ByteBuffer buffer) {
+ int type = PhysicalPlanType.DELETE_STORAGE_GROUP.ordinal();
+ buffer.put((byte) type);
+ buffer.putInt(this.getPaths().size());
+ for (Path path : this.getPaths()) {
+ putString(buffer, path.getFullPath());
+ }
+ }
+
+ @Override
+ public void deserialize(ByteBuffer buffer) {
+ int pathNum = buffer.getInt();
+ this.deletePathList = new ArrayList<>();
+ for (int i = 0; i < pathNum; i++) {
+ deletePathList.add(new Path(readString(buffer)));
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DeleteTimeSeriesPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DeleteTimeSeriesPlan.java
index 48611c9..80da6d5 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DeleteTimeSeriesPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DeleteTimeSeriesPlan.java
@@ -18,25 +18,59 @@
*/
package org.apache.iotdb.db.qp.physical.sys;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.List;
-
import org.apache.iotdb.db.qp.logical.Operator;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.tsfile.read.common.Path;
public class DeleteTimeSeriesPlan extends PhysicalPlan {
-
+
private List<Path> deletePathList;
-
- public DeleteTimeSeriesPlan (List<Path> deletePathList) {
- super(false, Operator.OperatorType.DELETE_TIMESERIES);
- this.deletePathList = deletePathList;
+
+ public DeleteTimeSeriesPlan(List<Path> deletePathList) {
+ super(false, Operator.OperatorType.DELETE_TIMESERIES);
+ this.deletePathList = deletePathList;
}
-
+
+ public DeleteTimeSeriesPlan() {
+ super(false, Operator.OperatorType.DELETE_TIMESERIES);
+ }
+
@Override
public List<Path> getPaths() {
return deletePathList;
}
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ int type = PhysicalPlanType.DELETE_TIMESERIES.ordinal();
+ stream.writeByte((byte) type);
+ stream.writeInt(deletePathList.size());
+ for (Path path : deletePathList) {
+ putString(stream, path.getFullPath());
+ }
+ }
+
+ @Override
+ public void serialize(ByteBuffer buffer) {
+ int type = PhysicalPlanType.DELETE_TIMESERIES.ordinal();
+ buffer.put((byte) type);
+ buffer.putInt(deletePathList.size());
+ for (Path path : deletePathList) {
+ putString(buffer, path.getFullPath());
+ }
+ }
+ @Override
+ public void deserialize(ByteBuffer buffer) {
+ int pathNumber = buffer.getInt();
+ deletePathList = new ArrayList<>();
+ for (int i = 0; i < pathNumber; i++) {
+ deletePathList.add(new Path(readString(buffer)));
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/LoadConfigurationPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/LoadConfigurationPlan.java
index 5727ca4..f4b2010 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/LoadConfigurationPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/LoadConfigurationPlan.java
@@ -17,19 +17,100 @@
* under the License.
*
*/
+
package org.apache.iotdb.db.qp.physical.sys;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.tsfile.read.common.Path;
-
-import java.util.List;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
public class LoadConfigurationPlan extends PhysicalPlan {
+ // an array of properties, the size of which is always 2.
+ // The first element is the properties for iotdb-engine, the second
+ // is for cluster-config
+ private Properties[] propertiesArray;
+
+ LoadConfigurationPlanType loadConfigurationPlanType;
+
+ public LoadConfigurationPlan(LoadConfigurationPlanType loadConfigurationPlanType,
+ Properties[] propertiesArray)
+ throws QueryProcessException {
+ super(false, OperatorType.LOAD_CONFIGURATION);
+ if (loadConfigurationPlanType != LoadConfigurationPlanType.GLOBAL) {
+ throw new QueryProcessException(
+ "The constructor with 2 parameters is for load global configuration");
+ }
+ if (propertiesArray.length != 2) {
+ throw new QueryProcessException("The size of propertiesArray is not 2.");
+ }
+ this.loadConfigurationPlanType = loadConfigurationPlanType;
+ this.propertiesArray = propertiesArray;
+ }
+
+ public LoadConfigurationPlan(LoadConfigurationPlanType loadConfigurationPlanType)
+ throws QueryProcessException {
+ super(false, OperatorType.LOAD_CONFIGURATION);
+ if (loadConfigurationPlanType != LoadConfigurationPlanType.LOCAL) {
+ throw new QueryProcessException(
+ "The constructor with 1 parameters is for load local configuration");
+ }
+ this.loadConfigurationPlanType = loadConfigurationPlanType;
+ }
+
+ // only for deserialize
public LoadConfigurationPlan() {
- super(false);
- this.setOperatorType(OperatorType.LOAD_CONFIGURATION);
+ super(false, OperatorType.LOAD_CONFIGURATION);
+ }
+
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ int type = PhysicalPlanType.LOAD_CONFIGURATION.ordinal();
+ stream.writeByte((byte) type);
+ stream.writeInt(loadConfigurationPlanType.ordinal());
+ if (loadConfigurationPlanType == LoadConfigurationPlanType.GLOBAL) {
+ stream.writeInt(propertiesArray.length);
+ for (Properties properties : propertiesArray) {
+ if (properties == null) {
+ stream.writeInt(0);
+ } else {
+ stream.writeInt(1);
+ stream.writeInt(properties.entrySet().size());
+ for (Entry<Object, Object> entry : properties.entrySet()) {
+ putString(stream, String.valueOf(entry.getKey()));
+ putString(stream, String.valueOf(entry.getValue()));
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public void deserialize(ByteBuffer buffer) {
+ loadConfigurationPlanType = LoadConfigurationPlanType.values()[buffer.getInt()];
+ if (loadConfigurationPlanType == LoadConfigurationPlanType.GLOBAL) {
+ int propertiesNum = buffer.getInt();
+ propertiesArray = new Properties[propertiesNum];
+ for (int i = 0; i < propertiesArray.length; i++) {
+ if (buffer.getInt() == 1) {
+ propertiesArray[i] = new Properties();
+ int size = buffer.getInt();
+ for (int j = 0; j < size; j++) {
+ propertiesArray[i].setProperty(readString(buffer), readString(buffer));
+ }
+ }
+ }
+ }
}
@Override
@@ -41,4 +122,20 @@ public class LoadConfigurationPlan extends PhysicalPlan {
public String toString() {
return getOperatorType().toString();
}
+
+ public Properties getIoTDBProperties() {
+ return propertiesArray[0];
+ }
+
+ public Properties getClusterProperties() {
+ return propertiesArray[1];
+ }
+
+ public LoadConfigurationPlanType getLoadConfigurationPlanType() {
+ return loadConfigurationPlanType;
+ }
+
+ public enum LoadConfigurationPlanType {
+ GLOBAL, LOCAL
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/SetStorageGroupPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/SetStorageGroupPlan.java
index 28d19cd..42f06ea 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/SetStorageGroupPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/SetStorageGroupPlan.java
@@ -58,7 +58,7 @@ public class SetStorageGroupPlan extends PhysicalPlan {
}
@Override
- public void serializeTo(DataOutputStream stream) throws IOException {
+ public void serialize(DataOutputStream stream) throws IOException {
stream.write((byte) PhysicalPlanType.SET_STORAGE_GROUP.ordinal());
byte[] fullPathBytes = path.getFullPath().getBytes();
stream.writeInt(fullPathBytes.length);
@@ -66,7 +66,7 @@ public class SetStorageGroupPlan extends PhysicalPlan {
}
@Override
- public void deserializeFrom(ByteBuffer buffer) {
+ public void deserialize(ByteBuffer buffer) {
int length = buffer.getInt();
byte[] fullPathBytes = new byte[length];
buffer.get(fullPathBytes);
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/SetTTLPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/SetTTLPlan.java
index 268d554..2600830 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/SetTTLPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/SetTTLPlan.java
@@ -20,6 +20,9 @@
package org.apache.iotdb.db.qp.physical.sys;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.List;
import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
@@ -30,6 +33,10 @@ public class SetTTLPlan extends PhysicalPlan {
private String storageGroup;
private long dataTTL;
+ public SetTTLPlan() {
+ super(false, OperatorType.TTL);
+ }
+
public SetTTLPlan(String storageGroup, long dataTTL) {
// set TTL
super(false, OperatorType.TTL);
@@ -47,6 +54,28 @@ public class SetTTLPlan extends PhysicalPlan {
return null;
}
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ int type = PhysicalPlanType.TTL.ordinal();
+ stream.writeByte((byte) type);
+ stream.writeLong(dataTTL);
+ putString(stream, storageGroup);
+ }
+
+ @Override
+ public void serialize(ByteBuffer buffer) {
+ int type = PhysicalPlanType.TTL.ordinal();
+ buffer.put((byte) type);
+ buffer.putLong(dataTTL);
+ putString(buffer, storageGroup);
+ }
+
+ @Override
+ public void deserialize(ByteBuffer buffer) {
+ this.dataTTL = buffer.getLong();
+ this.storageGroup = readString(buffer);
+ }
+
public String getStorageGroup() {
return storageGroup;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowChildPathsPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowChildPathsPlan.java
index 8490c6b..60c41ec 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowChildPathsPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowChildPathsPlan.java
@@ -29,7 +29,7 @@ public class ShowChildPathsPlan extends ShowPlan {
public ShowChildPathsPlan(ShowContentType showContentType, Path prefixPath) {
super(showContentType);
this.prefixPath = prefixPath;
- canbeSplit = false;
+ canBeSplit = false;
}
public Path getPath() {
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowTimeSeriesPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowTimeSeriesPlan.java
index fbf8fcb..92e764c 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowTimeSeriesPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/ShowTimeSeriesPlan.java
@@ -18,6 +18,9 @@
*/
package org.apache.iotdb.db.qp.physical.sys;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
import org.apache.iotdb.tsfile.read.common.Path;
public class ShowTimeSeriesPlan extends ShowPlan {
@@ -47,6 +50,10 @@ public class ShowTimeSeriesPlan extends ShowPlan {
this.offset = offset;
}
+ public ShowTimeSeriesPlan() {
+ super(ShowContentType.TIMESERIES);
+ }
+
public Path getPath() {
return this.path;
}
@@ -70,4 +77,28 @@ public class ShowTimeSeriesPlan extends ShowPlan {
public int getOffset() {
return offset;
}
+
+ @Override
+ public void serialize(DataOutputStream outputStream) throws IOException {
+ outputStream.write(PhysicalPlanType.SHOW_TIMESERIES.ordinal());
+
+ putString(outputStream, path.getFullPath());
+ outputStream.writeBoolean(isContains);
+ putString(outputStream, key);
+ putString(outputStream, value);
+
+ outputStream.writeInt(limit);
+ outputStream.writeInt(offset);
+ }
+
+ @Override
+ public void deserialize(ByteBuffer buffer) {
+ path = new Path(readString(buffer));
+ isContains = buffer.get() == 1;
+ key = readString(buffer);
+ value = readString(buffer);
+
+ limit = buffer.getInt();
+ limit = buffer.getInt();
+ }
}
\ No newline at end of file
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 e3a7cee..53de61f 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
@@ -29,6 +29,7 @@ import org.apache.iotdb.db.qp.logical.crud.*;
import org.apache.iotdb.db.qp.logical.sys.*;
import org.apache.iotdb.db.qp.logical.sys.AlterTimeSeriesOperator.AlterType;
import org.apache.iotdb.db.qp.logical.sys.AuthorOperator.AuthorType;
+import org.apache.iotdb.db.qp.logical.sys.LoadConfigurationOperator.LoadConfigurationOperatorType;
import org.apache.iotdb.db.qp.strategy.SqlBaseParser.*;
import org.apache.iotdb.db.query.executor.fill.IFill;
import org.apache.iotdb.db.query.executor.fill.LinearFill;
@@ -190,7 +191,12 @@ public class LogicalGenerator extends SqlBaseBaseListener {
@Override
public void enterLoadConfigurationStatement(LoadConfigurationStatementContext ctx) {
super.enterLoadConfigurationStatement(ctx);
- initializedOperator = new LoadConfigurationOperator();
+ if(ctx.GLOBAL()!=null){
+ initializedOperator = new LoadConfigurationOperator(LoadConfigurationOperatorType.GLOBAL);
+ }else{
+ initializedOperator = new LoadConfigurationOperator(LoadConfigurationOperatorType.LOCAL);
+ }
+
}
@Override
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 fb34866..e6053b8 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
@@ -47,6 +47,8 @@ import org.apache.iotdb.db.qp.logical.sys.CreateTimeSeriesOperator;
import org.apache.iotdb.db.qp.logical.sys.DataAuthOperator;
import org.apache.iotdb.db.qp.logical.sys.DeleteStorageGroupOperator;
import org.apache.iotdb.db.qp.logical.sys.DeleteTimeSeriesOperator;
+import org.apache.iotdb.db.qp.logical.sys.LoadConfigurationOperator;
+import org.apache.iotdb.db.qp.logical.sys.LoadConfigurationOperator.LoadConfigurationOperatorType;
import org.apache.iotdb.db.qp.logical.sys.FlushOperator;
import org.apache.iotdb.db.qp.logical.sys.LoadDataOperator;
import org.apache.iotdb.db.qp.logical.sys.LoadFilesOperator;
@@ -77,9 +79,10 @@ import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.DataAuthPlan;
import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan;
import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.FlushPlan;
import org.apache.iotdb.db.qp.physical.sys.LoadConfigurationPlan;
+import org.apache.iotdb.db.qp.physical.sys.LoadConfigurationPlan.LoadConfigurationPlanType;
import org.apache.iotdb.db.qp.physical.sys.LoadDataPlan;
+import org.apache.iotdb.db.qp.physical.sys.FlushPlan;
import org.apache.iotdb.db.qp.physical.sys.MergePlan;
import org.apache.iotdb.db.qp.physical.sys.OperateFilePlan;
import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
@@ -211,7 +214,9 @@ public class PhysicalGenerator {
"not supported operator type %s in ttl operation.", operator.getType()));
}
case LOAD_CONFIGURATION:
- return new LoadConfigurationPlan();
+ LoadConfigurationOperatorType type = ((LoadConfigurationOperator) operator)
+ .getLoadConfigurationOperatorType();
+ return generateLoadConfigurationPlan(type);
case SHOW:
switch (operator.getTokenIntType()) {
case SQLConstant.TOK_DYNAMIC_PARAMETER:
@@ -274,8 +279,23 @@ public class PhysicalGenerator {
}
}
- protected List<TSDataType> getSeriesTypes(List<String> paths, String aggregation)
- throws MetadataException {
+
+ protected PhysicalPlan generateLoadConfigurationPlan(LoadConfigurationOperatorType type)
+ throws QueryProcessException {
+ switch (type) {
+ case GLOBAL:
+ return new LoadConfigurationPlan(LoadConfigurationPlanType.GLOBAL);
+ case LOCAL:
+ return new LoadConfigurationPlan(LoadConfigurationPlanType.LOCAL);
+ default:
+ throw new QueryProcessException(
+ String.format("Unrecognized load configuration operator type, %s", type.name()));
+ }
+
+ }
+
+ protected List<TSDataType> getSeriesTypes(List<String> paths,
+ String aggregation) throws MetadataException {
return SchemaUtils.getSeriesTypesByString(paths, aggregation);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregateResult.java b/server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregateResult.java
index 0142322..832e69a 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregateResult.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregateResult.java
@@ -19,6 +19,9 @@
package org.apache.iotdb.db.query.aggregation;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.query.factory.AggregateResultFactory;
import org.apache.iotdb.db.query.reader.series.IReaderByTimestamp;
@@ -29,10 +32,6 @@ import org.apache.iotdb.tsfile.read.common.BatchData;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.nio.ByteBuffer;
-
public abstract class AggregateResult {
private final AggregationType aggregationType;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/control/FileReaderManager.java b/server/src/main/java/org/apache/iotdb/db/query/control/FileReaderManager.java
index f217a79..0961788 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/control/FileReaderManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/control/FileReaderManager.java
@@ -174,7 +174,7 @@ public class FileReaderManager implements IService {
* of a reader equals zero, the reader can be closed and removed.
*/
void increaseFileReaderReference(TsFileResource tsFile, boolean isClosed) {
- tsFile.getWriteQueryLock().readLock().lock();
+ tsFile.readLock();
synchronized (this) {
if (!isClosed) {
unclosedReferenceMap.computeIfAbsent(tsFile.getPath(), k -> new AtomicInteger()).getAndIncrement();
@@ -196,7 +196,7 @@ public class FileReaderManager implements IService {
closedReferenceMap.get(tsFile.getPath()).decrementAndGet();
}
}
- tsFile.getWriteQueryLock().readLock().unlock();
+ tsFile.readUnlock();
}
/**
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java
index 5b4e9b0..992bd4f 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/AlignByDeviceDataSet.java
@@ -119,13 +119,8 @@ public class AlignByDeviceDataSet extends QueryDataSet {
while (deviceIterator.hasNext()) {
currentDevice = deviceIterator.next();
// get all measurements of current device
- Set<String> measurementOfGivenDevice;
- try {
- MNode deviceNode = MManager.getInstance().getNodeByPath(currentDevice);
- measurementOfGivenDevice = deviceNode.getChildren().keySet();
- } catch (MetadataException e) {
- throw new IOException("Cannot get node from " + currentDevice);
- }
+ Set<String> measurementOfGivenDevice = getDeviceMeasurements(currentDevice);
+
// extract paths and aggregations queried from all measurements
// executeColumns is for calculating rowRecord
executeColumns = new ArrayList<>();
@@ -193,6 +188,15 @@ public class AlignByDeviceDataSet extends QueryDataSet {
return false;
}
+ protected Set<String> getDeviceMeasurements(String device) throws IOException {
+ try {
+ MNode deviceNode = MManager.getInstance().getNodeByPath(currentDevice);
+ return deviceNode.getChildren().keySet();
+ } catch (MetadataException e) {
+ throw new IOException("Cannot get node from " + currentDevice, e);
+ }
+ }
+
protected RowRecord nextWithoutConstraint() throws IOException {
RowRecord originRowRecord = currentDataSet.next();
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/ShowTimeSeriesResult.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/ShowTimeSeriesResult.java
index 73b65aa..5874401 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/ShowTimeSeriesResult.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/ShowTimeSeriesResult.java
@@ -18,9 +18,15 @@
*/
package org.apache.iotdb.db.query.dataset;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
import java.util.Map;
+import java.util.Map.Entry;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-public class ShowTimeSeriesResult {
+public class ShowTimeSeriesResult implements Comparable<ShowTimeSeriesResult> {
private String name;
private String alias;
@@ -41,6 +47,10 @@ public class ShowTimeSeriesResult {
this.tagAndAttribute = tagAndAttribute;
}
+ public ShowTimeSeriesResult() {
+
+ }
+
public String getName() {
return name;
}
@@ -68,4 +78,53 @@ public class ShowTimeSeriesResult {
public Map<String, String> getTagAndAttribute() {
return tagAndAttribute;
}
+
+ @Override
+ public int compareTo(ShowTimeSeriesResult o) {
+ return this.name.compareTo(o.name);
+ }
+
+ public void serialize(OutputStream outputStream) throws IOException {
+ ReadWriteIOUtils.write(name, outputStream);
+ ReadWriteIOUtils.write(alias != null, outputStream);
+ if (alias != null) {
+ ReadWriteIOUtils.write(alias, outputStream);
+ }
+ ReadWriteIOUtils.write(sgName, outputStream);
+ ReadWriteIOUtils.write(dataType, outputStream);
+ ReadWriteIOUtils.write(encoding, outputStream);
+ ReadWriteIOUtils.write(compressor, outputStream);
+
+ ReadWriteIOUtils.write(tagAndAttribute != null, outputStream);
+ if (tagAndAttribute != null) {
+ ReadWriteIOUtils.write(tagAndAttribute.size(), outputStream);
+ for (Entry<String, String> stringStringEntry : tagAndAttribute.entrySet()) {
+ ReadWriteIOUtils.write(stringStringEntry.getKey(), outputStream);
+ ReadWriteIOUtils.write(stringStringEntry.getValue(), outputStream);
+ }
+ }
+ }
+
+ public static ShowTimeSeriesResult deserialize(ByteBuffer buffer) {
+ ShowTimeSeriesResult result = new ShowTimeSeriesResult();
+ result.name = ReadWriteIOUtils.readString(buffer);
+ if (buffer.get() == 1) {
+ result.alias = ReadWriteIOUtils.readString(buffer);
+ }
+ result.sgName = ReadWriteIOUtils.readString(buffer);
+ result.dataType = ReadWriteIOUtils.readString(buffer);
+ result.encoding = ReadWriteIOUtils.readString(buffer);
+ result.compressor = ReadWriteIOUtils.readString(buffer);
+
+ if (buffer.get() == 1) {
+ int tagSize = buffer.getInt();
+ result.tagAndAttribute = new HashMap<>(tagSize);
+ for (int i = 0; i < tagSize; i++) {
+ String key = ReadWriteIOUtils.readString(buffer);
+ String value = ReadWriteIOUtils.readString(buffer);
+ result.tagAndAttribute.put(key, value);
+ }
+ }
+ return result;
+ }
}
\ No newline at end of file
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor.java
index b69f90f..d9b60a7 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/LocalGroupByExecutor.java
@@ -49,11 +49,13 @@ public class LocalGroupByExecutor implements GroupByExecutor {
private List<AggregateResult> results = new ArrayList<>();
private TimeRange timeRange;
+ private QueryDataSource queryDataSource;
+
public LocalGroupByExecutor(Path path, Set<String> allSensors, TSDataType dataType,
QueryContext context, Filter timeFilter, TsFileFilter fileFilter)
throws StorageEngineException, QueryProcessException {
- QueryDataSource queryDataSource =
- QueryResourceManager.getInstance().getQueryDataSource(path, context, timeFilter);
+ queryDataSource = QueryResourceManager.getInstance()
+ .getQueryDataSource(path, context, timeFilter);
// update filter by TTL
timeFilter = queryDataSource.updateFilterUsingTTL(timeFilter);
this.reader = new SeriesAggregateReader(path, allSensors, dataType, context, queryDataSource,
@@ -62,6 +64,10 @@ public class LocalGroupByExecutor implements GroupByExecutor {
timeRange = new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE);
}
+ public boolean isEmpty() {
+ return queryDataSource.getSeqResources().isEmpty() && queryDataSource.getUnseqResources().isEmpty();
+ }
+
@Override
public void addAggregateResult(AggregateResult aggrResult) {
results.add(aggrResult);
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/FillQueryExecutor.java b/server/src/main/java/org/apache/iotdb/db/query/executor/FillQueryExecutor.java
index 34ea654..10864b6 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/FillQueryExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/FillQueryExecutor.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.query.executor;
+import java.util.Set;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
@@ -85,7 +86,7 @@ public class FillQueryExecutor {
} else {
fill = typeIFillMap.get(dataType).copy();
}
- fill.configureFill(path, dataType, queryTime,
+ fill = configureFill(fill, path, dataType, queryTime,
fillQueryPlan.getAllMeasurementsInDevice(path.getDevice()), context);
TimeValuePair timeValuePair = fill.getFillResult();
@@ -100,4 +101,11 @@ public class FillQueryExecutor {
dataSet.setRecord(record);
return dataSet;
}
+
+ protected IFill configureFill(IFill fill, Path path, TSDataType dataType, long queryTime,
+ Set<String> deviceMeasurements, QueryContext context)
+ throws QueryProcessException, StorageEngineException {
+ fill.configureFill(path, dataType, queryTime, deviceMeasurements, context);
+ return fill;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/fill/IFill.java b/server/src/main/java/org/apache/iotdb/db/query/executor/fill/IFill.java
index 47bd394..951cdad 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/fill/IFill.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/fill/IFill.java
@@ -28,11 +28,12 @@ import org.apache.iotdb.tsfile.read.common.Path;
import java.io.IOException;
import java.util.Set;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
public abstract class IFill {
- long queryTime;
- TSDataType dataType;
+ protected long queryTime;
+ protected TSDataType dataType;
public IFill(TSDataType dataType, long queryTime) {
this.dataType = dataType;
@@ -45,7 +46,7 @@ public abstract class IFill {
public abstract IFill copy();
public abstract void configureFill(Path path, TSDataType dataType, long queryTime,
- Set<String> sensors, QueryContext context);
+ Set<String> deviceMeasurements, QueryContext context);
public abstract TimeValuePair getFillResult()
throws IOException, QueryProcessException, StorageEngineException;
@@ -62,5 +63,9 @@ public abstract class IFill {
this.queryTime = queryTime;
}
+ public long getQueryTime() {
+ return queryTime;
+ }
+
abstract void constructFilter();
}
\ No newline at end of file
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/fill/LastPointReader.java b/server/src/main/java/org/apache/iotdb/db/query/executor/fill/LastPointReader.java
index 2e59c31..a2998de 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/fill/LastPointReader.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/fill/LastPointReader.java
@@ -45,7 +45,10 @@ public class LastPointReader {
long queryTime;
TSDataType dataType;
private QueryContext context;
- private Set<String> allSensors;
+
+ // measurements of the same device as "seriesPath"
+ private Set<String> deviceMeasurements;
+
private Filter timeFilter;
private QueryDataSource dataSource;
@@ -56,14 +59,14 @@ public class LastPointReader {
}
- public LastPointReader(Path seriesPath, TSDataType dataType, Set<String> sensors,
+ public LastPointReader(Path seriesPath, TSDataType dataType, Set<String> deviceMeasurements,
QueryContext context, QueryDataSource dataSource, long queryTime, Filter timeFilter) {
this.seriesPath = seriesPath;
this.dataType = dataType;
this.dataSource = dataSource;
this.context = context;
this.queryTime = queryTime;
- this.allSensors = sensors;
+ this.deviceMeasurements = deviceMeasurements;
this.timeFilter = timeFilter;
}
@@ -94,7 +97,7 @@ public class LastPointReader {
TsFileResource resource = seqFileResource.get(index);
TimeseriesMetadata timeseriesMetadata =
FileLoaderUtils.loadTimeSeriesMetadata(
- resource, seriesPath, context, timeFilter, allSensors);
+ resource, seriesPath, context, timeFilter, deviceMeasurements);
if (timeseriesMetadata != null) {
if (endtimeContainedByTimeFilter(timeseriesMetadata.getStatistics())) {
return constructLastPair(
@@ -129,7 +132,7 @@ public class LastPointReader {
&& (lBoundTime <= unseqFileResource.peek().getEndTimeMap().get(seriesPath.getDevice()))) {
TimeseriesMetadata timeseriesMetadata =
FileLoaderUtils.loadTimeSeriesMetadata(
- unseqFileResource.poll(), seriesPath, context, timeFilter, allSensors);
+ unseqFileResource.poll(), seriesPath, context, timeFilter, deviceMeasurements);
if (timeseriesMetadata == null || (!timeseriesMetadata.isModified()
&& timeseriesMetadata.getStatistics().getEndTime() < lBoundTime)) {
@@ -209,7 +212,9 @@ public class LastPointReader {
return Long.compare(o2.getVersion(), o1.getVersion());
});
for (TimeseriesMetadata timeseriesMetadata : unseqTimeseriesMetadataList) {
- chunkMetadataList.addAll(timeseriesMetadata.loadChunkMetadataList());
+ if (timeseriesMetadata != null) {
+ chunkMetadataList.addAll(timeseriesMetadata.loadChunkMetadataList());
+ }
}
return chunkMetadataList;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/fill/LinearFill.java b/server/src/main/java/org/apache/iotdb/db/query/executor/fill/LinearFill.java
index a4aa19b..4dbac70 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/fill/LinearFill.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/fill/LinearFill.java
@@ -44,13 +44,14 @@ import java.io.IOException;
public class LinearFill extends IFill {
- private Path seriesPath;
- private long beforeRange;
- private long afterRange;
- private Filter beforeFilter;
- private Filter afterFilter;
- private QueryContext context;
- private Set<String> allSensors;
+ protected Path seriesPath;
+ protected long beforeRange;
+ protected long afterRange;
+ protected Filter beforeFilter;
+ protected Filter afterFilter;
+ protected QueryContext context;
+ // all measurements sharing the same device as "seriesPath"
+ protected Set<String> deviceMeasurements;
public LinearFill(long beforeRange, long afterRange) {
this.beforeRange = beforeRange;
@@ -106,20 +107,16 @@ public class LinearFill extends IFill {
this.dataType = dataType;
this.queryTime = queryTime;
this.context = context;
- this.allSensors = sensors;
+ this.deviceMeasurements = sensors;
constructFilter();
}
@Override
public TimeValuePair getFillResult()
throws IOException, QueryProcessException, StorageEngineException {
- QueryDataSource dataSource =
- QueryResourceManager.getInstance().getQueryDataSource(seriesPath, context, beforeFilter);
- LastPointReader lastReader =
- new LastPointReader(seriesPath, dataType, allSensors, context, dataSource, queryTime, beforeFilter);
- TimeValuePair beforePair = lastReader.readLastPoint();
- TimeValuePair afterPair = calculateFirstPointAfterQueryTime();
+ TimeValuePair beforePair = calculatePrecedingPoint();
+ TimeValuePair afterPair = calculateSucceedingPoint();
// no before data or has data on the query timestamp
if (beforePair.getValue() == null || beforePair.getTimestamp() == queryTime) {
@@ -136,7 +133,17 @@ public class LinearFill extends IFill {
return average(beforePair, afterPair);
}
- private TimeValuePair calculateFirstPointAfterQueryTime()
+ protected TimeValuePair calculatePrecedingPoint()
+ throws QueryProcessException, StorageEngineException, IOException {
+ QueryDataSource dataSource =
+ QueryResourceManager.getInstance().getQueryDataSource(seriesPath, context, beforeFilter);
+ LastPointReader lastReader =
+ new LastPointReader(seriesPath, dataType, deviceMeasurements, context, dataSource, queryTime, beforeFilter);
+
+ return lastReader.readLastPoint();
+ }
+
+ protected TimeValuePair calculateSucceedingPoint()
throws IOException, StorageEngineException, QueryProcessException {
TimeValuePair result = new TimeValuePair(0, null);
@@ -146,7 +153,7 @@ public class LinearFill extends IFill {
aggregateResultList.add(minTimeResult);
aggregateResultList.add(firstValueResult);
AggregationExecutor.aggregateOneSeries(
- seriesPath, allSensors, context, afterFilter, dataType, aggregateResultList, null);
+ seriesPath, deviceMeasurements, context, afterFilter, dataType, aggregateResultList, null);
if (minTimeResult.getResult() != null) {
long timestamp = (long)(minTimeResult.getResult());
diff --git a/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesReader.java b/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesReader.java
index 0e04e23..c7c3210 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesReader.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/reader/series/SeriesReader.java
@@ -41,7 +41,7 @@ import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;
-class SeriesReader {
+public class SeriesReader {
private final Path seriesPath;
@@ -100,7 +100,7 @@ class SeriesReader {
private boolean hasCachedNextOverlappedPage;
private BatchData cachedBatchData;
- SeriesReader(Path seriesPath, Set<String> allSensors, TSDataType dataType, QueryContext context,
+ public SeriesReader(Path seriesPath, Set<String> allSensors, TSDataType dataType, QueryContext context,
QueryDataSource dataSource, Filter timeFilter, Filter valueFilter, TsFileFilter fileFilter) {
this.seriesPath = seriesPath;
this.allSensors = allSensors;
@@ -127,6 +127,10 @@ class SeriesReader {
this.valueFilter = valueFilter;
}
+ public boolean isEmpty() {
+ return seqFileResource.isEmpty() && unseqFileResource.isEmpty();
+ }
+
boolean hasNextFile() throws IOException {
if (!cachedPageReaders.isEmpty()
diff --git a/server/src/main/java/org/apache/iotdb/db/service/JMXService.java b/server/src/main/java/org/apache/iotdb/db/service/JMXService.java
index 5cf0e51..7993151 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/JMXService.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/JMXService.java
@@ -85,7 +85,7 @@ public class JMXService implements IService {
public void start() throws StartupException {
String jmxPort = System.getProperty(IoTDBConstant.IOTDB_JMX_PORT);
if (jmxPort == null) {
- logger.warn("{} JMX port is undefined", this.getID().getName());
+ logger.debug("{} JMX port is undefined", this.getID().getName());
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/service/RPCService.java b/server/src/main/java/org/apache/iotdb/db/service/RPCService.java
index 52b4b79..c0fddf5 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/RPCService.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/RPCService.java
@@ -34,6 +34,7 @@ import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TProtocolFactory;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TThreadPoolServer;
+import org.apache.thrift.transport.TFastFramedTransport;
import org.apache.thrift.transport.TServerSocket;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
@@ -216,6 +217,7 @@ public class RPCService implements RPCServiceMBean, IService {
ThreadName.RPC_CLIENT.getName());
poolArgs.processor(processor);
poolArgs.protocolFactory(protocolFactory);
+ poolArgs.transportFactory(new TFastFramedTransport.Factory());
poolServer = new TThreadPoolServer(poolArgs);
poolServer.serve();
} catch (TTransportException e) {
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 bfce48f..28ed40a 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
@@ -858,7 +858,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
*/
private QueryDataSet createQueryDataSet(long queryId, PhysicalPlan physicalPlan)
throws QueryProcessException, QueryFilterOptimizationException, StorageEngineException,
- IOException, MetadataException, SQLException {
+ IOException, MetadataException, SQLException, TException, InterruptedException {
QueryContext context = genQueryContext(queryId);
QueryDataSet queryDataSet = executor.processQuery(physicalPlan, context);
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/OpenFileNumUtil.java b/server/src/main/java/org/apache/iotdb/db/utils/OpenFileNumUtil.java
index 566331c..7997f4c 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/OpenFileNumUtil.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/OpenFileNumUtil.java
@@ -216,7 +216,7 @@ public class OpenFileNumUtil {
}
/**
- * Check if runtime OS is supported then return the result list. If pid is abnormal then all
+ * Check if runtime OS is supported then return the result list. If pid is abnormal then all
* statistics returns -1, if OS is not supported then all statistics returns -2
*
* @return map
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/SerializeUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/SerializeUtils.java
new file mode 100644
index 0000000..d4af3c2
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/utils/SerializeUtils.java
@@ -0,0 +1,492 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.utils;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import javax.activation.UnsupportedDataTypeException;
+import org.apache.iotdb.cluster.rpc.thrift.Node;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.TimeValuePair;
+import org.apache.iotdb.tsfile.read.common.BatchData;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
+
+public class SerializeUtils {
+
+ private SerializeUtils() {
+ // util class
+ }
+
+ public static void serialize(String str, DataOutputStream dataOutputStream) {
+ try {
+ byte[] strBytes = str.getBytes(TSFileConfig.STRING_CHARSET);
+ dataOutputStream.writeInt(strBytes.length);
+ dataOutputStream.write(strBytes);
+ } catch (IOException e) {
+ // unreachable
+ }
+ }
+
+ public static String deserializeString(ByteBuffer buffer) {
+ int length = buffer.getInt();
+ byte[] strBytes = new byte[length];
+ buffer.get(strBytes);
+ return new String(strBytes);
+ }
+
+ public static void serializeStringList(List<String> strs, DataOutputStream dataOutputStream) {
+ try {
+ dataOutputStream.writeInt(strs.size());
+ } catch (IOException e) {
+ // unreachable
+ }
+ for (String str : strs) {
+ serialize(str, dataOutputStream);
+ }
+ }
+
+ public static List<String> deserializeStringList(ByteBuffer buffer) {
+ int length = buffer.getInt();
+ List<String> result = new ArrayList<>(length);
+ for (int i = 0; i < length; i++) {
+ result.add(deserializeString(buffer));
+ }
+ return result;
+ }
+
+ public static void serialize(List<Integer> ints, DataOutputStream dataOutputStream) {
+ try {
+ dataOutputStream.writeInt(ints.size());
+ for (Integer anInt : ints) {
+ dataOutputStream.writeInt(anInt);
+ }
+ } catch (IOException e) {
+ // unreachable
+ }
+ }
+
+ public static void deserialize(List<Integer> ints, ByteBuffer buffer) {
+ int length = buffer.getInt();
+ for (int i = 0; i < length; i++) {
+ ints.add(buffer.getInt());
+ }
+ }
+
+ public static void serialize(Set<Integer> ints, DataOutputStream dataOutputStream) {
+ try {
+ dataOutputStream.writeInt(ints.size());
+ for (Integer anInt : ints) {
+ dataOutputStream.writeInt(anInt);
+ }
+ } catch (IOException e) {
+ // unreachable
+ }
+ }
+
+ public static void deserialize(Set<Integer> ints, ByteBuffer buffer) {
+ int length = buffer.getInt();
+ for (int i = 0; i < length; i++) {
+ ints.add(buffer.getInt());
+ }
+ }
+
+ public static void serialize(Node node, DataOutputStream dataOutputStream) {
+ try {
+ byte[] ipBytes = node.ip.getBytes();
+ dataOutputStream.writeInt(ipBytes.length);
+ dataOutputStream.write(ipBytes);
+ dataOutputStream.writeInt(node.metaPort);
+ dataOutputStream.writeInt(node.nodeIdentifier);
+ dataOutputStream.writeInt(node.dataPort);
+ } catch (IOException e) {
+ // unreachable
+ }
+ }
+
+ public static void deserialize(Node node, ByteBuffer buffer) {
+ int ipLength = buffer.getInt();
+ byte[] ipBytes = new byte[ipLength];
+ buffer.get(ipBytes);
+ node.setIp(new String(ipBytes));
+ node.setMetaPort(buffer.getInt());
+ node.setNodeIdentifier(buffer.getInt());
+ node.setDataPort(buffer.getInt());
+ }
+
+ public static void deserialize(Node node, DataInputStream stream) throws IOException {
+ int ipLength = stream.readInt();
+ byte[] ipBytes = new byte[ipLength];
+ stream.read(ipBytes);
+ node.setIp(new String(ipBytes));
+ node.setMetaPort(stream.readInt());
+ node.setNodeIdentifier(stream.readInt());
+ node.setDataPort(stream.readInt());
+ }
+
+ public static void serializeBatchData(BatchData batchData, DataOutputStream outputStream) {
+ try {
+ int length = batchData.length();
+ TSDataType dataType = batchData.getDataType();
+ outputStream.writeInt(length);
+ outputStream.write(dataType.ordinal());
+ switch (dataType) {
+ case BOOLEAN:
+ for (int i = 0; i < length; i++) {
+ outputStream.writeLong(batchData.getTimeByIndex(i));
+ outputStream.writeBoolean(batchData.getBooleanByIndex(i));
+ }
+ break;
+ case DOUBLE:
+ for (int i = 0; i < length; i++) {
+ outputStream.writeLong(batchData.getTimeByIndex(i));
+ outputStream.writeDouble(batchData.getDoubleByIndex(i));
+ }
+ break;
+ case FLOAT:
+ for (int i = 0; i < length; i++) {
+ outputStream.writeLong(batchData.getTimeByIndex(i));
+ outputStream.writeFloat(batchData.getFloatByIndex(i));
+ }
+ break;
+ case TEXT:
+ for (int i = 0; i < length; i++) {
+ outputStream.writeLong(batchData.getTimeByIndex(i));
+ Binary binary = batchData.getBinaryByIndex(i);
+ outputStream.writeInt(binary.getLength());
+ outputStream.write(binary.getValues());
+ }
+ break;
+ case INT64:
+ for (int i = 0; i < length; i++) {
+ outputStream.writeLong(batchData.getTimeByIndex(i));
+ outputStream.writeLong(batchData.getLongByIndex(i));
+ }
+ break;
+ case INT32:
+ for (int i = 0; i < length; i++) {
+ outputStream.writeLong(batchData.getTimeByIndex(i));
+ outputStream.writeInt(batchData.getIntByIndex(i));
+ }
+ break;
+ }
+ } catch (IOException ignored) {
+ // ignored
+ }
+ }
+
+ public static BatchData deserializeBatchData(ByteBuffer buffer) {
+ if (buffer == null || buffer.limit() == 0) {
+ return null;
+ }
+
+ int length = buffer.getInt();
+ TSDataType dataType = TSDataType.values()[buffer.get()];
+ BatchData batchData = new BatchData(dataType);
+ switch (dataType) {
+ case INT32:
+ for (int i = 0; i < length; i++) {
+ batchData.putInt(buffer.getLong(), buffer.getInt());
+ }
+ break;
+ case INT64:
+ for (int i = 0; i < length; i++) {
+ batchData.putLong(buffer.getLong(), buffer.getLong());
+ }
+ break;
+ case TEXT:
+ for (int i = 0; i < length; i++) {
+ long time = buffer.getLong();
+ int len = buffer.getInt();
+ byte[] bytes = new byte[len];
+ buffer.get(bytes);
+ batchData.putBinary(time, new Binary(bytes));
+ }
+ break;
+ case FLOAT:
+ for (int i = 0; i < length; i++) {
+ batchData.putFloat(buffer.getLong(), buffer.getFloat());
+ }
+ break;
+ case DOUBLE:
+ for (int i = 0; i < length; i++) {
+ batchData.putDouble(buffer.getLong(), buffer.getDouble());
+ }
+ break;
+ case BOOLEAN:
+ for (int i = 0; i < length; i++) {
+ batchData.putBoolean(buffer.getLong(), buffer.get() == 1);
+ }
+ break;
+ }
+ return batchData;
+ }
+
+ public static void serializeTVPairs(List<TimeValuePair> timeValuePairs,
+ DataOutputStream dataOutputStream) {
+ try {
+ dataOutputStream.writeInt(timeValuePairs.size());
+ switch (timeValuePairs.get(0).getValue().getDataType()) {
+ case TEXT:
+ for (TimeValuePair timeValuePair : timeValuePairs) {
+ dataOutputStream.writeLong(timeValuePair.getTimestamp());
+ dataOutputStream.writeInt(timeValuePair.getValue().getBinary().getLength());
+ dataOutputStream.write(timeValuePair.getValue().getBinary().getValues());
+ }
+ break;
+ case BOOLEAN:
+ for (TimeValuePair timeValuePair : timeValuePairs) {
+ dataOutputStream.writeLong(timeValuePair.getTimestamp());
+ dataOutputStream.writeBoolean(timeValuePair.getValue().getBoolean());
+ }
+ break;
+ case INT64:
+ for (TimeValuePair timeValuePair : timeValuePairs) {
+ dataOutputStream.writeLong(timeValuePair.getTimestamp());
+ dataOutputStream.writeLong(timeValuePair.getValue().getLong());
+ }
+ break;
+ case INT32:
+ for (TimeValuePair timeValuePair : timeValuePairs) {
+ dataOutputStream.writeLong(timeValuePair.getTimestamp());
+ dataOutputStream.writeInt(timeValuePair.getValue().getInt());
+ }
+ break;
+ case FLOAT:
+ for (TimeValuePair timeValuePair : timeValuePairs) {
+ dataOutputStream.writeLong(timeValuePair.getTimestamp());
+ dataOutputStream.writeFloat(timeValuePair.getValue().getFloat());
+ }
+ break;
+ case DOUBLE:
+ for (TimeValuePair timeValuePair : timeValuePairs) {
+ dataOutputStream.writeLong(timeValuePair.getTimestamp());
+ dataOutputStream.writeDouble(timeValuePair.getValue().getDouble());
+ }
+ }
+ } catch (IOException ignored) {
+ // unreachable
+ }
+ }
+
+ public static void serializeTVPair(TimeValuePair timeValuePair,
+ DataOutputStream dataOutputStream) {
+ if (timeValuePair.getValue() == null) {
+ return;
+ }
+ TSDataType dataType = timeValuePair.getValue().getDataType();
+ try {
+ dataOutputStream.write(dataType.ordinal());
+ switch (dataType) {
+ case TEXT:
+ dataOutputStream.writeLong(timeValuePair.getTimestamp());
+ dataOutputStream.writeInt(timeValuePair.getValue().getBinary().getLength());
+ dataOutputStream.write(timeValuePair.getValue().getBinary().getValues());
+ break;
+ case BOOLEAN:
+ dataOutputStream.writeLong(timeValuePair.getTimestamp());
+ dataOutputStream.writeBoolean(timeValuePair.getValue().getBoolean());
+ break;
+ case INT64:
+ dataOutputStream.writeLong(timeValuePair.getTimestamp());
+ dataOutputStream.writeLong(timeValuePair.getValue().getLong());
+ break;
+ case INT32:
+ dataOutputStream.writeLong(timeValuePair.getTimestamp());
+ dataOutputStream.writeInt(timeValuePair.getValue().getInt());
+ break;
+ case FLOAT:
+ dataOutputStream.writeLong(timeValuePair.getTimestamp());
+ dataOutputStream.writeFloat(timeValuePair.getValue().getFloat());
+ break;
+ case DOUBLE:
+ dataOutputStream.writeLong(timeValuePair.getTimestamp());
+ dataOutputStream.writeDouble(timeValuePair.getValue().getDouble());
+ }
+ } catch (IOException e) {
+ // unreachable
+ }
+ }
+
+ public static List<TimeValuePair> deserializeTVPairs(ByteBuffer buffer) {
+ if (buffer == null || buffer.limit() == 0) {
+ return Collections.emptyList();
+ }
+ TSDataType dataType = TSDataType.values()[buffer.get()];
+ int size = buffer.getInt();
+ List<TimeValuePair> ret = new ArrayList<>(size);
+ switch (dataType) {
+ case DOUBLE:
+ for (int i = 0; i < size; i++) {
+ TimeValuePair pair = new TimeValuePair(buffer.getLong(),
+ TsPrimitiveType.getByType(dataType, buffer.getDouble()));
+ ret.add(pair);
+ }
+ break;
+ case FLOAT:
+ for (int i = 0; i < size; i++) {
+ TimeValuePair pair = new TimeValuePair(buffer.getLong(),
+ TsPrimitiveType.getByType(dataType, buffer.getFloat()));
+ ret.add(pair);
+ }
+ break;
+ case INT32:
+ for (int i = 0; i < size; i++) {
+ TimeValuePair pair = new TimeValuePair(buffer.getLong(),
+ TsPrimitiveType.getByType(dataType, buffer.getInt()));
+ ret.add(pair);
+ }
+ break;
+ case INT64:
+ for (int i = 0; i < size; i++) {
+ TimeValuePair pair = new TimeValuePair(buffer.getLong(),
+ TsPrimitiveType.getByType(dataType, buffer.getLong()));
+ ret.add(pair);
+ }
+ break;
+ case BOOLEAN:
+ for (int i = 0; i < size; i++) {
+ TimeValuePair pair = new TimeValuePair(buffer.getLong(),
+ TsPrimitiveType.getByType(dataType, buffer.get() == 1));
+ ret.add(pair);
+ }
+ break;
+ case TEXT:
+ for (int i = 0; i < size; i++) {
+ long time = buffer.getLong();
+ int bytesLen = buffer.getInt();
+ byte[] bytes = new byte[bytesLen];
+ buffer.get(bytes);
+ TsPrimitiveType primitiveType = TsPrimitiveType.getByType(dataType, bytes);
+ TimeValuePair pair = new TimeValuePair(time, primitiveType);
+ ret.add(pair);
+ }
+ break;
+ }
+ return ret;
+ }
+
+ public static TimeValuePair deserializeTVPair(ByteBuffer buffer)
+ throws UnsupportedDataTypeException {
+ if (buffer == null || (buffer.limit() - buffer.position() == 0)) {
+ return null;
+ }
+ TSDataType dataType = TSDataType.values()[buffer.get()];
+ switch (dataType) {
+ case DOUBLE:
+ return new TimeValuePair(buffer.getLong(),
+ TsPrimitiveType.getByType(dataType, buffer.getDouble()));
+ case FLOAT:
+ return new TimeValuePair(buffer.getLong(),
+ TsPrimitiveType.getByType(dataType, buffer.getFloat()));
+ case INT32:
+ return new TimeValuePair(buffer.getLong(),
+ TsPrimitiveType.getByType(dataType, buffer.getInt()));
+ case INT64:
+ return new TimeValuePair(buffer.getLong(),
+ TsPrimitiveType.getByType(dataType, buffer.getLong()));
+ case BOOLEAN:
+ return new TimeValuePair(buffer.getLong(),
+ TsPrimitiveType.getByType(dataType, buffer.get() == 1));
+ case TEXT:
+ long time = buffer.getLong();
+ int bytesLen = buffer.getInt();
+ byte[] bytes = new byte[bytesLen];
+ buffer.get(bytes);
+ TsPrimitiveType primitiveType = TsPrimitiveType.getByType(dataType, bytes);
+ return new TimeValuePair(time, primitiveType);
+ }
+ throw new UnsupportedDataTypeException(dataType.toString());
+ }
+
+ public static ByteBuffer serializeFilter(Filter filter) {
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
+ filter.serialize(dataOutputStream);
+ return ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
+ }
+
+ public static void serializeObject(Object object, DataOutputStream dataOutputStream) {
+ ReadWriteIOUtils.writeObject(object, dataOutputStream);
+ }
+
+ public static void serializeObjects(Object[] object, DataOutputStream dataOutputStream) {
+ try {
+ dataOutputStream.writeInt(object.length);
+ for (Object o : object) {
+ ReadWriteIOUtils.writeObject(o, dataOutputStream);
+ }
+ } catch (IOException e) {
+ // ignore
+ }
+ }
+
+ public static Object deserializeObject(ByteBuffer buffer) {
+ if (buffer == null || buffer.limit() == 0) {
+ return null;
+ }
+ return ReadWriteIOUtils.readObject(buffer);
+ }
+
+ public static Object[] deserializeObjects(ByteBuffer buffer) {
+ if (buffer == null || buffer.limit() == 0) {
+ return null;
+ }
+ int size = buffer.getInt();
+ Object[] ret = new Object[size];
+ for (int i = 0; i < ret.length; i++) {
+ ret[i] = ReadWriteIOUtils.readObject(buffer);
+ }
+ return ret;
+ }
+
+ public static ByteBuffer serializeLongs(long[] longs) {
+ //TODO-Cluster: replace with a no-copy method
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
+ try {
+ for (long aLong : longs) {
+ dataOutputStream.writeLong(aLong);
+ }
+ } catch (IOException e) {
+ // ignore
+ }
+ return ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
+ }
+
+ public static long[] deserializeLongs(ByteBuffer buffer) {
+ int size = (buffer.limit() - buffer.position()) / Long.BYTES;
+ long[] ret = new long[size];
+ for (int i = 0; i < size; i++) {
+ ret[i] = buffer.getLong();
+ }
+ return ret;
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/UpgradeUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/UpgradeUtils.java
index 62826ac..5d3b37e 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/UpgradeUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/UpgradeUtils.java
@@ -57,7 +57,7 @@ public class UpgradeUtils {
* judge whether a tsfile needs to be upgraded
*/
public static boolean isNeedUpgrade(TsFileResource tsFileResource) {
- tsFileResource.getWriteQueryLock().readLock().lock();
+ tsFileResource.readLock();
//case the TsFile's length is equal to 0, the TsFile does not need to be upgraded
if (tsFileResource.getFile().length() == 0) {
return false;
@@ -71,7 +71,7 @@ public class UpgradeUtils {
logger.error("meet error when judge whether file needs to be upgraded, the file's path:{}",
tsFileResource.getFile().getAbsolutePath(), e);
} finally {
- tsFileResource.getWriteQueryLock().readLock().unlock();
+ tsFileResource.readUnlock();
}
return false;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/manager/MultiFileLogNodeManager.java b/server/src/main/java/org/apache/iotdb/db/writelog/manager/MultiFileLogNodeManager.java
index b904a6e..57126e5 100644
--- a/server/src/main/java/org/apache/iotdb/db/writelog/manager/MultiFileLogNodeManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/manager/MultiFileLogNodeManager.java
@@ -106,7 +106,7 @@ public class MultiFileLogNodeManager implements WriteLogNodeManager, IService {
@Override
public void close() {
if (!isActivated(forceThread)) {
- logger.warn("MultiFileLogNodeManager has not yet started");
+ logger.debug("MultiFileLogNodeManager has not yet started");
return;
}
logger.info("LogNodeManager starts closing..");
@@ -142,7 +142,7 @@ public class MultiFileLogNodeManager implements WriteLogNodeManager, IService {
InstanceHolder.instance.forceThread.start();
}
} else {
- logger.warn("MultiFileLogNodeManager has already started");
+ logger.debug("MultiFileLogNodeManager has already started");
}
} catch (Exception e) {
throw new StartupException(this.getID().getName(), e.getMessage());
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java b/server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java
index dda1a7e..f206395 100644
--- a/server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java
@@ -99,12 +99,12 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
private void putLog(PhysicalPlan plan) {
logBuffer.mark();
try {
- plan.serializeTo(logBuffer);
+ plan.serialize(logBuffer);
} catch (BufferOverflowException e) {
logger.info("WAL BufferOverflow !");
logBuffer.reset();
sync();
- plan.serializeTo(logBuffer);
+ plan.serialize(logBuffer);
}
bufferedLogNum ++;
}
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java
index f627484..8a12d55 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java
@@ -53,6 +53,7 @@ import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
import org.apache.iotdb.tsfile.read.reader.IBatchReader;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.thrift.TException;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -322,7 +323,7 @@ public class TTLTest {
@Test
public void testShowTTL()
throws IOException, QueryProcessException, QueryFilterOptimizationException,
- StorageEngineException, MetadataException {
+ StorageEngineException, MetadataException, TException, InterruptedException {
MManager.getInstance().setTTL(sg1, ttl);
ShowTTLPlan plan = new ShowTTLPlan(Collections.emptyList());
diff --git a/server/src/test/java/org/apache/iotdb/db/metadata/MManagerAdvancedTest.java b/server/src/test/java/org/apache/iotdb/db/metadata/MManagerAdvancedTest.java
index 877580f..85698d7 100644
--- a/server/src/test/java/org/apache/iotdb/db/metadata/MManagerAdvancedTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/metadata/MManagerAdvancedTest.java
@@ -21,11 +21,10 @@ package org.apache.iotdb.db.metadata;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
-import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.apache.iotdb.db.exception.metadata.MetadataException;
-import org.apache.iotdb.db.exception.WriteProcessException;
import org.apache.iotdb.db.metadata.mnode.LeafMNode;
import org.apache.iotdb.db.metadata.mnode.MNode;
import org.apache.iotdb.db.utils.EnvironmentUtils;
@@ -34,6 +33,7 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -45,7 +45,6 @@ public class MManagerAdvancedTest {
@Before
public void setUp() throws Exception {
-
EnvironmentUtils.envSetUp();
mmanager = MManager.getInstance();
@@ -156,4 +155,47 @@ public class MManagerAdvancedTest {
Assert.assertEquals(tv2.getTimestamp(), ((LeafMNode)node).getCachedLast().getTimestamp());
}
+ @Test
+ public void testRemoteCache() throws MetadataException {
+ mmanager.createTimeseries("root.vehicle.d2.s0", TSDataType.DOUBLE, TSEncoding.RLE,
+ TSFileDescriptor.getInstance().getConfig().getCompressor(), Collections.emptyMap());
+ MeasurementSchema schema1 = new MeasurementSchema("root.test1.a.b", TSDataType.INT32);
+ MeasurementSchema schema2= new MeasurementSchema("root.test2.a.c", TSDataType.BOOLEAN);
+ MeasurementSchema schema3 = new MeasurementSchema("root.test1.a.d", TSDataType.TEXT);
+
+ mmanager.cacheSchema("root.test1.a.b", schema1);
+ mmanager.cacheSchema("root.test2.a.c", schema2);
+ mmanager.cacheSchema("root.test1.a.d", schema3);
+ mmanager.cacheSchema("root.vehicle.d2.s0", schema2);
+
+ Assert.assertEquals(mmanager.getSeriesType("root.test1.a.b"), TSDataType.INT32);
+ Assert.assertEquals(mmanager.getSeriesType("root.test2.a.c"), TSDataType.BOOLEAN);
+ Assert.assertEquals(mmanager.getSeriesType("root.test1.a.d"), TSDataType.TEXT);
+ Assert.assertEquals(mmanager.getSeriesType("root.vehicle.d2.s0"), TSDataType.DOUBLE);
+
+ List<String> groups = new ArrayList<>();
+ groups.add("root.test2");
+ try {
+ mmanager.deleteStorageGroups(groups);
+ } catch (MetadataException e) {
+ // ignore
+ }
+ try {
+ mmanager.getSeriesType("root.test2.a.c");
+ fail();
+ } catch (MetadataException e) {
+ // ignore
+ }
+ try {
+ mmanager.deleteTimeseries("root.test1.a.b");
+ } catch (MetadataException e) {
+ // ignore
+ }
+ try {
+ mmanager.getSeriesType("root.test1.a.b");
+ fail();
+ } catch (MetadataException e) {
+ //ignore
+ }
+ }
}
diff --git a/server/src/test/java/org/apache/iotdb/db/qp/plan/PhysicalPlanTest.java b/server/src/test/java/org/apache/iotdb/db/qp/plan/PhysicalPlanTest.java
index 7fe7250..9bf5d7e 100644
--- a/server/src/test/java/org/apache/iotdb/db/qp/plan/PhysicalPlanTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/qp/plan/PhysicalPlanTest.java
@@ -18,6 +18,10 @@
*/
package org.apache.iotdb.db.qp.plan;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
import org.antlr.v4.runtime.misc.ParseCancellationException;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.metadata.MetadataException;
@@ -27,6 +31,7 @@ import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.qp.Planner;
import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+
import org.apache.iotdb.db.qp.physical.crud.*;
import org.apache.iotdb.db.qp.physical.sys.*;
import org.apache.iotdb.db.query.executor.fill.LinearFill;
@@ -63,14 +68,18 @@ public class PhysicalPlanTest {
public void before() throws MetadataException {
MManager.getInstance().init();
MManager.getInstance().setStorageGroup("root.vehicle");
- MManager.getInstance().createTimeseries("root.vehicle.d1.s1", TSDataType.FLOAT, TSEncoding.PLAIN,
- CompressionType.UNCOMPRESSED, null);
- MManager.getInstance().createTimeseries("root.vehicle.d2.s1", TSDataType.FLOAT, TSEncoding.PLAIN,
- CompressionType.UNCOMPRESSED, null);
- MManager.getInstance().createTimeseries("root.vehicle.d3.s1", TSDataType.FLOAT, TSEncoding.PLAIN,
- CompressionType.UNCOMPRESSED, null);
- MManager.getInstance().createTimeseries("root.vehicle.d4.s1", TSDataType.FLOAT, TSEncoding.PLAIN,
- CompressionType.UNCOMPRESSED, null);
+ MManager.getInstance()
+ .createTimeseries("root.vehicle.d1.s1", TSDataType.FLOAT, TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED, null);
+ MManager.getInstance()
+ .createTimeseries("root.vehicle.d2.s1", TSDataType.FLOAT, TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED, null);
+ MManager.getInstance()
+ .createTimeseries("root.vehicle.d3.s1", TSDataType.FLOAT, TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED, null);
+ MManager.getInstance()
+ .createTimeseries("root.vehicle.d4.s1", TSDataType.FLOAT, TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED, null);
}
@After
@@ -84,7 +93,9 @@ public class PhysicalPlanTest {
String metadata = "create timeseries root.vehicle.d1.s2 with datatype=INT32,encoding=RLE";
Planner processor = new Planner();
CreateTimeSeriesPlan plan = (CreateTimeSeriesPlan) processor.parseSQLToPhysicalPlan(metadata);
- assertEquals("seriesPath: root.vehicle.d1.s2, resultDataType: INT32, encoding: RLE, compression: SNAPPY", plan.toString());
+ assertEquals(
+ "seriesPath: root.vehicle.d1.s2, resultDataType: INT32, encoding: RLE, compression: SNAPPY",
+ plan.toString());
}
@Test
@@ -92,7 +103,9 @@ public class PhysicalPlanTest {
String metadata = "create timeseries root.vehicle.d1.s2 with datatype=int32,encoding=rle";
Planner processor = new Planner();
CreateTimeSeriesPlan plan = (CreateTimeSeriesPlan) processor.parseSQLToPhysicalPlan(metadata);
- assertEquals("seriesPath: root.vehicle.d1.s2, resultDataType: INT32, encoding: RLE, compression: SNAPPY", plan.toString());
+ assertEquals(
+ "seriesPath: root.vehicle.d1.s2, resultDataType: INT32, encoding: RLE, compression: SNAPPY",
+ plan.toString());
}
@Test
@@ -796,4 +809,36 @@ public class PhysicalPlanTest {
assertEquals(TSDataType.FLOAT, dt);
}
}
+
+ @Test
+ public void testDelete1() throws QueryProcessException {
+ Path path = new Path("root.vehicle.d1", "s1");
+ List<Path> pathList = new ArrayList<>(Arrays.asList(path));
+ String sqlStr = "delete FROM root.vehicle.d1.s1 WHERE time < 5000";
+ PhysicalPlan plan = processor.parseSQLToPhysicalPlan(sqlStr);
+ assertEquals(OperatorType.DELETE, plan.getOperatorType());
+ assertEquals(pathList, ((DeletePlan) plan).getPaths());
+ }
+
+ @Test
+ public void testDelete2() throws QueryProcessException {
+ Path path1 = new Path("root.vehicle.d1", "s1");
+ Path path2 = new Path("root.vehicle.d1", "s2");
+ List<Path> pathList = new ArrayList<>(Arrays.asList(path1, path2));
+ String sqlStr = "delete FROM root.vehicle.d1.s1,root.vehicle.d1.s2 WHERE time < 5000";
+ PhysicalPlan plan = processor.parseSQLToPhysicalPlan(sqlStr);
+ assertEquals(OperatorType.DELETE, plan.getOperatorType());
+ assertEquals(pathList, ((DeletePlan) plan).getPaths());
+ }
+
+ @Test
+ public void testDelete3() throws QueryProcessException {
+ Path path1 = new Path("root.vehicle.d1", "s1");
+ Path path2 = new Path("root.vehicle.d2", "s3");
+ List<Path> pathList = new ArrayList<>(Arrays.asList(path1, path2));
+ String sqlStr = "delete FROM root.vehicle.d1.s1,root.vehicle.d2.s3 WHERE time < 5000";
+ PhysicalPlan plan = processor.parseSQLToPhysicalPlan(sqlStr);
+ assertEquals(OperatorType.DELETE, plan.getOperatorType());
+ assertEquals(pathList, ((DeletePlan) plan).getPaths());
+ }
}
diff --git a/server/src/test/java/org/apache/iotdb/db/tools/WalCheckerTest.java b/server/src/test/java/org/apache/iotdb/db/tools/WalCheckerTest.java
index 1e7401a..13dc373 100644
--- a/server/src/test/java/org/apache/iotdb/db/tools/WalCheckerTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/tools/WalCheckerTest.java
@@ -78,7 +78,7 @@ public class WalCheckerTest {
String[] measurements = new String[]{"s1", "s2", "s3"};
String[] values = new String[]{"5", "6", "7"};
for (int j = 0; j < 10; j++) {
- new InsertPlan(deviceId, j, measurements, values).serializeTo(binaryPlans);
+ new InsertPlan(deviceId, j, measurements, values).serialize(binaryPlans);
}
binaryPlans.flip();
logWriter.write(binaryPlans);
@@ -111,7 +111,7 @@ public class WalCheckerTest {
String[] measurements = new String[]{"s1", "s2", "s3"};
String[] values = new String[]{"5", "6", "7"};
for (int j = 0; j < 10; j++) {
- new InsertPlan(deviceId, j, measurements, values).serializeTo(binaryPlans);
+ new InsertPlan(deviceId, j, measurements, values).serialize(binaryPlans);
}
if (i > 2) {
binaryPlans.put("not a wal".getBytes());
diff --git a/server/src/test/java/org/apache/iotdb/db/writelog/io/LogWriterReaderTest.java b/server/src/test/java/org/apache/iotdb/db/writelog/io/LogWriterReaderTest.java
index 6dea9ac..e8ef60d 100644
--- a/server/src/test/java/org/apache/iotdb/db/writelog/io/LogWriterReaderTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/writelog/io/LogWriterReaderTest.java
@@ -53,7 +53,7 @@ public class LogWriterReaderTest {
plans.add(insertPlan2);
plans.add(deletePlan);
for (PhysicalPlan plan : plans) {
- plan.serializeTo(logsBuffer);
+ plan.serialize(logsBuffer);
}
}
diff --git a/server/src/test/java/org/apache/iotdb/db/writelog/io/MultiFileLogReaderTest.java b/server/src/test/java/org/apache/iotdb/db/writelog/io/MultiFileLogReaderTest.java
index 3fec657..091a39b 100644
--- a/server/src/test/java/org/apache/iotdb/db/writelog/io/MultiFileLogReaderTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/writelog/io/MultiFileLogReaderTest.java
@@ -51,7 +51,7 @@ public class MultiFileLogReaderTest {
ByteBuffer buffer = ByteBuffer.allocate(64*1024);
for (PhysicalPlan plan : fileLogs[i]) {
- plan.serializeTo(buffer);
+ plan.serialize(buffer);
}
ILogWriter writer = new LogWriter(logFiles[i]);
writer.write(buffer);
diff --git a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java b/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
index 2c6c989..ddb641c 100644
--- a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
+++ b/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
@@ -63,10 +63,18 @@ public enum TSStatusCode {
READ_ONLY_SYSTEM_ERROR(502),
DISK_SPACE_INSUFFICIENT_ERROR(503),
START_UP_ERROR(504),
+
WRONG_LOGIN_PASSWORD_ERROR(600),
NOT_LOGIN_ERROR(601),
NO_PERMISSION_ERROR(602),
UNINITIALIZED_AUTH_ERROR(603),
+
+ PARTITION_NOT_READY(700),
+ TIME_OUT(701),
+ NO_LEADER(702),
+ UNSUPPORTED_OPERATION(703),
+ NODE_READ_ONLY(704),
+
;
private int statusCode;
diff --git a/service-rpc/src/main/thrift/cluster.thrift b/service-rpc/src/main/thrift/cluster.thrift
new file mode 100644
index 0000000..ffa2e94
--- /dev/null
+++ b/service-rpc/src/main/thrift/cluster.thrift
@@ -0,0 +1,386 @@
+/*
+ * 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.
+ */
+
+include "rpc.thrift"
+namespace java org.apache.iotdb.cluster.rpc.thrift
+
+typedef i32 int
+typedef i16 short
+typedef i64 long
+
+// leader -> follower
+struct HeartBeatRequest {
+ 1: required long term // leader's meta log
+ 2: required long commitLogIndex // leader's meta log
+ 3: required long commitLogTerm
+ 4: required Node leader
+ // if the leader does not know the follower's id, and require it reports to the leader, then true
+ 5: required bool requireIdentifier
+ 6: required bool regenerateIdentifier //if the leader finds the follower's id is conflicted,
+ // then true
+ // serialized partitionTable
+ 7: optional binary partitionTableBytes
+
+ // because a data server may play many data groups members, this is used to identify which
+ // member should process the request or response. Only used in data group communication.
+ 8: optional Node header
+}
+
+// follower -> leader
+struct HeartBeatResponse {
+ 1: required long term
+ 2: optional long lastLogIndex // follower's meta log
+ 3: optional long lastLogTerm // follower's meta log
+ // used to perform a catch up when necessary
+ 4: optional Node follower
+ 5: optional int followerIdentifier
+ 6: required bool requirePartitionTable
+
+ // because a data server may play many data groups members, this is used to identify which
+ // member should process the request or response. Only used in data group communication.
+ 7: optional Node header
+}
+
+// node -> node
+struct ElectionRequest {
+ 1: required long term
+ 2: required long lastLogTerm
+ 3: required long lastLogIndex
+ 4: required Node elector
+
+ // because a data server may play many data groups members, this is used to identify which
+ // member should process the request or response. Only used in data group communication.
+ 5: optional Node header
+ 6: optional long dataLogLastIndex
+ 7: optional long dataLogLastTerm
+}
+
+// leader -> follower
+struct AppendEntryRequest {
+ 1: required long term // leader's
+ 2: required Node leader
+ 3: required long prevLogIndex
+ 4: required long prevLogTerm
+ 5: required long leaderCommit
+ 6: required binary entry // data
+
+ // because a data server may play many data groups members, this is used to identify which
+ // member should process the request or response. Only used in data group communication.
+ 7: optional Node header
+}
+
+// leader -> follower
+struct AppendEntriesRequest {
+ 1: required long term // leader's
+ 2: required Node leader
+ 3: required list<binary> entries // data
+ 4: required long prevLogIndex
+ 5: required long prevLogTerm
+ 6: required long leaderCommit
+
+ // because a data server may play many data groups members, this is used to identify which
+ // member should process the request or response. Only used in data group communication.
+ 7: optional Node header
+}
+
+struct AddNodeResponse {
+ // -1: accept to add new node or the node is already in this cluster, otherwise: fail to
+ // add new node
+ 1: required int respNum
+ 2: optional binary partitionTableBytes
+ 3: optional CheckStatusResponse checkStatusResponse
+}
+
+struct Node {
+ 1: required string ip
+ 2: required int metaPort
+ 3: required int nodeIdentifier
+ 4: required int dataPort
+}
+
+// leader -> follower
+struct StartUpStatus {
+ 1: required long partitionInterval
+ 2: required int hashSalt
+ 3: required int replicationNumber
+}
+
+// follower -> leader
+struct CheckStatusResponse {
+ 1: required bool partitionalIntervalEquals
+ 2: required bool hashSaltEquals
+ 3: required bool replicationNumEquals
+}
+
+struct SendSnapshotRequest {
+ 1: required binary snapshotBytes
+ // for data group
+ 2: optional Node header
+}
+
+struct PullSnapshotRequest {
+ 1: required list<int> requiredSlots
+ // for data group
+ 2: optional Node header
+ // set to true if the previous holder has been removed from the cluster.
+ // This will make the previous holder read-only so that different new
+ // replicas can pull the same snapshot.
+ 3: required bool requireReadOnly
+}
+
+struct PullSnapshotResp {
+ 1: optional map<int, binary> snapshotBytes
+}
+
+struct ExecutNonQueryReq {
+ 1: required binary planBytes
+ 2: optional Node header
+}
+
+struct PullSchemaRequest {
+ 1: required list<string> prefixPaths
+ 2: optional Node header
+}
+
+struct PullSchemaResp {
+ 1: required binary schemaBytes
+}
+
+struct SingleSeriesQueryRequest {
+ 1: required string path
+ 2: optional binary timeFilterBytes
+ 3: optional binary valueFilterBytes
+ 4: required long queryId
+ 5: required Node requester
+ 6: required Node header
+ 7: required int dataTypeOrdinal
+ 8: required set<string> deviceMeasurements
+}
+
+struct PreviousFillRequest {
+ 1: required string path
+ 2: required long queryTime
+ 3: required long beforeRange
+ 4: required long queryId
+ 5: required Node requester
+ 6: required Node header
+ 7: required int dataTypeOrdinal
+ 8: required set<string> deviceMeasurements
+}
+
+// the spec and load of a node, for query coordinating
+struct TNodeStatus {
+
+}
+
+struct GetAggrResultRequest {
+ 1: required string path
+ 2: required list<string> aggregations
+ 3: required int dataTypeOrdinal
+ 4: optional binary timeFilterBytes
+ 5: required Node header
+ 6: required long queryId
+ 7: required Node requestor
+ 8: required set<string> deviceMeasurements
+}
+
+struct GroupByRequest {
+ 1: required string path
+ 2: required int dataTypeOrdinal
+ 3: optional binary timeFilterBytes
+ 4: required long queryId
+ 5: required list<int> aggregationTypeOrdinals
+ 6: required Node header
+ 7: required Node requestor
+ 8: required set<string> deviceMeasurements
+}
+
+service RaftService {
+ /**
+ * Leader will call this method to all followers to ensure its authority.
+ * <br>For the receiver,
+ * The method will check the authority of the leader.
+ *
+ * @param request information of the leader
+ * @return if the leader is valid, HeartBeatResponse.term will set -1, and the follower will tell
+ * leader its lastLogIndex; otherwise, the follower will tell the fake leader its term.
+ **/
+ HeartBeatResponse sendHeartbeat(1:HeartBeatRequest request);
+
+ /**
+ * If a node wants to be a leader, it'll call the method to other nodes to get a vote.
+ * <br>For the receiver,
+ * The method will check whether the node can be a leader.
+ *
+ * @param voteRequest a candidate that wants to be a leader.
+ * @return -1 means agree, otherwise return the voter's term
+ **/
+ long startElection(1:ElectionRequest request);
+
+ /**
+ * Leader will call this method to send a batch of entries to all followers.
+ * <br>For the receiver,
+ * The method will check the authority of the leader and if the local log is complete.
+ * If the leader is valid and local log is complete, the follower will append these entries to local log.
+ *
+ * @param request entries that need to be appended and the information of the leader.
+ * @return -1: agree, -2: log index mismatch , otherwise return the follower's term
+ **/
+ long appendEntries(1:AppendEntriesRequest request)
+
+ /**
+ * Leader will call this method to send a entry to all followers.
+ * <br>For the receiver,
+ * The method will check the authority of the leader and if the local log is complete.
+ * If the leader is valid and local log is complete, the follower will append the entry to local log.
+ *
+ * @param request entry that needs to be appended and the information of the leader.
+ * @return -1: agree, -2: log index mismatch , otherwise return the follower's term
+ **/
+ long appendEntry(1:AppendEntryRequest request)
+
+ void sendSnapshot(1:SendSnapshotRequest request)
+
+ /**
+ * Execute a binarized non-query PhysicalPlan
+ **/
+ rpc.TSStatus executeNonQueryPlan(1:ExecutNonQueryReq request)
+
+ /**
+ * Ask the leader for its commit index, used to check whether the node has caught up with the
+ * leader.
+ **/
+ long requestCommitIndex(1:Node header)
+
+ binary readFile(1:string filePath, 2:i64 offset, 3:i32 length)
+}
+
+
+
+service TSDataService extends RaftService {
+
+ /**
+ * Query a time series without value filter.
+ * @return a readerId >= 0 if the query succeeds, otherwise the query fails
+ * TODO-Cluster: support query multiple series in a request
+ **/
+ long querySingleSeries(1:SingleSeriesQueryRequest request)
+
+ /**
+ * Fetch at max fetchSize time-value pairs using the resultSetId generated by querySingleSeries.
+ * @return a ByteBuffer containing the serialized time-value pairs or an empty buffer if there
+ * are not more results.
+ **/
+ binary fetchSingleSeries(1:Node header, 2:long readerId)
+
+ /**
+ * Query a time series and generate an IReaderByTimestamp.
+ * @return a readerId >= 0 if the query succeeds, otherwise the query fails
+ **/
+ long querySingleSeriesByTimestamp(1:SingleSeriesQueryRequest request)
+
+ /**
+ * Fetch one value at given timestamp using the resultSetId generated by
+ * querySingleSeriesByTimestamp.
+ * @return a ByteBuffer containing the serialized value or an empty buffer if there
+ * are not more results.
+ **/
+ binary fetchSingleSeriesByTimestamp(1:Node header, 2:long readerId, 3:long timestamp)
+
+ /**
+ * Find the local query established for the remote query and release all its resource.
+ **/
+ void endQuery(1:Node header, 2:Node thisNode, 3:long queryId)
+
+ /**
+ * Given path patterns (paths with wildcard), return all paths they match.
+ **/
+ list<string> getAllPaths(1:Node header, 2:list<string> path)
+
+ /**
+ * Given path patterns (paths with wildcard), return all devices they match.
+ **/
+ set<string> getAllDevices(1:Node header, 2:list<string> path)
+
+ list<string> getNodeList(1:Node header, 2:string path, 3:int nodeLevel)
+
+ set<string> getChildNodePathInNextLevel(1: Node header, 2: string path)
+
+ binary getAllMeasurementSchema(1: Node header, 2: binary planBinary)
+
+
+ list<binary> getAggrResult(1:GetAggrResultRequest request)
+
+ PullSnapshotResp pullSnapshot(1:PullSnapshotRequest request)
+
+ /**
+ * Create a GroupByExecutor for a path, executing the given aggregations.
+ * @return the executorId
+ **/
+ long getGroupByExecutor(1:GroupByRequest request)
+
+ /**
+ * Fetch the group by result in the interval [startTime, endTime) from the given executor.
+ * @return the serialized AggregationResults, each is the result of one of the previously
+ * required aggregations, and their orders are the same.
+ **/
+ list<binary> getGroupByResult(1:Node header, 2:long executorId, 3:long startTime, 4:long endTime)
+
+
+ /**
+ * Pull all timeseries schemas prefixed by a given path.
+ **/
+ PullSchemaResp pullTimeSeriesSchema(1: PullSchemaRequest request)
+
+ /**
+ * Perform a previous fill and return the timevalue pair in binary.
+ **/
+ binary previousFill(1: PreviousFillRequest request)
+}
+
+service TSMetaService extends RaftService {
+ /**
+ * Node which is not leader will call this method to try to add itself into the cluster as a new node.
+ * <br>For the receiver,
+ * If the local node is leader, it'll check whether the cluster can add this new node;
+ * otherwise, the local node will transfer the request to the leader.
+ *
+ * @param node a new node that needs to be added
+ **/
+ AddNodeResponse addNode(1: Node node, 2: StartUpStatus startUpStatus)
+
+ /**
+ * Remove a node from the cluster. If the node is not in the cluster or the cluster size will
+ * less than replication number, the request will be rejected.
+ * return -1(RESPONSE_AGREE) or -3(RESPONSE_REJECT) or -9(RESPONSE_CLUSTER_TOO_SMALL)
+ **/
+ long removeNode(1: Node node)
+
+ /**
+ * When a node is removed from the cluster, if it is not the meta leader, it cannot receive
+ * the commit command by heartbeat since it has been removed, so the leader should tell it
+ * directly that it is no longer in the cluster.
+ **/
+ void exile()
+
+ TNodeStatus queryNodeStatus()
+
+ Node checkAlive()
+
+}
diff --git a/session/src/main/java/org/apache/iotdb/session/Session.java b/session/src/main/java/org/apache/iotdb/session/Session.java
index 7ef4d62..1f6aebd 100644
--- a/session/src/main/java/org/apache/iotdb/session/Session.java
+++ b/session/src/main/java/org/apache/iotdb/session/Session.java
@@ -55,7 +55,9 @@ import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TCompactProtocol;
+import org.apache.thrift.transport.TFastFramedTransport;
import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -70,7 +72,7 @@ public class Session {
private String password;
private TSIService.Iface client = null;
private long sessionId;
- private TSocket transport;
+ private TTransport transport;
private boolean isClosed = true;
private ZoneId zoneId;
private long statementId;
@@ -101,7 +103,7 @@ public class Session {
}
public synchronized void open() throws IoTDBConnectionException {
- open(false, Config.DEFAULT_TIMEOUT_MS);
+ open(true, Config.DEFAULT_TIMEOUT_MS);
}
public synchronized void open(boolean enableRPCCompression) throws IoTDBConnectionException {
@@ -113,7 +115,7 @@ public class Session {
if (!isClosed) {
return;
}
- transport = new TSocket(host, port, connectionTimeoutInMs);
+ transport = new TFastFramedTransport(new TSocket(host, port, connectionTimeoutInMs));
if (!transport.isOpen()) {
try {
transport.open();
@@ -642,6 +644,7 @@ public class Session {
throws StatementExecutionException, IoTDBConnectionException {
TSExecuteStatementReq execReq = new TSExecuteStatementReq(sessionId, sql, statementId);
+
execReq.setFetchSize(fetchSize);
TSExecuteStatementResp execResp;
try {
diff --git a/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java b/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
index 62e0356..8c262db 100644
--- a/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
+++ b/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
@@ -79,7 +79,7 @@ public class SessionPool {
private boolean enableCompression = false;
public SessionPool(String ip, int port, String user, String password, int maxSize) {
- this(ip, port, user, password, maxSize, Config.DEFAULT_FETCH_SIZE, 60_000, false);
+ this(ip, port, user, password, maxSize, Config.DEFAULT_FETCH_SIZE, 60_000, true);
}
public SessionPool(String ip, int port, String user, String password, int maxSize,
diff --git a/session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java b/session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java
index 00d03ff..050a613 100644
--- a/session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java
+++ b/session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java
@@ -166,7 +166,7 @@ public class SessionPoolTest {
@Test
public void tryIfTheServerIsRestart() {
- SessionPool pool = new SessionPool("127.0.0.1", 6667, "root", "root", 3, 1, 6000, false);
+ SessionPool pool = new SessionPool("127.0.0.1", 6667, "root", "root", 3, 1, 6000, true);
for (int i = 0; i < 10; i++) {
try {
pool.insertRecord("root.sg1.d1", i, Collections.singletonList("s" + i), Collections.singletonList("" + i));
@@ -201,7 +201,7 @@ public class SessionPoolTest {
@Test
public void tryIfTheServerIsRestartButDataIsGotten() {
- SessionPool pool = new SessionPool("127.0.0.1", 6667, "root", "root", 3, 1, 60000, false);
+ SessionPool pool = new SessionPool("127.0.0.1", 6667, "root", "root", 3, 1, 60000, true);
for (int i = 0; i < 10; i++) {
try {
pool.insertRecord("root.sg1.d1", i, Collections.singletonList("s" + i), Collections.singletonList("" + i));
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/common/cache/LRUCache.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/common/cache/LRUCache.java
index e0573a2..ed97dfc 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/common/cache/LRUCache.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/common/cache/LRUCache.java
@@ -27,7 +27,7 @@ import java.util.Map;
*/
public abstract class LRUCache<K, T> implements Cache<K, T> {
- private Map<K, T> cache;
+ protected Map<K, T> cache;
public LRUCache(int cacheSize) {
this.cache = new LinkedHashMap<K, T>(cacheSize, 0.75f, true) {
@@ -59,4 +59,8 @@ public abstract class LRUCache<K, T> implements Cache<K, T> {
}
protected abstract T loadObjectByKey(K key) throws IOException;
+
+ public synchronized void removeItem(K key) {
+ cache.remove(key);
+ }
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TimeValuePair.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TimeValuePair.java
index 035d3a5..c52ee5b 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TimeValuePair.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TimeValuePair.java
@@ -78,5 +78,4 @@ public class TimeValuePair implements Serializable, Comparable<TimeValuePair> {
public int compareTo(TimeValuePair o) {
return Long.compare(this.getTimestamp(), o.getTimestamp());
}
-
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java
index c8dbea3..ce719b7 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java
@@ -327,6 +327,15 @@ public class ReadWriteIOUtils {
}
/**
+ * write byteBuffer.array to outputStream without capacity.
+ */
+ public static int writeWithoutSize(ByteBuffer byteBuffer, OutputStream outputStream) throws IOException {
+ byte[] bytes = byteBuffer.array();
+ outputStream.write(bytes);
+ return bytes.length;
+ }
+
+ /**
* write byteBuffer.capacity and byteBuffer.array to byteBuffer.
*/
public static int write(ByteBuffer byteBuffer, ByteBuffer buffer) {