You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2024/10/15 10:04:22 UTC
(iotdb) 01/01: Fix some sort orders bug about AggTableScan
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch tyAggTest
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit b103d1c208b211cef36c367c5f59975c8208421b
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Tue Oct 15 18:04:04 2024 +0800
Fix some sort orders bug about AggTableScan
---
.../it/query/recent/IoTDBGapFillTableIT.java | 21 +++++
.../plan/relational/planner/QueryPlanner.java | 6 +-
.../plan/relational/planner/SymbolAllocator.java | 7 +-
.../distribute/TableDistributedPlanGenerator.java | 95 +++++++++++++---------
.../iterative/rule/PruneGapFillColumns.java | 13 ++-
.../planner/node/AggregationTableScanNode.java | 4 +-
6 files changed, 99 insertions(+), 47 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBGapFillTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBGapFillTableIT.java
index 522b66e0d93..a59d8e23251 100644
--- a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBGapFillTableIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBGapFillTableIT.java
@@ -96,12 +96,23 @@ public class IoTDBGapFillTableIT {
retArray,
DATABASE_NAME);
+ tableResultSetEqualTest(
+ "select date_bin_gapfill(1h, time) as hour_time, avg(s1) as avg_s1 from table1 where (time >= 2024-09-24T04:00:00.000+00:00 AND time < 2024-09-24T12:00:00.00+00:00) AND device_id = 'd1' GROUP BY 1,city,device_id",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+
// case 2: avg_s1 of one device with having
tableResultSetEqualTest(
"select date_bin_gapfill(1h, time) as hour_time, avg(s1) as avg_s1 from table1 where (time >= 2024-09-24T04:00:00.000+00:00 AND time < 2024-09-24T12:00:00.00+00:00) AND device_id = 'd1' GROUP BY 1 HAVING avg(s1) IS NOT NULL",
expectedHeader,
retArray,
DATABASE_NAME);
+ tableResultSetEqualTest(
+ "select date_bin_gapfill(1h, time) as hour_time, avg(s1) as avg_s1 from table1 where (time >= 2024-09-24T04:00:00.000+00:00 AND time < 2024-09-24T12:00:00.00+00:00) AND device_id = 'd1' GROUP BY 1,city,device_id HAVING avg(s1) IS NOT NULL",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
// case 3: avg_s1 of each device
expectedHeader = new String[] {"hour_time", "city", "device_id", "avg_s1"};
@@ -276,6 +287,11 @@ public class IoTDBGapFillTableIT {
expectedHeader,
retArray,
DATABASE_NAME);
+ tableResultSetEqualTest(
+ "select date_bin_gapfill(1h, time) as hour_time, avg(s1) as avg_s1 from table1 where (time >= 2024-09-24T04:00:00.000+00:00 AND time < 2024-09-24T12:00:00.00+00:00) AND device_id = 'd1' GROUP BY 1,city,device_id FILL METHOD PREVIOUS",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
// case 2: avg_s1 of one device with having
tableResultSetEqualTest(
@@ -283,6 +299,11 @@ public class IoTDBGapFillTableIT {
expectedHeader,
retArray,
DATABASE_NAME);
+ tableResultSetEqualTest(
+ "select date_bin_gapfill(1h, time) as hour_time, avg(s1) as avg_s1 from table1 where (time >= 2024-09-24T04:00:00.000+00:00 AND time < 2024-09-24T12:00:00.00+00:00) AND device_id = 'd1' GROUP BY 1,device_id,city HAVING avg(s1) IS NOT NULL FILL METHOD PREVIOUS",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
// case 3: avg_s1 of each device
expectedHeader = new String[] {"hour_time", "city", "device_id", "avg_s1"};
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/QueryPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/QueryPlanner.java
index 04763c40fa5..f0186d7f062 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/QueryPlanner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/QueryPlanner.java
@@ -85,6 +85,7 @@ import static org.apache.iotdb.db.queryengine.plan.relational.planner.OrderingTr
import static org.apache.iotdb.db.queryengine.plan.relational.planner.PlanBuilder.newPlanBuilder;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.ScopeAware.scopeAwareKey;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.SortOrder.ASC_NULLS_LAST;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.SymbolAllocator.GROUP_KEY_SUFFIX;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.ir.GapFillStartAndEndTimeExtractVisitor.CAN_NOT_INFER_TIME_RANGE;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationNode.groupingSets;
@@ -395,7 +396,7 @@ public class QueryPlanner {
Symbol[] fields = new Symbol[subPlan.getTranslations().getFieldSymbolsList().size()];
for (FieldId field : groupingSetAnalysis.getAllFields()) {
Symbol input = subPlan.getTranslations().getFieldSymbolsList().get(field.getFieldIndex());
- Symbol output = symbolAllocator.newSymbol(input, "gid");
+ Symbol output = symbolAllocator.newSymbol(input, GROUP_KEY_SUFFIX);
fields[field.getFieldIndex()] = output;
groupingSetMappings.put(output, input);
}
@@ -405,7 +406,8 @@ public class QueryPlanner {
if (!complexExpressions.containsKey(
scopeAwareKey(expression, analysis, subPlan.getScope()))) {
Symbol input = subPlan.translate(expression);
- Symbol output = symbolAllocator.newSymbol(expression, analysis.getType(expression), "gid");
+ Symbol output =
+ symbolAllocator.newSymbol(expression, analysis.getType(expression), GROUP_KEY_SUFFIX);
complexExpressions.put(scopeAwareKey(expression, analysis, subPlan.getScope()), output);
groupingSetMappings.put(output, input);
typeProvider.putTableModelType(output, typeProvider.getTableModelType(input));
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/SymbolAllocator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/SymbolAllocator.java
index c52d71ac872..154613acfe4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/SymbolAllocator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/SymbolAllocator.java
@@ -29,6 +29,11 @@ import static com.google.common.base.Preconditions.checkArgument;
import static java.util.Objects.requireNonNull;
public class SymbolAllocator {
+
+ public static final String GROUP_KEY_SUFFIX = "gid";
+
+ public static final String SEPARATOR = "$";
+
private final Map<Symbol, Type> symbolMap;
private int nextId;
@@ -54,7 +59,7 @@ public class SymbolAllocator {
requireNonNull(symbolType, "type is null");
if (suffix != null) {
- symbolHint = symbolHint + "$" + suffix;
+ symbolHint = symbolHint + SEPARATOR + suffix;
}
Symbol symbol = new Symbol(symbolHint);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
index 3e0316b94de..6b43c57b6fe 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
@@ -74,15 +74,20 @@ import java.util.stream.Collectors;
import java.util.stream.IntStream;
import static com.google.common.collect.ImmutableList.toImmutableList;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.SymbolAllocator.GROUP_KEY_SUFFIX;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.SymbolAllocator.SEPARATOR;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationNode.Step.SINGLE;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.PushPredicateIntoTableScan.containsDiffFunction;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.Util.split;
+import static org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.TableBuiltinScalarFunction.DATE_BIN;
import static org.apache.iotdb.db.utils.constant.TestConstant.TIMESTAMP_STR;
import static org.apache.tsfile.utils.Preconditions.checkArgument;
/** This class is used to generate distributed plan for table model. */
public class TableDistributedPlanGenerator
extends PlanVisitor<List<PlanNode>, TableDistributedPlanGenerator.PlanContext> {
+ private static final String PUSH_DOWN_DATE_BIN_SYMBOL_NAME =
+ DATE_BIN.getFunctionName() + SEPARATOR + GROUP_KEY_SUFFIX;
private final QueryId queryId;
private final Analysis analysis;
private final SymbolAllocator symbolAllocator;
@@ -648,13 +653,7 @@ public class TableDistributedPlanGenerator
}
context.mostUsedRegion = mostUsedDataRegion;
- if (node.isStreamable()) {
- context.expectedOrderingScheme = constructOrderingSchema(node.getPreGroupedSymbols());
- processSortProperty(node, resultTableScanNodeList, context);
- } else {
- context.expectedOrderingScheme = null;
- context.hasSortProperty = false;
- }
+ processSortProperty(node, resultTableScanNodeList, context);
if (needSplit) {
if (resultTableScanNodeList.size() == 1) {
@@ -707,12 +706,14 @@ public class TableDistributedPlanGenerator
final OrderingScheme expectedOrderingScheme = context.expectedOrderingScheme;
for (final Symbol symbol : expectedOrderingScheme.getOrderBy()) {
- if (TIMESTAMP_STR.equalsIgnoreCase(symbol.getName())) {
+ if (timeRelatedSymbol(symbol)) {
if (!expectedOrderingScheme.getOrderings().get(symbol).isAscending()) {
// TODO(beyyes) move scan order judgement into logical plan optimizer
resultTableScanNodeList.forEach(
node -> ((TableScanNode) node).setScanOrder(Ordering.DESC));
}
+ newOrderingSymbols.add(symbol);
+ newSortOrders.add(expectedOrderingScheme.getOrdering(symbol));
break;
} else if (!tableScanNode.getIdAndAttributeIndexMap().containsKey(symbol)) {
break;
@@ -729,7 +730,11 @@ public class TableDistributedPlanGenerator
final List<Function<DeviceEntry, String>> orderingRules = new ArrayList<>();
for (final Symbol symbol : newOrderingSymbols) {
- final int idx = tableScanNode.getIdAndAttributeIndexMap().get(symbol);
+ final Integer idx = tableScanNode.getIdAndAttributeIndexMap().get(symbol);
+ if (idx == null) {
+ // time column or date_bin column
+ break;
+ }
if (tableScanNode.getAssignments().get(symbol).getColumnCategory()
== TsTableColumnCategory.ID) {
// segments[0] is always tableName
@@ -738,45 +743,47 @@ public class TableDistributedPlanGenerator
orderingRules.add(deviceEntry -> deviceEntry.getAttributeColumnValues().get(idx));
}
}
+ Comparator<DeviceEntry> comparator = null;
- Comparator<DeviceEntry> comparator;
- if (newSortOrders.get(0).isNullsFirst()) {
- comparator =
- newSortOrders.get(0).isAscending()
- ? Comparator.comparing(
- orderingRules.get(0), Comparator.nullsFirst(Comparator.naturalOrder()))
- : Comparator.comparing(
- orderingRules.get(0), Comparator.nullsFirst(Comparator.naturalOrder()))
- .reversed();
- } else {
- comparator =
- newSortOrders.get(0).isAscending()
- ? Comparator.comparing(
- orderingRules.get(0), Comparator.nullsLast(Comparator.naturalOrder()))
- : Comparator.comparing(
- orderingRules.get(0), Comparator.nullsLast(Comparator.naturalOrder()))
- .reversed();
- }
- for (int i = 1; i < orderingRules.size(); i++) {
- final Comparator<DeviceEntry> thenComparator;
- if (newSortOrders.get(i).isNullsFirst()) {
- thenComparator =
- newSortOrders.get(i).isAscending()
+ if (!orderingRules.isEmpty()) {
+ if (newSortOrders.get(0).isNullsFirst()) {
+ comparator =
+ newSortOrders.get(0).isAscending()
? Comparator.comparing(
- orderingRules.get(i), Comparator.nullsFirst(Comparator.naturalOrder()))
+ orderingRules.get(0), Comparator.nullsFirst(Comparator.naturalOrder()))
: Comparator.comparing(
- orderingRules.get(i), Comparator.nullsFirst(Comparator.naturalOrder()))
+ orderingRules.get(0), Comparator.nullsFirst(Comparator.naturalOrder()))
.reversed();
} else {
- thenComparator =
- newSortOrders.get(i).isAscending()
+ comparator =
+ newSortOrders.get(0).isAscending()
? Comparator.comparing(
- orderingRules.get(i), Comparator.nullsLast(Comparator.naturalOrder()))
+ orderingRules.get(0), Comparator.nullsLast(Comparator.naturalOrder()))
: Comparator.comparing(
- orderingRules.get(i), Comparator.nullsLast(Comparator.naturalOrder()))
+ orderingRules.get(0), Comparator.nullsLast(Comparator.naturalOrder()))
.reversed();
}
- comparator = comparator.thenComparing(thenComparator);
+ for (int i = 1; i < orderingRules.size(); i++) {
+ final Comparator<DeviceEntry> thenComparator;
+ if (newSortOrders.get(i).isNullsFirst()) {
+ thenComparator =
+ newSortOrders.get(i).isAscending()
+ ? Comparator.comparing(
+ orderingRules.get(i), Comparator.nullsFirst(Comparator.naturalOrder()))
+ : Comparator.comparing(
+ orderingRules.get(i), Comparator.nullsFirst(Comparator.naturalOrder()))
+ .reversed();
+ } else {
+ thenComparator =
+ newSortOrders.get(i).isAscending()
+ ? Comparator.comparing(
+ orderingRules.get(i), Comparator.nullsLast(Comparator.naturalOrder()))
+ : Comparator.comparing(
+ orderingRules.get(i), Comparator.nullsLast(Comparator.naturalOrder()))
+ .reversed();
+ }
+ comparator = comparator.thenComparing(thenComparator);
+ }
}
final OrderingScheme newOrderingScheme =
@@ -788,10 +795,18 @@ public class TableDistributedPlanGenerator
for (final PlanNode planNode : resultTableScanNodeList) {
final TableScanNode scanNode = (TableScanNode) planNode;
nodeOrderingMap.put(scanNode.getPlanNodeId(), newOrderingScheme);
- scanNode.getDeviceEntries().sort(comparator);
+ if (comparator != null) {
+ scanNode.getDeviceEntries().sort(comparator);
+ }
}
}
+ // time column or push down date_bin function call in agg which should only have one such column
+ private boolean timeRelatedSymbol(Symbol symbol) {
+ return TIMESTAMP_STR.equalsIgnoreCase(symbol.getName())
+ || PUSH_DOWN_DATE_BIN_SYMBOL_NAME.equals(symbol.getName());
+ }
+
// ------------------- schema related interface ---------------------------------------------
@Override
public List<PlanNode> visitTableDeviceQueryScan(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneGapFillColumns.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneGapFillColumns.java
index ae8052ccd9a..e13c2c15502 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneGapFillColumns.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneGapFillColumns.java
@@ -23,9 +23,13 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.GapFillNode;
+import com.google.common.collect.Streams;
+
import java.util.Optional;
import java.util.Set;
+import java.util.stream.Stream;
+import static com.google.common.collect.ImmutableSet.toImmutableSet;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.Util.restrictChildOutputs;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.node.Patterns.gapFill;
@@ -38,6 +42,13 @@ public class PruneGapFillColumns extends ProjectOffPushDownRule<GapFillNode> {
@Override
protected Optional<PlanNode> pushDownProjectOff(
Context context, GapFillNode gapFillNode, Set<Symbol> referencedOutputs) {
- return restrictChildOutputs(context.getIdAllocator(), gapFillNode, referencedOutputs);
+ Set<Symbol> referencedInputs =
+ Streams.concat(
+ referencedOutputs.stream(),
+ gapFillNode.getGapFillGroupingKeys().stream(),
+ Stream.of(gapFillNode.getGapFillColumn()))
+ .collect(toImmutableSet());
+
+ return restrictChildOutputs(context.getIdAllocator(), gapFillNode, referencedInputs);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTableScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTableScanNode.java
index 0b61595d7e6..f2909ee6bb7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTableScanNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTableScanNode.java
@@ -199,9 +199,7 @@ public class AggregationTableScanNode extends TableScanNode {
}
public boolean isStreamable() {
- return !preGroupedSymbols.isEmpty()
- && groupingSets.getGroupingSetCount() == 1
- && groupingSets.getGlobalGroupingSets().isEmpty();
+ return true;
}
public AggregationNode.Step getStep() {