Skip to content

Commit d328639

Browse files
shuwenweishuwenwei
authored
Move query state machine support to calc commons (#17912)
1 parent e7bc492 commit d328639

6 files changed

Lines changed: 113 additions & 30 deletions

File tree

‎iotdb-core/calc-commons/src/main/i18n/en/org/apache/iotdb/calc/i18n/CalcMessages.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -151,4 +151,14 @@ private CalcMessages() {}
151151
"pushed element is null";
152152

153153
public static final String FAILED_TO_DELETE_TEMP_DIR = "Failed to delete temp dir {}.";
154+
155+
// --- Execution ---
156+
157+
public static final String ERROR_SETTING_FUTURE_STATE_FOR =
158+
"Error setting future state for {}";
159+
public static final String ERROR_NOTIFYING_STATE_CHANGE_LISTENER_FOR =
160+
"Error notifying state change listener for {}";
161+
public static final String SERVER_IS_SHUTTING_DOWN =
162+
"Server is shutting down";
163+
154164
}

‎iotdb-core/calc-commons/src/main/i18n/zh/org/apache/iotdb/calc/i18n/CalcMessages.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -143,4 +143,13 @@ private CalcMessages() {}
143143
"推入的元素为 null";
144144

145145
public static final String FAILED_TO_DELETE_TEMP_DIR = "删除临时目录 {} 失败。";
146+
147+
// --- Execution ---
148+
149+
public static final String ERROR_SETTING_FUTURE_STATE_FOR =
150+
"为 {} 设置 future 状态时出错";
151+
public static final String ERROR_NOTIFYING_STATE_CHANGE_LISTENER_FOR =
152+
"通知 {} 的状态变更监听器时出错";
153+
public static final String SERVER_IS_SHUTTING_DOWN =
154+
"服务器正在关闭";
146155
}

‎iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/plan/planner/TableOperatorGenerator.java‎

Lines changed: 90 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -422,7 +422,7 @@ public Operator visitProject(ProjectNode node, C context) {
422422
context);
423423
}
424424

425-
private List<TSDataType> getInputColumnTypes(PlanNode node, ITableTypeProvider typeProvider) {
425+
protected List<TSDataType> getInputColumnTypes(PlanNode node, ITableTypeProvider typeProvider) {
426426
// ignore "time" column
427427
return node.getChildren().stream()
428428
.map(PlanNode::getOutputSymbols)
@@ -1304,7 +1304,7 @@ public Operator visitAggregation(AggregationNode node, C context) {
13041304
return planGroupByAggregation(node, child, context.getTableTypeProvider(), context);
13051305
}
13061306

1307-
private Operator planGlobalAggregation(
1307+
protected AggregationOperator planGlobalAggregation(
13081308
AggregationNode node, Operator child, ITableTypeProvider typeProvider, C context) {
13091309
CommonOperatorContext operatorContext =
13101310
addOperatorContext(
@@ -1328,7 +1328,7 @@ private Operator planGlobalAggregation(
13281328
false,
13291329
null,
13301330
Collections.emptySet())));
1331-
return new AggregationOperator(operatorContext, child, aggregatorBuilder.build());
1331+
return createAggregationOperator(operatorContext, child, aggregatorBuilder.build());
13321332
}
13331333

13341334
// timeColumnName and measurementColumnNames will only be set for AggTableScan.
@@ -1411,7 +1411,7 @@ protected Operator planGroupByAggregation(
14111411
CommonOperatorContext operatorContext =
14121412
addOperatorContext(
14131413
context, node.getPlanNodeId(), StreamingAggregationOperator.class.getSimpleName());
1414-
return new StreamingAggregationOperator(
1414+
return createStreamingAggregationOperator(
14151415
operatorContext,
14161416
child,
14171417
groupByTypes,
@@ -1457,7 +1457,7 @@ protected Operator planGroupByAggregation(
14571457
context,
14581458
node.getPlanNodeId(),
14591459
StreamingHashAggregationOperator.class.getSimpleName());
1460-
return new StreamingHashAggregationOperator(
1460+
return createStreamingHashAggregationOperator(
14611461
operatorContext,
14621462
child,
14631463
preGroupedChannels,
@@ -1484,7 +1484,7 @@ protected Operator planGroupByAggregation(
14841484
addOperatorContext(
14851485
context, node.getPlanNodeId(), HashAggregationOperator.class.getSimpleName());
14861486

1487-
return new HashAggregationOperator(
1487+
return createHashAggregationOperator(
14881488
operatorContext,
14891489
child,
14901490
groupByTypes,
@@ -1497,6 +1497,89 @@ protected Operator planGroupByAggregation(
14971497
Long.MAX_VALUE);
14981498
}
14991499

1500+
protected AggregationOperator createAggregationOperator(
1501+
CommonOperatorContext operatorContext, Operator child, List<TableAggregator> aggregators) {
1502+
return new AggregationOperator(operatorContext, child, aggregators);
1503+
}
1504+
1505+
protected StreamingAggregationOperator createStreamingAggregationOperator(
1506+
CommonOperatorContext operatorContext,
1507+
Operator child,
1508+
List<Type> groupByTypes,
1509+
List<Integer> groupByChannels,
1510+
Comparator<SortKey> groupKeyComparator,
1511+
List<TableAggregator> aggregators,
1512+
long maxPartialMemory,
1513+
boolean spillEnabled,
1514+
long unSpillMemoryLimit) {
1515+
return new StreamingAggregationOperator(
1516+
operatorContext,
1517+
child,
1518+
groupByTypes,
1519+
groupByChannels,
1520+
groupKeyComparator,
1521+
aggregators,
1522+
maxPartialMemory,
1523+
spillEnabled,
1524+
unSpillMemoryLimit);
1525+
}
1526+
1527+
protected StreamingHashAggregationOperator createStreamingHashAggregationOperator(
1528+
CommonOperatorContext operatorContext,
1529+
Operator child,
1530+
List<Integer> preGroupedChannels,
1531+
List<Integer> preGroupedIndexInResult,
1532+
List<Type> unPreGroupedTypes,
1533+
List<Integer> unPreGroupedChannels,
1534+
List<Integer> unPreGroupedIndexInResult,
1535+
Comparator<SortKey> groupKeyComparator,
1536+
List<GroupedAggregator> aggregators,
1537+
AggregationNode.Step step,
1538+
int expectedGroups,
1539+
long maxPartialMemory,
1540+
boolean spillEnabled,
1541+
long unSpillMemoryLimit) {
1542+
return new StreamingHashAggregationOperator(
1543+
operatorContext,
1544+
child,
1545+
preGroupedChannels,
1546+
preGroupedIndexInResult,
1547+
unPreGroupedTypes,
1548+
unPreGroupedChannels,
1549+
unPreGroupedIndexInResult,
1550+
groupKeyComparator,
1551+
aggregators,
1552+
step,
1553+
expectedGroups,
1554+
maxPartialMemory,
1555+
spillEnabled,
1556+
unSpillMemoryLimit);
1557+
}
1558+
1559+
protected HashAggregationOperator createHashAggregationOperator(
1560+
CommonOperatorContext operatorContext,
1561+
Operator child,
1562+
List<Type> groupByTypes,
1563+
List<Integer> groupByChannels,
1564+
List<GroupedAggregator> aggregators,
1565+
AggregationNode.Step step,
1566+
int expectedGroups,
1567+
long maxPartialMemory,
1568+
boolean spillEnabled,
1569+
long unSpillMemoryLimit) {
1570+
return new HashAggregationOperator(
1571+
operatorContext,
1572+
child,
1573+
groupByTypes,
1574+
groupByChannels,
1575+
aggregators,
1576+
step,
1577+
expectedGroups,
1578+
maxPartialMemory,
1579+
spillEnabled,
1580+
unSpillMemoryLimit);
1581+
}
1582+
15001583
protected Comparator<SortKey> genGroupKeyComparator(
15011584
List<Type> groupTypes, List<Integer> groupByChannels) {
15021585
return getComparatorForTable(
@@ -2248,8 +2331,7 @@ public Operator visitValuesNode(ValuesNode node, C context) {
22482331
public Operator visitRowNumber(RowNumberNode node, C context) {
22492332
Operator child = node.getChild().accept(this, context);
22502333
CommonOperatorContext operatorContext =
2251-
addOperatorContext(
2252-
context, node.getPlanNodeId(), MappingCollectOperator.class.getSimpleName());
2334+
addOperatorContext(context, node.getPlanNodeId(), RowNumberOperator.class.getSimpleName());
22532335

22542336
List<Symbol> partitionBySymbols = node.getPartitionBy();
22552337
Map<Symbol, Integer> childLayout =

‎iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -30,15 +30,6 @@ public final class DataNodeQueryMessages {
3030
public static final String THIS_NODE_ISN_T_INSTANCE_OF_SCHEMAMEASUREMENTNODE =
3131
"This node isn't instance of SchemaMeasurementNode.";
3232

33-
// --- Execution ---
34-
35-
public static final String ERROR_SETTING_FUTURE_STATE_FOR =
36-
"Error setting future state for {}";
37-
public static final String ERROR_NOTIFYING_STATE_CHANGE_LISTENER_FOR =
38-
"Error notifying state change listener for {}";
39-
public static final String SERVER_IS_SHUTTING_DOWN =
40-
"Server is shutting down";
41-
4233
// --- Execution / Aggregation ---
4334

4435
public static final String INVALID_AGGREGATION_FUNCTION =

‎iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -30,15 +30,6 @@ public final class DataNodeQueryMessages {
3030
public static final String THIS_NODE_ISN_T_INSTANCE_OF_SCHEMAMEASUREMENTNODE =
3131
"该节点不是 SchemaMeasurementNode 实例。";
3232

33-
// --- Execution ---
34-
35-
public static final String ERROR_SETTING_FUTURE_STATE_FOR =
36-
"为 {} 设置 future 状态时出错";
37-
public static final String ERROR_NOTIFYING_STATE_CHANGE_LISTENER_FOR =
38-
"通知 {} 的状态变更监听器时出错";
39-
public static final String SERVER_IS_SHUTTING_DOWN =
40-
"服务器正在关闭";
41-
4233
// --- Execution / Aggregation ---
4334

4435
public static final String INVALID_AGGREGATION_FUNCTION =

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/StateMachine.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@
1919

2020
package org.apache.iotdb.db.queryengine.execution;
2121

22-
import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
22+
import org.apache.iotdb.calc.i18n.CalcMessages;
2323

2424
import com.google.common.annotations.VisibleForTesting;
2525
import com.google.common.collect.ImmutableList;
@@ -242,7 +242,7 @@ private void fireStateChanged(
242242
try {
243243
futureStateChange.complete(newState);
244244
} catch (Throwable e) {
245-
LOGGER.error(DataNodeQueryMessages.ERROR_SETTING_FUTURE_STATE_FOR, name, e);
245+
LOGGER.error(CalcMessages.ERROR_SETTING_FUTURE_STATE_FOR, name, e);
246246
}
247247
for (StateChangeListener<T> stateChangeListener : stateChangeListeners) {
248248
fireStateChangedListener(newState, stateChangeListener);
@@ -255,7 +255,7 @@ private void fireStateChangedListener(T newState, StateChangeListener<T> stateCh
255255
try {
256256
stateChangeListener.stateChanged(newState);
257257
} catch (Throwable e) {
258-
LOGGER.error(DataNodeQueryMessages.ERROR_NOTIFYING_STATE_CHANGE_LISTENER_FOR, name, e);
258+
LOGGER.error(CalcMessages.ERROR_NOTIFYING_STATE_CHANGE_LISTENER_FOR, name, e);
259259
}
260260
}
261261

@@ -327,7 +327,7 @@ private void safeExecute(Runnable command) {
327327
executor.execute(command);
328328
} catch (RejectedExecutionException e) {
329329
if ((executor instanceof ExecutorService) && ((ExecutorService) executor).isShutdown()) {
330-
throw new RuntimeException(DataNodeQueryMessages.SERVER_IS_SHUTTING_DOWN, e);
330+
throw new RuntimeException(CalcMessages.SERVER_IS_SHUTTING_DOWN, e);
331331
}
332332
throw e;
333333
}

0 commit comments

Comments
 (0)