Skip to content

Commit 4891c67

Browse files
CaideyipiCaideyipi
committed
fix
1 parent c3ff16a commit 4891c67

File tree

4 files changed

+105
-9
lines changed

4 files changed

+105
-9
lines changed

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -165,8 +165,7 @@ public class IoTDBDataNodeReceiver extends IoTDBFileReceiver {
165165
private final PipeTreeStatementDataTypeConvertExecutionVisitor
166166
treeStatementDataTypeConvertExecutionVisitor =
167167
new PipeTreeStatementDataTypeConvertExecutionVisitor(this::executeStatementForTreeModel);
168-
private final PipeTreeStatementToBatchVisitor batchVisitor =
169-
new PipeTreeStatementToBatchVisitor();
168+
public final PipeTreeStatementToBatchVisitor batchVisitor = new PipeTreeStatementToBatchVisitor();
170169

171170
// Used for data transfer: confignode (cluster A) -> datanode (cluster B) -> confignode (cluster
172171
// B).

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/IoTDBSchemaRegionSource.java‎

Lines changed: 43 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -41,12 +41,15 @@
4141
import org.apache.iotdb.db.pipe.event.common.schema.PipeSchemaRegionWritePlanEvent;
4242
import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics;
4343
import org.apache.iotdb.db.pipe.metric.schema.PipeSchemaRegionSourceMetrics;
44+
import org.apache.iotdb.db.pipe.receiver.visitor.PipeTreeStatementToBatchVisitor;
4445
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
4546
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
4647
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
4748
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.AlterTimeSeriesNode;
4849
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeOperateSchemaQueueNode;
4950
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Node;
51+
import org.apache.iotdb.db.queryengine.plan.statement.Statement;
52+
import org.apache.iotdb.db.queryengine.plan.statement.StatementNode;
5053
import org.apache.iotdb.db.schemaengine.SchemaEngine;
5154
import org.apache.iotdb.db.tools.schema.SRStatementGenerator;
5255
import org.apache.iotdb.db.tools.schema.SchemaRegionSnapshotParser;
@@ -63,6 +66,7 @@
6366
import java.nio.file.Paths;
6467
import java.util.Collections;
6568
import java.util.HashSet;
69+
import java.util.Iterator;
6670
import java.util.Objects;
6771
import java.util.Optional;
6872
import java.util.Set;
@@ -76,16 +80,22 @@ public class IoTDBSchemaRegionSource extends IoTDBNonDataRegionSource {
7680
new PipePlanTablePatternParseVisitor();
7781
public static final PipePlanTablePrivilegeParseVisitor TABLE_PRIVILEGE_PARSE_VISITOR =
7882
new PipePlanTablePrivilegeParseVisitor();
79-
private static final PipeStatementToPlanVisitor STATEMENT_TO_PLAN_VISITOR =
80-
new PipeStatementToPlanVisitor();
83+
private static final PipeTableStatementToPlanVisitor TABLE_STATEMENT_TO_PLAN_VISITOR =
84+
new PipeTableStatementToPlanVisitor();
85+
private static final PipeTreeStatementToPlanVisitor TREE_STATEMENT_TO_PLAN_VISITOR =
86+
new PipeTreeStatementToPlanVisitor();
87+
private final PipeTreeStatementToBatchVisitor batchVisitor =
88+
new PipeTreeStatementToBatchVisitor();
8189

8290
// Local for exception
8391
private PipePlanTreePrivilegeParseVisitor treePrivilegeParseVisitor;
8492
private SchemaRegionId schemaRegionId;
8593

8694
private Set<PlanNodeType> listenedTypeSet = new HashSet<>();
8795
private String database;
96+
private boolean isTableModel;
8897
private SRStatementGenerator generator;
98+
private Iterator<Statement> remainBatches;
8999

90100
@Override
91101
public void customize(
@@ -132,6 +142,7 @@ public void start() throws Exception {
132142
}
133143

134144
database = SchemaEngine.getInstance().getSchemaRegion(schemaRegionId).getDatabaseFullPath();
145+
isTableModel = PathUtils.isTableModelDatabase(database);
135146
super.start();
136147
}
137148

@@ -170,7 +181,7 @@ protected long getMaxBlockingTimeMs() {
170181
@Override
171182
protected boolean canSkipSnapshotPrivilegeCheck(final PipeSnapshotEvent event) {
172183
try {
173-
if (PathUtils.isTableModelDatabase(database)) {
184+
if (isTableModel) {
174185
AuthorityChecker.getAccessControl()
175186
.checkCanSelectFromDatabase4Pipe(userName, database, userEntity);
176187
return true;
@@ -209,14 +220,40 @@ protected void initSnapshotGenerator(final PipeSnapshotEvent event)
209220

210221
@Override
211222
protected boolean hasNextEventInCurrentSnapshot() {
212-
return Objects.nonNull(generator) && generator.hasNext();
223+
return Objects.nonNull(generator) && generator.hasNext()
224+
|| Objects.nonNull(remainBatches) && remainBatches.hasNext();
213225
}
214226

215227
@Override
216228
protected PipeWritePlanEvent getNextEventInCurrentSnapshot() {
217-
// Currently only support table model event
229+
if (isTableModel) {
230+
return new PipeSchemaRegionWritePlanEvent(
231+
TABLE_STATEMENT_TO_PLAN_VISITOR.process((Node) generator.next()), false);
232+
}
233+
while (generator.hasNext()) {
234+
final Optional<Statement> statement =
235+
batchVisitor.process((StatementNode) generator.next(), null);
236+
if (statement.isPresent()) {
237+
if (!generator.hasNext()) {
238+
remainBatches =
239+
batchVisitor.getRemainBatches().stream()
240+
.filter(Optional::isPresent)
241+
.map(Optional::get)
242+
.iterator();
243+
}
244+
return new PipeSchemaRegionWritePlanEvent(
245+
TREE_STATEMENT_TO_PLAN_VISITOR.process(statement.get(), null), false);
246+
}
247+
}
248+
if (Objects.isNull(remainBatches)) {
249+
remainBatches =
250+
batchVisitor.getRemainBatches().stream()
251+
.filter(Optional::isPresent)
252+
.map(Optional::get)
253+
.iterator();
254+
}
218255
return new PipeSchemaRegionWritePlanEvent(
219-
STATEMENT_TO_PLAN_VISITOR.process((Node) generator.next()), false);
256+
TREE_STATEMENT_TO_PLAN_VISITOR.process(remainBatches.next(), null), false);
220257
}
221258

222259
@Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/PipeStatementToPlanVisitor.java renamed to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/PipeTableStatementToPlanVisitor.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@
2626
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateOrUpdateDevice;
2727
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Node;
2828

29-
public class PipeStatementToPlanVisitor extends AstVisitor<PlanNode, Void> {
29+
public class PipeTableStatementToPlanVisitor extends AstVisitor<PlanNode, Void> {
3030
@Override
3131
public PlanNode visitNode(final Node node, final Void context) {
3232
throw new UnsupportedOperationException(

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/schemaregion/PipeTreeStatementToPlanVisitor.java‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
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.db.pipe.source.schemaregion;
21+
22+
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
23+
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
24+
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.BatchActivateTemplateNode;
25+
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.InternalCreateMultiTimeSeriesNode;
26+
import org.apache.iotdb.db.queryengine.plan.statement.StatementNode;
27+
import org.apache.iotdb.db.queryengine.plan.statement.StatementVisitor;
28+
import org.apache.iotdb.db.queryengine.plan.statement.internal.InternalCreateMultiTimeSeriesStatement;
29+
import org.apache.iotdb.db.queryengine.plan.statement.metadata.template.BatchActivateTemplateStatement;
30+
31+
import org.apache.tsfile.utils.Pair;
32+
33+
import java.util.stream.Collectors;
34+
35+
public class PipeTreeStatementToPlanVisitor extends StatementVisitor<PlanNode, Void> {
36+
37+
@Override
38+
public PlanNode visitNode(final StatementNode node, final Void context) {
39+
throw new UnsupportedOperationException(
40+
String.format(
41+
"PipeTreeStatementToPlanVisitor does not support visiting general statement, Statement: %s",
42+
node));
43+
}
44+
45+
@Override
46+
public InternalCreateMultiTimeSeriesNode visitInternalCreateMultiTimeSeries(
47+
final InternalCreateMultiTimeSeriesStatement statement, final Void context) {
48+
// The value pair will not be used at the receiver
49+
return new InternalCreateMultiTimeSeriesNode(new PlanNodeId(""), statement.getDeviceMap());
50+
}
51+
52+
@Override
53+
public BatchActivateTemplateNode visitBatchActivateTemplate(
54+
final BatchActivateTemplateStatement node, final Void context) {
55+
// The value pair will not be used at the receiver
56+
return new BatchActivateTemplateNode(
57+
new PlanNodeId(""),
58+
node.getPaths().stream().collect(Collectors.toMap(k -> k, k -> new Pair<>(0, 0))));
59+
}
60+
}

0 commit comments

Comments
 (0)