You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ca...@apache.org on 2024/07/30 08:40:04 UTC
(iotdb) branch fix_cnf created (now f6b66af2f83)
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a change to branch fix_cnf
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at f6b66af2f83 fix possible cnf problem
This branch includes the following new commits:
new f6b66af2f83 fix possible cnf problem
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/01: fix possible cnf problem
Posted by ca...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch fix_cnf
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f6b66af2f831ceb9839d9e5e74464c0816161e3d
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Tue Jul 30 16:39:43 2024 +0800
fix possible cnf problem
---
.../alignbydevice/IoTDBAlignByDeviceTableIT.java | 21 ++++++++++
.../planner/distribution/DistributionPlanner.java | 3 +-
.../plan/planner/plan/DistributedQueryPlan.java | 20 +++-------
.../distribute/TableDistributedPlanner.java | 46 ++++++++++++----------
.../optimizations/PushPredicateIntoTableScan.java | 12 +++++-
.../plan/relational/analyzer/AnalyzerTest.java | 21 +++++-----
6 files changed, 74 insertions(+), 49 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/old/alignbydevice/IoTDBAlignByDeviceTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/old/alignbydevice/IoTDBAlignByDeviceTableIT.java
index 952def9147e..35f56e56ab6 100644
--- a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/old/alignbydevice/IoTDBAlignByDeviceTableIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/old/alignbydevice/IoTDBAlignByDeviceTableIT.java
@@ -288,6 +288,27 @@ public class IoTDBAlignByDeviceTableIT {
DATABASE_NAME);
}
+ @Test
+ public void predicateCannotNormalizedTest() {
+ String[] expectedHeader = new String[] {"time", "device_id", "s0", "s1", "s2"};
+ String[] retArray =
+ new String[] {
+ "1970-01-01T00:00:00.001Z,d0,101,1101,null,",
+ "1970-01-01T00:00:00.002Z,d0,10000,40000,2.22,",
+ "1970-01-01T00:00:00.050Z,d0,10000,50000,null,",
+ "1970-01-01T00:00:00.100Z,d0,99,199,null,",
+ "1970-01-01T00:00:00.101Z,d0,99,199,null,",
+ "1970-01-01T00:00:00.103Z,d0,99,199,null,",
+ "1970-01-01T00:00:00.105Z,d0,99,199,11.11,",
+ "1970-01-01T00:00:01.000Z,d0,22222,55555,1000.11,",
+ };
+ tableResultSetEqualTest(
+ "select time, device_id, s0,s1,s2 from vehicle where (((\"time\" > 10) AND (\"s1\" > 190)) OR (\"s2\" > 190.0) OR ((\"time\" < 4) AND (\"s1\" > 100))) order by device_id, time",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
//
// @Test
// public void aggregateTest() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanner.java
index dd0974ac635..2fd9ee783a7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanner.java
@@ -218,8 +218,7 @@ public class DistributionPlanner {
setSinkForRootInstance(subPlan, fragmentInstances);
}
- return new DistributedQueryPlan(
- logicalPlan.getContext(), subPlan, subPlan.getPlanFragmentList(), fragmentInstances);
+ return new DistributedQueryPlan(subPlan, fragmentInstances);
}
// Convert fragment to detailed instance
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/DistributedQueryPlan.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/DistributedQueryPlan.java
index cedc99669a7..0a9135ff9f3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/DistributedQueryPlan.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/DistributedQueryPlan.java
@@ -18,39 +18,29 @@
*/
package org.apache.iotdb.db.queryengine.plan.planner.plan;
-import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.commons.utils.TestOnly;
import java.util.List;
+import java.util.stream.Collectors;
public class DistributedQueryPlan {
- private final MPPQueryContext context;
private final SubPlan rootSubPlan;
- private final List<PlanFragment> fragments;
private final List<FragmentInstance> instances;
- public DistributedQueryPlan(
- MPPQueryContext context,
- SubPlan rootSubPlan,
- List<PlanFragment> fragments,
- List<FragmentInstance> instances) {
- this.context = context;
+ public DistributedQueryPlan(SubPlan rootSubPlan, List<FragmentInstance> instances) {
this.rootSubPlan = rootSubPlan;
- this.fragments = fragments;
this.instances = instances;
}
+ @TestOnly
public List<PlanFragment> getFragments() {
- return fragments;
+ return instances.stream().map(FragmentInstance::getFragment).collect(Collectors.toList());
}
public SubPlan getRootSubPlan() {
return rootSubPlan;
}
- public MPPQueryContext getContext() {
- return context;
- }
-
public List<FragmentInstance> getInstances() {
return instances;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanner.java
index 51e53ef093e..a31fbfa6003 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanner.java
@@ -76,6 +76,7 @@ public class TableDistributedPlanner {
TableDistributedPlanGenerator.PlanContext planContext =
new TableDistributedPlanGenerator.PlanContext();
PlanNode outputNodeWithExchange = generateDistributedPlanWithOptimize(planContext);
+
if (analysis.getStatement() instanceof Query) {
analysis
.getRespDatasetHeader()
@@ -83,27 +84,7 @@ public class TableDistributedPlanner {
}
adjustUpStream(outputNodeWithExchange, planContext);
- // generate subPlan
- SubPlan subPlan =
- new SubPlanGenerator()
- .splitToSubPlan(logicalQueryPlan.getContext().getQueryId(), outputNodeWithExchange);
- subPlan.getPlanFragment().setRoot(true);
-
- // generate fragment instances
- List<FragmentInstance> fragmentInstances =
- mppQueryContext.getQueryType() == QueryType.READ
- ? new TableModelQueryFragmentPlanner(subPlan, analysis, mppQueryContext).plan()
- : new WriteFragmentParallelPlanner(
- subPlan, analysis, mppQueryContext, WritePlanNode::splitByPartition)
- .parallelPlan();
-
- // only execute this step for READ operation
- if (mppQueryContext.getQueryType() == QueryType.READ) {
- setSinkForRootInstance(subPlan, fragmentInstances);
- }
-
- return new DistributedQueryPlan(
- logicalQueryPlan.getContext(), subPlan, subPlan.getPlanFragmentList(), fragmentInstances);
+ return generateDistributedPlan(outputNodeWithExchange);
}
public PlanNode generateDistributedPlanWithOptimize(
@@ -140,6 +121,29 @@ public class TableDistributedPlanner {
return new AddExchangeNodes(mppQueryContext).addExchangeNodes(distributedPlan, planContext);
}
+ private DistributedQueryPlan generateDistributedPlan(PlanNode outputNodeWithExchange) {
+ // generate subPlan
+ SubPlan subPlan =
+ new SubPlanGenerator()
+ .splitToSubPlan(logicalQueryPlan.getContext().getQueryId(), outputNodeWithExchange);
+ subPlan.getPlanFragment().setRoot(true);
+
+ // generate fragment instances
+ List<FragmentInstance> fragmentInstances =
+ mppQueryContext.getQueryType() == QueryType.READ
+ ? new TableModelQueryFragmentPlanner(subPlan, analysis, mppQueryContext).plan()
+ : new WriteFragmentParallelPlanner(
+ subPlan, analysis, mppQueryContext, WritePlanNode::splitByPartition)
+ .parallelPlan();
+
+ // only execute this step for READ operation
+ if (mppQueryContext.getQueryType() == QueryType.READ) {
+ setSinkForRootInstance(subPlan, fragmentInstances);
+ }
+
+ return new DistributedQueryPlan(subPlan, fragmentInstances);
+ }
+
public void setSinkForRootInstance(SubPlan subPlan, List<FragmentInstance> instances) {
FragmentInstance rootInstance = null;
for (FragmentInstance instance : instances) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
index 4d4c6152493..45f2e6c8b9f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
@@ -129,13 +129,21 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
if (node.getPredicate() != null) {
+ predicate = node.getPredicate();
+
// when exist diff function, predicate can not be pushed down into TableScanNode
- if (containsDiffFunction(node.getPredicate())) {
+ if (containsDiffFunction(predicate)) {
node.setChild(node.getChild().accept(this, context));
return node;
}
- predicate = node.getPredicate();
+ // for predicate which is not cnf normalized
+ if (predicate instanceof LogicalExpression
+ && ((LogicalExpression) predicate).getOperator() == LogicalExpression.Operator.OR
+ && ((LogicalExpression) predicate).getTerms().size() > 1) {
+ node.setChild(node.getChild().accept(this, context));
+ return node;
+ }
if (node.getChild() instanceof TableScanNode) {
// child of FilterNode is TableScanNode, means FilterNode must get from where clause
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java
index 50dab345fe0..3ff4e157d5c 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java
@@ -117,6 +117,7 @@ public class AnalyzerTest {
"db",
IClientSession.SqlDialect.TABLE);
Metadata metadata = new TestMatadata();
+ WarningCollector warningCollector = NOOP;
String sql;
Analysis actualAnalysis;
MPPQueryContext context;
@@ -777,8 +778,7 @@ public class AnalyzerTest {
context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
actualAnalysis = analyzeSQL(sql, metadata);
logicalQueryPlan =
- new LogicalPlanner(context, metadata, sessionInfo, WarningCollector.NOOP)
- .plan(actualAnalysis);
+ new LogicalPlanner(context, metadata, sessionInfo, warningCollector).plan(actualAnalysis);
rootNode = logicalQueryPlan.getRootNode();
assertTrue(rootNode.getChildren().get(0) instanceof OffsetNode);
OffsetNode offsetNode = (OffsetNode) rootNode.getChildren().get(0);
@@ -792,8 +792,7 @@ public class AnalyzerTest {
context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
actualAnalysis = analyzeSQL(sql, metadata);
logicalQueryPlan =
- new LogicalPlanner(context, metadata, sessionInfo, WarningCollector.NOOP)
- .plan(actualAnalysis);
+ new LogicalPlanner(context, metadata, sessionInfo, warningCollector).plan(actualAnalysis);
rootNode = logicalQueryPlan.getRootNode();
assertTrue(rootNode.getChildren().get(0) instanceof ProjectNode);
assertTrue(getChildrenNode(rootNode, 2) instanceof OffsetNode);
@@ -804,15 +803,19 @@ public class AnalyzerTest {
}
@Test
- public void sortTest() {
- // when TableScan locates multi regions, use default MergeSortNode
- sql = "SELECT * FROM table1 ";
+ public void predicateCannotNormalizedTest() {
+ sql = "SELECT * FROM table1 where (time > 1 and s1 > 1 or s2 < 7) or (time < 10 and s1 > 4)";
context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
actualAnalysis = analyzeSQL(sql, metadata);
logicalQueryPlan =
- new LogicalPlanner(context, metadata, sessionInfo, WarningCollector.NOOP)
- .plan(actualAnalysis);
+ new LogicalPlanner(context, metadata, sessionInfo, warningCollector).plan(actualAnalysis);
rootNode = logicalQueryPlan.getRootNode();
+ assertTrue(rootNode instanceof OutputNode);
+ assertTrue(getChildrenNode(rootNode, 1) instanceof FilterNode);
+ assertEquals(
+ "(((\"time\" > 1) AND (\"s1\" > 1)) OR (\"s2\" < 7) OR ((\"time\" < 10) AND (\"s1\" > 4)))",
+ ((FilterNode) getChildrenNode(rootNode, 1)).getPredicate().toString());
+ assertTrue(getChildrenNode(rootNode, 2) instanceof TableScanNode);
}
private Metadata mockMetadataForInsertion() {