Skip to content

Commit 7f5aa4b

Browse files
zerolbsonyzerolbsony
committed
Switch all original old method calls to the method of obtaining configurations at the database granularity
1 parent 1934321 commit 7f5aa4b

57 files changed

Lines changed: 558 additions & 220 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
import org.apache.iotdb.confignode.consensus.request.write.database.SetSchemaReplicationFactorPlan;
4444
import org.apache.iotdb.confignode.consensus.request.write.database.SetTTLPlan;
4545
import org.apache.iotdb.confignode.consensus.request.write.database.SetTimePartitionIntervalPlan;
46+
import org.apache.iotdb.confignode.consensus.request.write.database.SetTimePartitionOriginPlan;
4647
import org.apache.iotdb.confignode.consensus.request.write.datanode.RegisterDataNodePlan;
4748
import org.apache.iotdb.confignode.consensus.request.write.datanode.RemoveDataNodePlan;
4849
import org.apache.iotdb.confignode.consensus.request.write.datanode.UpdateDataNodePlan;
@@ -232,6 +233,9 @@ public static ConfigPhysicalPlan create(final ByteBuffer buffer) throws IOExcept
232233
case SetTimePartitionInterval:
233234
plan = new SetTimePartitionIntervalPlan();
234235
break;
236+
case SetTimePartitionOrigin:
237+
plan = new SetTimePartitionOriginPlan();
238+
break;
235239
case AdjustMaxRegionGroupNum:
236240
plan = new AdjustMaxRegionGroupNumPlan();
237241
break;

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@ public enum ConfigPhysicalPlanType {
4848
SetSchemaReplicationFactor((short) 202),
4949
SetDataReplicationFactor((short) 203),
5050
SetTimePartitionInterval((short) 204),
51+
SetTimePartitionOrigin((short) 212),
5152
AdjustMaxRegionGroupNum((short) 205),
5253
DeleteDatabase((short) 206),
5354
PreDeleteDatabase((short) 207),

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/database/SetTimePartitionOriginPlan.java‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iotdb.confignode.consensus.request.write.database;
21+
22+
import org.apache.iotdb.commons.utils.BasicStructureSerDeUtil;
23+
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
24+
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
25+
26+
import java.io.DataOutputStream;
27+
import java.io.IOException;
28+
import java.nio.ByteBuffer;
29+
import java.util.Objects;
30+
31+
public class SetTimePartitionOriginPlan extends ConfigPhysicalPlan {
32+
33+
private String storageGroup;
34+
35+
private long timePartitionOrigin;
36+
37+
public SetTimePartitionOriginPlan() {
38+
super(ConfigPhysicalPlanType.SetTimePartitionOrigin);
39+
}
40+
41+
public SetTimePartitionOriginPlan(String storageGroup, long timePartitionOrigin) {
42+
this();
43+
this.storageGroup = storageGroup;
44+
this.timePartitionOrigin = timePartitionOrigin;
45+
}
46+
47+
public String getDatabase() {
48+
return storageGroup;
49+
}
50+
51+
public long getTimePartitionOrigin() {
52+
return timePartitionOrigin;
53+
}
54+
55+
@Override
56+
protected void serializeImpl(DataOutputStream stream) throws IOException {
57+
stream.writeShort(getType().getPlanType());
58+
59+
BasicStructureSerDeUtil.write(storageGroup, stream);
60+
stream.writeLong(timePartitionOrigin);
61+
}
62+
63+
@Override
64+
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
65+
storageGroup = BasicStructureSerDeUtil.readString(buffer);
66+
timePartitionOrigin = buffer.getLong();
67+
}
68+
69+
@Override
70+
public boolean equals(Object o) {
71+
if (this == o) {
72+
return true;
73+
}
74+
if (o == null || getClass() != o.getClass()) {
75+
return false;
76+
}
77+
SetTimePartitionOriginPlan that = (SetTimePartitionOriginPlan) o;
78+
return timePartitionOrigin == that.timePartitionOrigin
79+
&& storageGroup.equals(that.storageGroup);
80+
}
81+
82+
@Override
83+
public int hashCode() {
84+
return Objects.hash(storageGroup, timePartitionOrigin);
85+
}
86+
}

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,7 @@
9191
import org.apache.iotdb.confignode.consensus.request.write.database.SetSchemaReplicationFactorPlan;
9292
import org.apache.iotdb.confignode.consensus.request.write.database.SetTTLPlan;
9393
import org.apache.iotdb.confignode.consensus.request.write.database.SetTimePartitionIntervalPlan;
94+
import org.apache.iotdb.confignode.consensus.request.write.database.SetTimePartitionOriginPlan;
9495
import org.apache.iotdb.confignode.consensus.request.write.datanode.RemoveDataNodePlan;
9596
import org.apache.iotdb.confignode.consensus.request.write.template.CreateSchemaTemplatePlan;
9697
import org.apache.iotdb.confignode.consensus.response.ainode.AINodeRegisterResp;
@@ -720,6 +721,16 @@ public TSStatus setTimePartitionInterval(
720721
}
721722
}
722723

724+
@Override
725+
public TSStatus setTimePartitionOrigin(SetTimePartitionOriginPlan setTimePartitionOriginPlan) {
726+
TSStatus status = confirmLeader();
727+
if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
728+
return clusterSchemaManager.setTimePartitionOrigin(setTimePartitionOriginPlan);
729+
} else {
730+
return status;
731+
}
732+
}
733+
723734
@Override
724735
public DataSet countMatchedDatabases(CountDatabasePlan countDatabasePlan) {
725736
TSStatus status = confirmLeader();

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@
5151
import org.apache.iotdb.confignode.consensus.request.write.database.SetSchemaReplicationFactorPlan;
5252
import org.apache.iotdb.confignode.consensus.request.write.database.SetTTLPlan;
5353
import org.apache.iotdb.confignode.consensus.request.write.database.SetTimePartitionIntervalPlan;
54+
import org.apache.iotdb.confignode.consensus.request.write.database.SetTimePartitionOriginPlan;
5455
import org.apache.iotdb.confignode.consensus.request.write.datanode.RemoveDataNodePlan;
5556
import org.apache.iotdb.confignode.manager.consensus.ConsensusManager;
5657
import org.apache.iotdb.confignode.manager.cq.CQManager;
@@ -386,6 +387,8 @@ public interface IManager {
386387

387388
TSStatus setTimePartitionInterval(SetTimePartitionIntervalPlan configPhysicalPlan);
388389

390+
TSStatus setTimePartitionOrigin(SetTimePartitionOriginPlan configPhysicalPlan);
391+
389392
/**
390393
* Count Databases.
391394
*

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@
6464
import org.apache.iotdb.confignode.consensus.request.write.database.SetDataReplicationFactorPlan;
6565
import org.apache.iotdb.confignode.consensus.request.write.database.SetSchemaReplicationFactorPlan;
6666
import org.apache.iotdb.confignode.consensus.request.write.database.SetTimePartitionIntervalPlan;
67+
import org.apache.iotdb.confignode.consensus.request.write.database.SetTimePartitionOriginPlan;
6768
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeEnrichedPlan;
6869
import org.apache.iotdb.confignode.consensus.request.write.table.SetTableColumnCommentPlan;
6970
import org.apache.iotdb.confignode.consensus.request.write.table.SetTableCommentPlan;
@@ -470,6 +471,17 @@ public TSStatus setTimePartitionInterval(
470471
}
471472
}
472473

474+
public TSStatus setTimePartitionOrigin(SetTimePartitionOriginPlan setTimePartitionOriginPlan) {
475+
try {
476+
return getConsensusManager().write(setTimePartitionOriginPlan);
477+
} catch (ConsensusException e) {
478+
LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
479+
TSStatus result = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
480+
result.setMessage(e.getMessage());
481+
return result;
482+
}
483+
}
484+
473485
/**
474486
* Only leader use this interface. Adjust the maxSchemaRegionGroupNum and maxDataRegionGroupNum of
475487
* each Database based on existing cluster resources

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,7 @@
7676
import org.apache.iotdb.confignode.consensus.request.write.database.SetSchemaReplicationFactorPlan;
7777
import org.apache.iotdb.confignode.consensus.request.write.database.SetTTLPlan;
7878
import org.apache.iotdb.confignode.consensus.request.write.database.SetTimePartitionIntervalPlan;
79+
import org.apache.iotdb.confignode.consensus.request.write.database.SetTimePartitionOriginPlan;
7980
import org.apache.iotdb.confignode.consensus.request.write.datanode.RegisterDataNodePlan;
8081
import org.apache.iotdb.confignode.consensus.request.write.datanode.RemoveDataNodePlan;
8182
import org.apache.iotdb.confignode.consensus.request.write.datanode.UpdateDataNodePlan;
@@ -435,6 +436,8 @@ public TSStatus executeNonQueryPlan(ConfigPhysicalPlan physicalPlan)
435436
case SetTimePartitionInterval:
436437
return clusterSchemaInfo.setTimePartitionInterval(
437438
(SetTimePartitionIntervalPlan) physicalPlan);
439+
case SetTimePartitionOrigin:
440+
return clusterSchemaInfo.setTimePartitionOrigin((SetTimePartitionOriginPlan) physicalPlan);
438441
case CreateRegionGroups:
439442
return partitionInfo.createRegionGroups((CreateRegionGroupsPlan) physicalPlan);
440443
case OfferRegionMaintainTasks:

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -620,7 +620,7 @@ public Map<TSeriesPartitionSlot, TConsensusGroupId> getLastDataAllotTable() {
620620
*/
621621
public void autoCleanPartitionTable(long TTL, TTimePartitionSlot currentTimeSlot) {
622622
long[] removedTimePartitionSlots =
623-
dataPartitionTable.autoCleanPartitionTable(TTL, currentTimeSlot).stream()
623+
dataPartitionTable.autoCleanPartitionTable(TTL, currentTimeSlot, databaseName).stream()
624624
.map(TTimePartitionSlot::getStartTime)
625625
.collect(Collectors.toList())
626626
.stream()

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -521,7 +521,7 @@ public TSStatus autoCleanPartitionTable(AutoCleanPartitionTablePlan plan) {
521521
if (isDatabaseExisted(database) && 0 < ttl && ttl < Long.MAX_VALUE) {
522522
databasePartitionTables
523523
.get(database)
524-
.autoCleanPartitionTable(ttl, plan.getCurrentTimeSlot());
524+
.autoCleanPartitionTable(ttl, plan.getCurrentTimeSlotMap());
525525
}
526526
});
527527
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,7 @@
5454
import org.apache.iotdb.confignode.consensus.request.write.database.SetDataReplicationFactorPlan;
5555
import org.apache.iotdb.confignode.consensus.request.write.database.SetSchemaReplicationFactorPlan;
5656
import org.apache.iotdb.confignode.consensus.request.write.database.SetTimePartitionIntervalPlan;
57+
import org.apache.iotdb.confignode.consensus.request.write.database.SetTimePartitionOriginPlan;
5758
import org.apache.iotdb.confignode.consensus.request.write.table.AddTableColumnPlan;
5859
import org.apache.iotdb.confignode.consensus.request.write.table.AlterColumnDataTypePlan;
5960
import org.apache.iotdb.confignode.consensus.request.write.table.CommitCreateTablePlan;
@@ -493,6 +494,32 @@ public TSStatus setTimePartitionInterval(final SetTimePartitionIntervalPlan plan
493494
return result;
494495
}
495496

497+
public TSStatus setTimePartitionOrigin(final SetTimePartitionOriginPlan plan) {
498+
final TSStatus result = new TSStatus();
499+
databaseReadWriteLock.writeLock().lock();
500+
try {
501+
final ConfigMTree mTree =
502+
PathUtils.isTableModelDatabase(plan.getDatabase()) ? tableModelMTree : treeModelMTree;
503+
final PartialPath path = getQualifiedDatabasePartialPath(plan.getDatabase());
504+
if (mTree.isDatabaseAlreadySet(path)) {
505+
mTree
506+
.getDatabaseNodeByDatabasePath(path)
507+
.getAsMNode()
508+
.getDatabaseSchema()
509+
.setTimePartitionOrigin(plan.getTimePartitionOrigin());
510+
result.setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode());
511+
} else {
512+
result.setCode(TSStatusCode.DATABASE_NOT_EXIST.getStatusCode());
513+
}
514+
} catch (final MetadataException e) {
515+
LOGGER.error(ERROR_NAME, e);
516+
result.setCode(TSStatusCode.DATABASE_NOT_EXIST.getStatusCode()).setMessage(ERROR_NAME);
517+
} finally {
518+
databaseReadWriteLock.writeLock().unlock();
519+
}
520+
return result;
521+
}
522+
496523
/**
497524
* Adjust the maximum RegionGroup count of each Database.
498525
*

0 commit comments

Comments
 (0)