You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ne...@apache.org on 2022/08/03 02:52:45 UTC
[iotdb] branch master updated: [IOTDB-3280] Can not set ttl to all sg in cli with one sql (#6842)
This is an automated email from the ASF dual-hosted git repository.
neuyilan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new e84b1e3185 [IOTDB-3280] Can not set ttl to all sg in cli with one sql (#6842)
e84b1e3185 is described below
commit e84b1e318590ed51e4f2df9e5e92cb7cacebf171
Author: 23931017wu <71...@users.noreply.github.com>
AuthorDate: Wed Aug 3 10:52:39 2022 +0800
[IOTDB-3280] Can not set ttl to all sg in cli with one sql (#6842)
---
.../consensus/request/write/SetTTLPlan.java | 29 ++-
.../confignode/manager/ClusterSchemaManager.java | 52 ++--
.../persistence/schema/ClusterSchemaInfo.java | 34 ++-
.../thrift/ConfigNodeRPCServiceProcessor.java | 2 +-
.../request/ConfigPhysicalPlanSerDeTest.java | 2 +-
.../thrift/ConfigNodeRPCServiceProcessorTest.java | 6 +-
docs/UserGuide/Operate-Metadata/TTL.md | 21 +-
docs/zh/UserGuide/Operate-Metadata/TTL.md | 18 ++
.../java/org/apache/iotdb/db/it/IoTDBTtlIT.java | 269 +++++++++++++++++++++
.../apache/iotdb/db/integration/IoTDBTtlIT.java | 130 ++++++++++
.../apache/iotdb/db/engine/StorageEngineV2.java | 17 +-
.../config/executor/ClusterConfigTaskExecutor.java | 8 +-
thrift-commons/src/main/thrift/common.thrift | 10 +-
thrift/src/main/thrift/datanode.thrift | 3 +-
14 files changed, 553 insertions(+), 48 deletions(-)
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetTTLPlan.java b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetTTLPlan.java
index e49b3403a7..10c4bc3f1e 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetTTLPlan.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/SetTTLPlan.java
@@ -25,11 +25,13 @@ import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.List;
import java.util.Objects;
public class SetTTLPlan extends ConfigPhysicalPlan {
- private String storageGroup;
+ private String[] storageGroupPathPattern;
private long TTL;
@@ -37,14 +39,14 @@ public class SetTTLPlan extends ConfigPhysicalPlan {
super(ConfigPhysicalPlanType.SetTTL);
}
- public SetTTLPlan(String storageGroup, long TTL) {
+ public SetTTLPlan(List<String> storageGroupPathPattern, long TTL) {
this();
- this.storageGroup = storageGroup;
+ this.storageGroupPathPattern = storageGroupPathPattern.toArray(new String[0]);
this.TTL = TTL;
}
- public String getStorageGroup() {
- return storageGroup;
+ public String[] getStorageGroupPathPattern() {
+ return storageGroupPathPattern;
}
public long getTTL() {
@@ -55,13 +57,21 @@ public class SetTTLPlan extends ConfigPhysicalPlan {
protected void serializeImpl(DataOutputStream stream) throws IOException {
stream.writeInt(getType().ordinal());
- BasicStructureSerDeUtil.write(storageGroup, stream);
+ stream.writeInt(storageGroupPathPattern.length);
+ for (String node : storageGroupPathPattern) {
+ BasicStructureSerDeUtil.write(node, stream);
+ }
stream.writeLong(TTL);
}
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
- storageGroup = BasicStructureSerDeUtil.readString(buffer);
+
+ int length = buffer.getInt();
+ storageGroupPathPattern = new String[length];
+ for (int i = 0; i < length; i++) {
+ storageGroupPathPattern[i] = BasicStructureSerDeUtil.readString(buffer);
+ }
TTL = buffer.getLong();
}
@@ -70,11 +80,12 @@ public class SetTTLPlan extends ConfigPhysicalPlan {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
SetTTLPlan setTTLPlan = (SetTTLPlan) o;
- return TTL == setTTLPlan.TTL && storageGroup.equals(setTTLPlan.storageGroup);
+ return TTL == setTTLPlan.TTL
+ && Arrays.equals(this.storageGroupPathPattern, setTTLPlan.storageGroupPathPattern);
}
@Override
public int hashCode() {
- return Objects.hash(storageGroup, TTL);
+ return Objects.hash(storageGroupPathPattern, TTL);
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterSchemaManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterSchemaManager.java
index 8312b72835..608b9b735b 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterSchemaManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterSchemaManager.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.common.rpc.thrift.TSetTTLReq;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.confignode.client.DataNodeRequestType;
import org.apache.iotdb.confignode.client.async.datanode.AsyncDataNodeClientPool;
import org.apache.iotdb.confignode.client.sync.datanode.SyncDataNodeClientPool;
@@ -153,37 +154,54 @@ public class ClusterSchemaManager {
}
/**
- * Update TTL for the specific StorageGroup
+ * Update TTL for the specific StorageGroup or all storage groups in a path
*
* @param setTTLPlan setTTLPlan
- * @return SUCCESS_STATUS if successfully update the TTL, STORAGE_GROUP_NOT_EXIST if the specific
- * StorageGroup doesn't exist
+ * @return SUCCESS_STATUS if successfully update the TTL, STORAGE_GROUP_NOT_EXIST if the path
+ * doesn't exist
*/
public TSStatus setTTL(SetTTLPlan setTTLPlan) {
- if (!getStorageGroupNames().contains(setTTLPlan.getStorageGroup())) {
+ Map<String, TStorageGroupSchema> storageSchemaMap =
+ clusterSchemaInfo.getMatchedStorageGroupSchemasByOneName(
+ setTTLPlan.getStorageGroupPathPattern());
+
+ if (storageSchemaMap.isEmpty()) {
return RpcUtils.getStatus(
TSStatusCode.STORAGE_GROUP_NOT_EXIST,
- "storageGroup " + setTTLPlan.getStorageGroup() + " does not exist");
+ "Path [" + new PartialPath(setTTLPlan.getStorageGroupPathPattern()) + "] does not exist");
+ }
+ Map<Integer, TDataNodeLocation> dataNodeLocationMaps = new ConcurrentHashMap<>();
+ Map<Integer, List<String>> dnlToSgMap = new ConcurrentHashMap<>();
+ for (String storageGroup : storageSchemaMap.keySet()) {
+
+ Set<TDataNodeLocation> dataNodeLocations =
+ getPartitionManager()
+ .getStorageGroupRelatedDataNodes(storageGroup, TConsensusGroupType.DataRegion);
+ for (TDataNodeLocation dataNodeLocation : dataNodeLocations) {
+ if (!dataNodeLocationMaps.containsKey(dataNodeLocation.getDataNodeId())) {
+ dataNodeLocationMaps.put(dataNodeLocation.getDataNodeId(), dataNodeLocation);
+ List<String> storageGroups = new ArrayList<>();
+ storageGroups.add(storageGroup);
+ dnlToSgMap.put(dataNodeLocation.getDataNodeId(), storageGroups);
+ } else {
+ List<String> storagegroups = dnlToSgMap.get(dataNodeLocation.getDataNodeId());
+ storagegroups.add(storageGroup);
+ dnlToSgMap.put(dataNodeLocation.getDataNodeId(), storagegroups);
+ }
+ }
}
- Map<Integer, TDataNodeLocation> dataNodeLocationMap = new ConcurrentHashMap<>();
- Set<TDataNodeLocation> dataNodeLocations =
- getPartitionManager()
- .getStorageGroupRelatedDataNodes(
- setTTLPlan.getStorageGroup(), TConsensusGroupType.DataRegion);
- dataNodeLocations.forEach(
- dataNodeLocation ->
- dataNodeLocationMap.put(dataNodeLocation.getDataNodeId(), dataNodeLocation));
- if (dataNodeLocations.size() > 0) {
- // TODO: Use procedure to protect SetTTL on DataNodes
+
+ for (Map.Entry<Integer, List<String>> entry : dnlToSgMap.entrySet()) {
+ Map<Integer, TDataNodeLocation> dataNodeLocationMap = new ConcurrentHashMap<>();
+ dataNodeLocationMap.put(entry.getKey(), dataNodeLocationMaps.get(entry.getKey()));
AsyncDataNodeClientPool.getInstance()
.sendAsyncRequestToDataNodeWithRetry(
- new TSetTTLReq(setTTLPlan.getStorageGroup(), setTTLPlan.getTTL()),
+ new TSetTTLReq(entry.getValue(), setTTLPlan.getTTL()),
dataNodeLocationMap,
DataNodeRequestType.SET_TTL,
null);
}
-
return getConsensusManager().write(setTTLPlan).getStatus();
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java
index 3407fd199c..624b2d5bf4 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java
@@ -213,12 +213,15 @@ public class ClusterSchemaInfo implements SnapshotProcessor {
TSStatus result = new TSStatus();
storageGroupReadWriteLock.writeLock().lock();
try {
- PartialPath path = new PartialPath(plan.getStorageGroup());
- if (mTree.isStorageGroupAlreadySet(path)) {
- mTree
- .getStorageGroupNodeByStorageGroupPath(path)
- .getStorageGroupSchema()
- .setTTL(plan.getTTL());
+ PartialPath patternPath = new PartialPath(plan.getStorageGroupPathPattern());
+ List<PartialPath> matchedPaths = mTree.getBelongedStorageGroups(patternPath);
+ if (matchedPaths.size() != 0) {
+ for (PartialPath path : matchedPaths) {
+ mTree
+ .getStorageGroupNodeByStorageGroupPath(path)
+ .getStorageGroupSchema()
+ .setTTL(plan.getTTL());
+ }
result.setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode());
} else {
result.setCode(TSStatusCode.STORAGE_GROUP_NOT_EXIST.getStatusCode());
@@ -691,6 +694,25 @@ public class ClusterSchemaInfo implements SnapshotProcessor {
return new AllTemplateSetInfoResp(outputStream.toByteArray());
}
+ public Map<String, TStorageGroupSchema> getMatchedStorageGroupSchemasByOneName(
+ String[] storageGroupPathPattern) {
+ Map<String, TStorageGroupSchema> schemaMap = new HashMap<>();
+ storageGroupReadWriteLock.readLock().lock();
+ try {
+ PartialPath patternPath = new PartialPath(storageGroupPathPattern);
+ List<PartialPath> matchedPaths = mTree.getBelongedStorageGroups(patternPath);
+ for (PartialPath path : matchedPaths) {
+ schemaMap.put(
+ path.getFullPath(), mTree.getStorageGroupNodeByPath(path).getStorageGroupSchema());
+ }
+ } catch (MetadataException e) {
+ LOGGER.warn("Error StorageGroup name", e);
+ } finally {
+ storageGroupReadWriteLock.readLock().unlock();
+ }
+ return schemaMap;
+ }
+
@TestOnly
public void clear() {
mTree.clear();
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
index e0cfbfbd79..6c5aa24a89 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
@@ -236,7 +236,7 @@ public class ConfigNodeRPCServiceProcessor implements IConfigNodeRPCService.Ifac
@Override
public TSStatus setTTL(TSetTTLReq req) throws TException {
- return configManager.setTTL(new SetTTLPlan(req.getStorageGroup(), req.getTTL()));
+ return configManager.setTTL(new SetTTLPlan(req.getStorageGroupPathPattern(), req.getTTL()));
}
@Override
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java b/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
index 71cf569c19..64a2ec3472 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
@@ -146,7 +146,7 @@ public class ConfigPhysicalPlanSerDeTest {
@Test
public void SetTTLPlanTest() throws IOException {
- SetTTLPlan req0 = new SetTTLPlan("root.sg0", Long.MAX_VALUE);
+ SetTTLPlan req0 = new SetTTLPlan(Arrays.asList("root", "sg0"), Long.MAX_VALUE);
SetTTLPlan req1 = (SetTTLPlan) ConfigPhysicalPlan.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0, req1);
}
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessorTest.java b/confignode/src/test/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessorTest.java
index 88b537c218..1506a6ea00 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessorTest.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessorTest.java
@@ -251,7 +251,7 @@ public class ConfigNodeRPCServiceProcessorTest {
}
@Test
- public void testSetAndQueryStorageGroup() throws TException {
+ public void testSetAndQueryStorageGroup() throws IllegalPathException, TException {
TSStatus status;
final String sg0 = "root.sg0";
final String sg1 = "root.sg1";
@@ -315,7 +315,9 @@ public class ConfigNodeRPCServiceProcessorTest {
TSStatusCode.STORAGE_GROUP_ALREADY_EXISTS.getStatusCode(), status.getCode());
// test StorageGroup setter interfaces
- status = processor.setTTL(new TSetTTLReq(sg1, Long.MAX_VALUE));
+ PartialPath patternPath = new PartialPath(sg1);
+ status =
+ processor.setTTL(new TSetTTLReq(Arrays.asList(patternPath.getNodes()), Long.MAX_VALUE));
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
status = processor.setSchemaReplicationFactor(new TSetSchemaReplicationFactorReq(sg1, 1));
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
diff --git a/docs/UserGuide/Operate-Metadata/TTL.md b/docs/UserGuide/Operate-Metadata/TTL.md
index 5e41403ee4..52d16072f6 100644
--- a/docs/UserGuide/Operate-Metadata/TTL.md
+++ b/docs/UserGuide/Operate-Metadata/TTL.md
@@ -33,6 +33,15 @@ IoTDB> set ttl to root.ln 3600000
This example means that for data in `root.ln`, only that of the latest 1 hour will remain, the older one is removed or made invisible.
+```
+IoTDB> set ttl to root.sgcc.** 3600000
+```
+It supports setting TTL for storage groups in a path. This example represents setting TTL for all storage groups in the `root.sgcc` path.
+```
+IoTDB> set ttl to root.** 3600000
+```
+This example represents setting TTL for all storage groups.
+
## Unset TTL
To unset TTL, we can use follwing SQL statement:
@@ -41,7 +50,17 @@ To unset TTL, we can use follwing SQL statement:
IoTDB> unset ttl to root.ln
```
-After unset TTL, all data will be accepted in `root.ln`
+After unset TTL, all data will be accepted in `root.ln`.
+```
+IoTDB> unset ttl to root.sgcc.**
+```
+
+Unset the TTL setting for all storage groups in the `root.sgcc` path.
+```
+IoTDB> unset ttl to root.**
+```
+
+Unset the TTL setting for all storage groups.
## Show TTL
diff --git a/docs/zh/UserGuide/Operate-Metadata/TTL.md b/docs/zh/UserGuide/Operate-Metadata/TTL.md
index 49888db8a1..4f76e0b153 100644
--- a/docs/zh/UserGuide/Operate-Metadata/TTL.md
+++ b/docs/zh/UserGuide/Operate-Metadata/TTL.md
@@ -32,6 +32,14 @@ IoTDB 支持对存储组级别设置数据存活时间（TTL），这使得 IoTD
IoTDB> set ttl to root.ln 3600000
```
这个例子表示在`root.ln`存储组中，只有最近一个小时的数据将会保存，旧数据会被移除或不可见。
+```
+IoTDB> set ttl to root.sgcc.** 3600000
+```
+支持给某一路径下的存储组设置TTL，这个例子表示`root.sgcc`路径下的所有存储组设置TTL。
+```
+IoTDB> set ttl to root.** 3600000
+```
+表示给所有存储组设置TTL。
## 取消 TTL
@@ -42,6 +50,16 @@ IoTDB> unset ttl to root.ln
```
取消设置 TTL 后，存储组`root.ln`中所有的数据都会被保存。
+```
+IoTDB> unset ttl to root.sgcc.**
+```
+
+取消设置`root.sgcc`路径下的所有存储组的 TTL 。
+```
+IoTDB> unset ttl to root.**
+```
+
+取消设置所有存储组的 TTL 。
## 显示 TTL
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBTtlIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBTtlIT.java
new file mode 100644
index 0000000000..7a82539d13
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBTtlIT.java
@@ -0,0 +1,269 @@
+/*
+ * 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.it;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import static org.junit.Assert.*;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({ClusterIT.class})
+public class IoTDBTtlIT {
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvFactory.getEnv().initBeforeClass();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanAfterClass();
+ }
+
+ @Test
+ public void testTTL() throws SQLException, InterruptedException {
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ try {
+ statement.execute("SET TTL TO root.TTL_SG1 1000");
+ } catch (SQLException e) {
+ assertEquals(322, e.getErrorCode());
+ }
+ try {
+ statement.execute("UNSET TTL TO root.TTL_SG1");
+ } catch (SQLException e) {
+ assertEquals(322, e.getErrorCode());
+ }
+
+ statement.execute("SET STORAGE GROUP TO root.TTL_SG1");
+ statement.execute("CREATE TIMESERIES root.TTL_SG1.s1 WITH DATATYPE=INT64,ENCODING=PLAIN");
+ try {
+ statement.execute("SET TTL TO root.TTL_SG1.s1 1000");
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+
+ statement.execute("SET STORAGE GROUP TO root.TTL_SG2");
+ statement.execute("CREATE TIMESERIES root.TTL_SG2.s1 WITH DATATYPE=INT64,ENCODING=PLAIN");
+ try {
+ statement.execute("SET TTL TO root.TTL_SG2.s1 1000");
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+
+ try {
+ statement.execute("SET TTL TO root.** 1000");
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ try {
+ statement.execute("UNSET TTL TO root.**");
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+
+ long now = System.currentTimeMillis();
+ for (int i = 0; i < 100; i++) {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.TTL_SG1(timestamp, s1) VALUES (%d, %d)", now - 100 + i, i));
+ }
+ for (int i = 0; i < 100; i++) {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.TTL_SG1(timestamp, s1) VALUES (%d, %d)", now - 100000 + i, i));
+ }
+
+ try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.TTL_SG1")) {
+ int cnt = 0;
+ while (resultSet.next()) {
+ cnt++;
+ }
+ assertEquals(200, cnt);
+ }
+
+ statement.execute("SET TTL TO root.TTL_SG1 10000");
+ try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.TTL_SG1")) {
+ int cnt = 0;
+ while (resultSet.next()) {
+ cnt++;
+ }
+ assertEquals(100, cnt);
+ }
+
+ for (int i = 0; i < 100; i++) {
+ boolean caught = false;
+ try {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.TTL_SG1(timestamp, s1) VALUES (%d, %d)", now - 500000 + i, i));
+ } catch (SQLException e) {
+ assertEquals(400, e.getErrorCode());
+ }
+ }
+
+ Thread.sleep(1000);
+ try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.TTL_SG1")) {
+ int cnt = 0;
+ while (resultSet.next()) {
+ cnt++;
+ }
+ assertEquals(100, cnt);
+ }
+
+ statement.execute("UNSET TTL TO root.TTL_SG1");
+ // make sure other nodes have applied UNSET TTL
+ Thread.sleep(1000);
+ for (int i = 0; i < 100; i++) {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.TTL_SG1(timestamp, s1) VALUES (%d, %d)", now - 30000 + i, i));
+ }
+
+ try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.TTL_SG1")) {
+ int cnt = 0;
+ while (resultSet.next()) {
+ cnt++;
+ }
+ assertTrue(cnt >= 200);
+ }
+ statement.execute("SET STORAGE GROUP TO root.sg.TTL_SG3");
+ statement.execute("SET STORAGE GROUP TO root.sg.TTL_SG4");
+ // SG3
+ for (int i = 0; i < 100; i++) {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.sg.TTL_SG3(timestamp, s1) VALUES (%d, %d)", now - 100 + i, i));
+ }
+ for (int i = 100; i < 200; i++) {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.sg.TTL_SG3(timestamp, s1) VALUES (%d, %d)",
+ now - 10000000 + i, i));
+ }
+
+ try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.sg.TTL_SG3")) {
+ int cnt = 0;
+ while (resultSet.next()) {
+ cnt++;
+ }
+ assertEquals(200, cnt);
+ }
+ // SG4
+ for (int i = 200; i < 300; i++) {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.sg.TTL_SG4(timestamp, s1) VALUES (%d, %d)", now - 100 + i, i));
+ }
+ for (int i = 300; i < 400; i++) {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.sg.TTL_SG4(timestamp, s1) VALUES (%d, %d)",
+ now - 10000000 + i, i));
+ }
+
+ try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.sg.TTL_SG4")) {
+ int cnt = 0;
+ while (resultSet.next()) {
+ cnt++;
+ }
+ assertEquals(200, cnt);
+ }
+
+ statement.execute("SET TTL TO root.sg.** 100000");
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM root.sg.**")) {
+ int cnt = 0;
+ while (resultSet.next()) {
+ cnt++;
+ }
+ assertEquals(200, cnt);
+ }
+ for (int i = 0; i < 100; i++) {
+ boolean caught = false;
+ try {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.sg.TTL_SG3(timestamp, s1) VALUES (%d, %d)",
+ now - 5000000 + i, i));
+ } catch (SQLException e) {
+ assertEquals(400, e.getErrorCode());
+ }
+ }
+ for (int i = 100; i < 200; i++) {
+ boolean caught = false;
+ try {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.sg.TTL_SG4(timestamp, s1) VALUES (%d, %d)",
+ now - 5000000 + i, i));
+ } catch (SQLException e) {
+ assertEquals(400, e.getErrorCode());
+ }
+ }
+
+ try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.sg.**")) {
+ int cnt = 0;
+ while (resultSet.next()) {
+ cnt++;
+ }
+ assertEquals(200, cnt);
+ }
+
+ statement.execute("UNSET TTL TO root.sg.**");
+ // make sure other nodes have applied UNSET TTL
+ Thread.sleep(1000);
+ for (int i = 0; i < 100; i++) {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.sg.TTL_SG3(timestamp, s1) VALUES (%d, %d)", now - 30000 + i, i));
+ }
+ for (int i = 100; i < 200; i++) {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.sg.TTL_SG4(timestamp, s1) VALUES (%d, %d)", now - 30000 + i, i));
+ }
+ try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.sg.**")) {
+ int cnt = 0;
+ while (resultSet.next()) {
+ cnt++;
+ }
+ assertTrue(cnt >= 400);
+ }
+ }
+ }
+}
diff --git a/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBTtlIT.java b/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBTtlIT.java
index bbfd40277e..b935dd4081 100644
--- a/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBTtlIT.java
+++ b/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBTtlIT.java
@@ -77,6 +77,31 @@ public class IoTDBTtlIT {
assertEquals(TSStatusCode.STORAGE_GROUP_NOT_EXIST.getStatusCode(), e.getErrorCode());
}
+ statement.execute("SET STORAGE GROUP TO root.TTL_SG2");
+ statement.execute("CREATE TIMESERIES root.TTL_SG2.s1 WITH DATATYPE=INT64,ENCODING=PLAIN");
+ try {
+ statement.execute("SET TTL TO root.TTL_SG2.s1 1000");
+ } catch (SQLException e) {
+ assertEquals(TSStatusCode.STORAGE_GROUP_NOT_EXIST.getStatusCode(), e.getErrorCode());
+ }
+
+ try {
+ statement.execute("SET TTL TO root.** 1000");
+ } catch (SQLException e) {
+ assertEquals(TSStatusCode.TIMESERIES_NOT_EXIST.getStatusCode(), e.getErrorCode());
+ }
+ try {
+ statement.execute("UNSET TTL TO root.**");
+ } catch (SQLException e) {
+ assertEquals(TSStatusCode.TIMESERIES_NOT_EXIST.getStatusCode(), e.getErrorCode());
+ }
+
+ try {
+ statement.execute("SET TTL TO root.**.s1 1000");
+ } catch (SQLException e) {
+ assertEquals(TSStatusCode.STORAGE_GROUP_NOT_EXIST.getStatusCode(), e.getErrorCode());
+ }
+
long now = System.currentTimeMillis();
for (int i = 0; i < 100; i++) {
statement.execute(
@@ -141,6 +166,111 @@ public class IoTDBTtlIT {
}
assertTrue(cnt >= 200);
}
+ statement.execute("SET STORAGE GROUP TO root.sg.TTL_SG3");
+ statement.execute("SET STORAGE GROUP TO root.sg.TTL_SG4");
+ // SG2
+ for (int i = 0; i < 100; i++) {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.sg.TTL_SG3(timestamp, s1) VALUES (%d, %d)", now - 100 + i, i));
+ }
+ for (int i = 100; i < 200; i++) {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.sg.TTL_SG3(timestamp, s1) VALUES (%d, %d)", now - 100000 + i, i));
+ }
+
+ try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.sg.TTL_SG3")) {
+ int cnt = 0;
+ while (resultSet.next()) {
+ cnt++;
+ }
+ assertEquals(200, cnt);
+ }
+ // SG1
+ for (int i = 200; i < 300; i++) {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.sg.TTL_SG4(timestamp, s1) VALUES (%d, %d)", now - 100 + i, i));
+ }
+ for (int i = 300; i < 400; i++) {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.sg.TTL_SG4(timestamp, s1) VALUES (%d, %d)", now - 100000 + i, i));
+ }
+
+ try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.sg.TTL_SG4")) {
+ int cnt = 0;
+ while (resultSet.next()) {
+ cnt++;
+ }
+ assertEquals(200, cnt);
+ }
+
+ statement.execute("SET TTL TO root.sg.** 10000");
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM root.sg.**")) {
+ int cnt = 0;
+ while (resultSet.next()) {
+ cnt++;
+ }
+ assertEquals(200, cnt);
+ }
+ for (int i = 0; i < 100; i++) {
+ boolean caught = false;
+ try {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.sg.TTL_SG3(timestamp, s1) VALUES (%d, %d)",
+ now - 500000 + i, i));
+ } catch (SQLException e) {
+ if (TSStatusCode.OUT_OF_TTL_ERROR.getStatusCode() == e.getErrorCode()) {
+ caught = true;
+ }
+ }
+ assertTrue(caught);
+ }
+ for (int i = 100; i < 200; i++) {
+ boolean caught = false;
+ try {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.sg.TTL_SG4(timestamp, s1) VALUES (%d, %d)",
+ now - 500000 + i, i));
+ } catch (SQLException e) {
+ if (TSStatusCode.OUT_OF_TTL_ERROR.getStatusCode() == e.getErrorCode()) {
+ caught = true;
+ }
+ }
+ assertTrue(caught);
+ }
+ try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.sg.**")) {
+ int cnt = 0;
+ while (resultSet.next()) {
+ cnt++;
+ }
+ assertEquals(200, cnt);
+ }
+
+ statement.execute("UNSET TTL TO root.sg.**");
+ // make sure other nodes have applied UNSET TTL
+ Thread.sleep(1000);
+ for (int i = 0; i < 100; i++) {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.sg.TTL_SG3(timestamp, s1) VALUES (%d, %d)", now - 30000 + i, i));
+ }
+ for (int i = 100; i < 200; i++) {
+ statement.execute(
+ String.format(
+ "INSERT INTO root.sg.TTL_SG4(timestamp, s1) VALUES (%d, %d)", now - 30000 + i, i));
+ }
+ try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.sg.**")) {
+ int cnt = 0;
+ while (resultSet.next()) {
+ cnt++;
+ }
+ assertTrue(cnt >= 400);
+ }
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java
index 72406ea7f9..efb859e1ba 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java
@@ -678,10 +678,25 @@ public class StorageEngineV2 implements IService {
dataRegionMap.put(regionId, newRegion);
}
+ // public TSStatus setTTL(TSetTTLReq req) {
+ // Map<String, List<DataRegionId>> localDataRegionInfo =
+ // StorageEngineV2.getInstance().getLocalDataRegionInfo();
+ // List<DataRegionId> dataRegionIdList = localDataRegionInfo.get(req.storageGroup);
+ // for (DataRegionId dataRegionId : dataRegionIdList) {
+ // DataRegion dataRegion = dataRegionMap.get(dataRegionId);
+ // if (dataRegion != null) {
+ // dataRegion.setDataTTL(req.TTL);
+ // }
+ // }
+ // return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
+ // }
+
public TSStatus setTTL(TSetTTLReq req) {
Map<String, List<DataRegionId>> localDataRegionInfo =
StorageEngineV2.getInstance().getLocalDataRegionInfo();
- List<DataRegionId> dataRegionIdList = localDataRegionInfo.get(req.storageGroup);
+ List<DataRegionId> dataRegionIdList = new ArrayList<>();
+ req.storageGroupPathPattern.forEach(
+ storageGroup -> dataRegionIdList.addAll(localDataRegionInfo.get(storageGroup)));
for (DataRegionId dataRegionId : dataRegionIdList) {
DataRegion dataRegion = dataRegionMap.get(dataRegionId);
if (dataRegion != null) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
index 104241e811..ee22753c99 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -244,9 +244,9 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
@Override
public SettableFuture<ConfigTaskResult> setTTL(SetTTLStatement setTTLStatement, String taskName) {
SettableFuture<ConfigTaskResult> future = SettableFuture.create();
- TSetTTLReq setTTLReq =
- new TSetTTLReq(
- setTTLStatement.getStorageGroupPath().getFullPath(), setTTLStatement.getTTL());
+ List<String> storageGroupPathPattern =
+ Arrays.asList(setTTLStatement.getStorageGroupPath().getNodes());
+ TSetTTLReq setTTLReq = new TSetTTLReq(storageGroupPathPattern, setTTLStatement.getTTL());
try (ConfigNodeClient configNodeClient =
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.partitionRegionId)) {
// Send request to some API server
@@ -258,7 +258,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
taskName,
setTTLStatement.getStorageGroupPath(),
tsStatus);
- future.setException(new StatementExecutionException(tsStatus));
+ future.setException(new IoTDBException(tsStatus.getMessage(), tsStatus.getCode()));
} else {
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
diff --git a/thrift-commons/src/main/thrift/common.thrift b/thrift-commons/src/main/thrift/common.thrift
index fe38f6bd32..5098d9f9ee 100644
--- a/thrift-commons/src/main/thrift/common.thrift
+++ b/thrift-commons/src/main/thrift/common.thrift
@@ -106,13 +106,13 @@ struct TClearCacheReq {
1: optional i32 dataNodeId
}
-struct TSetTTLReq {
- 1: required string storageGroup
- 2: required i64 TTL
-}
-
// for node management
struct TSchemaNode {
1: required string nodeName
2: required byte nodeType
}
+
+struct TSetTTLReq {
+ 1: required list<string> storageGroupPathPattern
+ 2: required i64 TTL
+}
diff --git a/thrift/src/main/thrift/datanode.thrift b/thrift/src/main/thrift/datanode.thrift
index e7941e4e1e..8e62898e25 100644
--- a/thrift/src/main/thrift/datanode.thrift
+++ b/thrift/src/main/thrift/datanode.thrift
@@ -332,9 +332,10 @@ service IDataNodeRPCService {
common.TSStatus flush(common.TFlushReq req)
+ common.TSStatus setTTL(common.TSetTTLReq req)
+
common.TSStatus clearCache(common.TClearCacheReq req)
- common.TSStatus setTTL(common.TSetTTLReq req)
/**
* configNode will notify all DataNodes when the capacity of the ConfigNodeGroup is expanded or reduced
*