Skip to content

Parallelize process of set semantic TVF when the conetent of partition keys were same #19155

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Wei-hao-Li wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Parallelize process of set semantic TVF when the conetent of partition keys were same #19155

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 94 additions & 0 deletions integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowTVFIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -251,6 +251,100 @@ public void testSessionFunction() {
DATABASE_NAME);
}

@Test
public void testNestedSessionWithPermutedPartitionColumns() {
String[] expectedHeader =
new String[] {
"window_start",
"window_end",
"inner_start",
"inner_end",
"time",
"factory_id",
"device_id",
"s1"
};
String[] retArray =
new String[] {
"1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.001Z,F1,device_1,10.0,",
"1970-01-01T00:00:00.005Z,1970-01-01T00:00:00.005Z,1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.005Z,F1,device_1,15.0,",
"1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.009Z,F1,device_1,5.0,",
"1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,F1,device_2,20.0,",
"1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,F1,device_2,25.0,",
"1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,F2,device_1,30.0,"
};
// The outer PARTITION BY lists the same columns as the inner one in a different order; the
// result must be identical to the same-order variant (next test).
String sql =
"SELECT window_start, window_end, inner_start, inner_end, time, factory_id, device_id, s1 "
+ "FROM SESSION(DATA => ("
+ " SELECT window_start AS inner_start, window_end AS inner_end, time, factory_id, device_id, s1 "
+ " FROM SESSION(DATA => table5 PARTITION BY (factory_id, device_id) ORDER BY time, GAP => 5ms)"
+ ") PARTITION BY (device_id, factory_id) ORDER BY time, GAP => 3ms) "
+ "ORDER BY factory_id, device_id, time";
tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
}

@Test
public void testNestedSessionWithSamePartitionColumnOrder() {
String[] expectedHeader =
new String[] {
"window_start",
"window_end",
"inner_start",
"inner_end",
"time",
"factory_id",
"device_id",
"s1"
};
String[] retArray =
new String[] {
"1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.001Z,F1,device_1,10.0,",
"1970-01-01T00:00:00.005Z,1970-01-01T00:00:00.005Z,1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.005Z,F1,device_1,15.0,",
"1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.009Z,F1,device_1,5.0,",
"1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,F1,device_2,20.0,",
"1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,F1,device_2,25.0,",
"1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,F2,device_1,30.0,"
};
// Reference case: both levels list the PARTITION BY columns in the same order.
String sql =
"SELECT window_start, window_end, inner_start, inner_end, time, factory_id, device_id, s1 "
+ "FROM SESSION(DATA => ("
+ " SELECT window_start AS inner_start, window_end AS inner_end, time, factory_id, device_id, s1 "
+ " FROM SESSION(DATA => table5 PARTITION BY (factory_id, device_id) ORDER BY time, GAP => 5ms)"
+ ") PARTITION BY (factory_id, device_id) ORDER BY time, GAP => 3ms) "
+ "ORDER BY factory_id, device_id, time";
tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
}

@Test
public void testThreeLevelNestedSessionWithPermutedPartitionColumns() {
String[] expectedHeader =
new String[] {"window_start", "window_end", "time", "factory_id", "device_id", "s1"};
String[] retArray =
new String[] {
"1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.001Z,F1,device_1,10.0,",
"1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.005Z,F1,device_1,15.0,",
"1970-01-01T00:00:00.001Z,1970-01-01T00:00:00.009Z,1970-01-01T00:00:00.009Z,F1,device_1,5.0,",
"1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.002Z,F1,device_2,20.0,",
"1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,1970-01-01T00:00:00.011Z,F1,device_2,25.0,",
"1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.003Z,F2,device_1,30.0,"
};
// Three levels list the same PARTITION BY columns in three different orders.
String sql =
"SELECT window_start, window_end, time, factory_id, device_id, s1 "
+ "FROM SESSION(DATA => ("
+ " SELECT window_start AS mid_start, window_end AS mid_end, inner_start, inner_end, time, factory_id, device_id, s1 "
+ " FROM SESSION(DATA => ("
+ " SELECT window_start AS inner_start, window_end AS inner_end, time, factory_id, device_id, s1 "
+ " FROM SESSION(DATA => table5 PARTITION BY (factory_id, device_id) ORDER BY time, GAP => 5ms)"
+ " ) PARTITION BY (device_id, factory_id) ORDER BY time, GAP => 3ms)"
+ ") PARTITION BY (factory_id, device_id) ORDER BY time, GAP => 5ms) "
+ "ORDER BY factory_id, device_id, time";
tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
}

@Test
public void testVariationFunction() {
String[] expectedHeader = new String[] {"window_index", "time", "stock_id", "price", "s1"};
Expand Down
34 changes: 22 additions & 12 deletions ...pache/iotdb/db/queryengine/plan/relational/planner/optimizations/ParallelizeGrouping.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,8 @@
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExternalTsFileAggregationScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExternalTsFileScanNode;

import com.google.common.collect.ImmutableSet;

import java.util.List;
import java.util.Map;
import java.util.Optional;
Expand Down Expand Up @@ -117,25 +119,33 @@ public PlanNode visitUnion(UnionNode node, Context context) {
.collect(Collectors.toList()));
}

/** We need to make sure: context#partitionKey can match the prefix of childOrderSchema */
/**
* We need to make sure: context#partitionKey can be satisfied by the prefix of childOrder.
*
* <p>The prefix is compared as a column set: partition columns are identified by their values,
* so a different writing order still clusters rows into the same partitions. This lets nested
* set-semantic table functions reuse their input clustering when the same PARTITION BY columns
* are listed in a different order.
*
* <p>The ORDER BY part of the requirement needs no check here. A group node sorts within every
* partition run by its full ordering scheme, so the input only has to be clustered; whatever
* order rows have inside a partition is re-established locally by the group's own sort.
*/
private void checkPrefixMatch(Context context, List<Symbol> childOrder) {
if (context.canSkip()) {
return;
}
if (context.partitionKeyCount > childOrder.size()) {
int partitionKeyCount = context.partitionKeyCount;
if (partitionKeyCount > childOrder.size()) {
context.canOptimized = TO_SORT;
return;
}
OrderingScheme prefix = context.sortKey;
for (int i = 0; i < context.partitionKeyCount; i++) {
Symbol lhs = prefix.getOrderBy().get(i);
Symbol rhs = childOrder.get(i);
if (!lhs.equals(rhs)) {
context.canOptimized = TO_SORT;
return;
}
}
context.canOptimized = ENABLE_PARALLEL;
List<Symbol> requiredPrefix = context.sortKey.getOrderBy().subList(0, partitionKeyCount);
List<Symbol> actualPrefix = childOrder.subList(0, partitionKeyCount);
context.canOptimized =
ImmutableSet.copyOf(requiredPrefix).equals(ImmutableSet.copyOf(actualPrefix))
? ENABLE_PARALLEL
: TO_SORT;
}

@Override
Expand Down
194 changes: 194 additions & 0 deletions ...test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TableFunctionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import org.junit.Test;

import java.util.Collections;
import java.util.List;
import java.util.function.Consumer;

import static org.apache.iotdb.commons.queryengine.plan.relational.function.tvf.ForecastTableFunction.DEFAULT_OUTPUT_INTERVAL;
Expand Down Expand Up @@ -226,6 +227,199 @@ public void testSimpleSetSemantic() {
assertPlan(planTester.getFragmentPlan(5), tableScan);
}

private static final String NESTED_SESSION_SQL =
"SELECT * FROM SESSION(DATA => (SELECT * FROM SESSION(DATA => table1 "
+ "PARTITION BY (tag1, tag2, tag3) ORDER BY time, GAP => 1m)) "
+ "PARTITION BY (tag3, tag2, tag1) ORDER BY time, GAP => 2m)";

private static Consumer<TableFunctionProcessorMatcher.Builder> sessionMatcher(
String windowStart, String windowEnd, long gap) {
return builder ->
builder
.name("session")
.properOutputs(windowStart, windowEnd)
.requiredSymbols("time")
.handle(new MapTableFunctionHandle.Builder().addProperty("GAP", gap).build());
}

private static PlanMatchPattern scanAllColumns() {
return tableScan(
"testdb.table1",
ImmutableList.of("time", "tag1", "tag2", "tag3", "attr1", "attr2", "s1", "s2", "s3"),
ImmutableSet.of("time", "tag1", "tag2", "tag3", "attr1", "attr2", "s1", "s2", "s3"));
}

private static List<PlanMatchPattern.Ordering> orderBy(String... fields) {
ImmutableList.Builder<PlanMatchPattern.Ordering> result = ImmutableList.builder();
for (String field : fields) {
result.add(sort(field, ASCENDING, LAST));
}
return result.build();
}

@Test
public void testNestedSetSemanticPermutedPartitionsKeepBothGroups() {
PlanTester planTester = new PlanTester();
LogicalQueryPlan logicalQueryPlan = planTester.createPlan(NESTED_SESSION_SQL);
Consumer<TableFunctionProcessorMatcher.Builder> outer =
sessionMatcher("window_start", "window_end", 120_000L);
Consumer<TableFunctionProcessorMatcher.Builder> inner =
sessionMatcher("window_start_0", "window_end_1", 60_000L);
// PARTITION BY columns are the same set in a different order and ORDER BY matches, so both
// grouping nodes survive: the outer keeps its declared order, the inner keeps its own.
assertPlan(
logicalQueryPlan,
anyTree(
tableFunctionProcessor(
outer,
group(
orderBy("tag3", "tag2", "tag1", "time"),
3,
tableFunctionProcessor(
inner,
group(orderBy("tag1", "tag2", "tag3", "time"), 3, scanAllColumns()))))));
// Both levels are split per region; only regions with several devices need a merged stream.
assertPlan(planTester.getFragmentPlan(0), output(collect(exchange(), exchange(), exchange())));
assertPlan(
planTester.getFragmentPlan(1),
tableFunctionProcessor(outer, tableFunctionProcessor(inner, scanAllColumns())));
assertPlan(
planTester.getFragmentPlan(2),
tableFunctionProcessor(outer, tableFunctionProcessor(inner, scanAllColumns())));
assertPlan(
planTester.getFragmentPlan(3),
tableFunctionProcessor(
outer,
tableFunctionProcessor(
inner,
mergeSort(orderBy("tag3", "tag2", "tag1", "time"), exchange(), exchange()))));
assertPlan(planTester.getFragmentPlan(4), scanAllColumns());
assertPlan(planTester.getFragmentPlan(5), scanAllColumns());
}

@Test
public void testNestedSetSemanticDifferentOrderByKeepsOuterGroup() {
PlanTester planTester = new PlanTester();
String sql =
"SELECT * FROM SESSION(DATA => (SELECT * FROM SESSION(DATA => table1 "
+ "PARTITION BY (tag1, tag2, tag3) ORDER BY time, GAP => 1m)) "
+ "PARTITION BY (tag3, tag2, tag1) ORDER BY s1, GAP => 2m)";
LogicalQueryPlan logicalQueryPlan = planTester.createPlan(sql);
Consumer<TableFunctionProcessorMatcher.Builder> outer =
sessionMatcher("window_start", "window_end", 120_000L);
Consumer<TableFunctionProcessorMatcher.Builder> inner =
sessionMatcher("window_start_0", "window_end_1", 60_000L);
// Only the partition prefix has to be clustered: the outer group re-establishes its s1 order
// inside every partition, so a different ORDER BY does not degrade it to a full sort.
assertPlan(
logicalQueryPlan,
anyTree(
tableFunctionProcessor(
outer,
group(
orderBy("tag3", "tag2", "tag1", "s1"),
3,
tableFunctionProcessor(
inner,
group(orderBy("tag1", "tag2", "tag3", "time"), 3, scanAllColumns()))))));
// Both levels are still split per region.
assertPlan(planTester.getFragmentPlan(0), output(collect(exchange(), exchange(), exchange())));
}

@Test
public void testNestedSetSemanticGlobalOrderByDeclaredOrder() {
PlanTester planTester = new PlanTester();
LogicalQueryPlan logicalQueryPlan =
planTester.createPlan(NESTED_SESSION_SQL + " ORDER BY tag3, tag2, tag1, time");
Consumer<TableFunctionProcessorMatcher.Builder> outer =
sessionMatcher("window_start", "window_end", 120_000L);
Consumer<TableFunctionProcessorMatcher.Builder> inner =
sessionMatcher("window_start_0", "window_end_1", 60_000L);
// The requested global order equals the outer declared order, so no per-region sort is needed
// and the cross-region merge uses exactly that order.
assertPlan(
planTester.getFragmentPlan(0),
output(
mergeSort(
orderBy("tag3", "tag2", "tag1", "time"), exchange(), exchange(), exchange())));
assertPlan(
planTester.getFragmentPlan(1),
tableFunctionProcessor(outer, tableFunctionProcessor(inner, scanAllColumns())));
assertPlan(
planTester.getFragmentPlan(3),
tableFunctionProcessor(
outer,
tableFunctionProcessor(
inner,
mergeSort(orderBy("tag3", "tag2", "tag1", "time"), exchange(), exchange()))));
}

@Test
public void testNestedSetSemanticGlobalOrderByPermutesPartitionColumns() {
PlanTester planTester = new PlanTester();
// The global order permutes the partition columns again; the distributed planner must prove it
// by rewriting grouping/device comparators instead of reusing the TVF declared order.
LogicalQueryPlan logicalQueryPlan =
planTester.createPlan(NESTED_SESSION_SQL + " ORDER BY tag2, tag3, tag1, time");
Consumer<TableFunctionProcessorMatcher.Builder> outer =
sessionMatcher("window_start", "window_end", 120_000L);
Consumer<TableFunctionProcessorMatcher.Builder> inner =
sessionMatcher("window_start_0", "window_end_1", 60_000L);
assertPlan(
planTester.getFragmentPlan(0),
output(
mergeSort(
orderBy("tag2", "tag3", "tag1", "time"), exchange(), exchange(), exchange())));
assertPlan(
planTester.getFragmentPlan(1),
tableFunctionProcessor(outer, tableFunctionProcessor(inner, scanAllColumns())));
assertPlan(
planTester.getFragmentPlan(3),
tableFunctionProcessor(
outer,
tableFunctionProcessor(
inner,
mergeSort(orderBy("tag2", "tag3", "tag1", "time"), exchange(), exchange()))));
}

@Test
public void testThreeLevelNestedSetSemanticPermutedPartitions() {
PlanTester planTester = new PlanTester();
String sql =
"SELECT * FROM SESSION(DATA => (SELECT * FROM SESSION(DATA => (SELECT * FROM SESSION(DATA => table1 "
+ "PARTITION BY (tag1, tag2, tag3) ORDER BY time, GAP => 1m)) "
+ "PARTITION BY (tag3, tag2, tag1) ORDER BY time, GAP => 2m)) "
+ "PARTITION BY (tag2, tag1, tag3) ORDER BY time, GAP => 3m)";
LogicalQueryPlan logicalQueryPlan = planTester.createPlan(sql);
Consumer<TableFunctionProcessorMatcher.Builder> outerMost =
sessionMatcher("window_start", "window_end", 180_000L);
Consumer<TableFunctionProcessorMatcher.Builder> middle =
sessionMatcher("window_start_0", "window_end_1", 120_000L);
Consumer<TableFunctionProcessorMatcher.Builder> inner =
sessionMatcher("window_start_2", "window_end_3", 60_000L);
// Every level lists the same PARTITION BY columns in a different order, so all three groups
// survive and are executed in parallel per region.
assertPlan(
logicalQueryPlan,
anyTree(
tableFunctionProcessor(
outerMost,
group(
orderBy("tag2", "tag1", "tag3", "time"),
3,
tableFunctionProcessor(
middle,
group(
orderBy("tag3", "tag2", "tag1", "time"),
3,
tableFunctionProcessor(
inner,
group(
orderBy("tag1", "tag2", "tag3", "time"),
3,
scanAllColumns()))))))));
}

@Test
public void testLeafFunction() {
PlanTester planTester = new PlanTester();
Expand Down
Loading
Loading