remove duplicate cause by or and fix bugs

spark-tsfile/src/main/java/org/apache/iotdb/tsfile/qp/QueryProcessor.java

@@ -18,7 +18,6 @@
  */
 package org.apache.iotdb.tsfile.qp;
 
-import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
 import org.apache.iotdb.tsfile.qp.common.FilterOperator;
 import org.apache.iotdb.tsfile.qp.common.SQLConstant;
 import org.apache.iotdb.tsfile.qp.common.SingleQuery;
@@ -29,9 +28,13 @@ import org.apache.iotdb.tsfile.qp.optimizer.DNFFilterOptimizer;
 import org.apache.iotdb.tsfile.qp.optimizer.MergeSingleFilterOptimizer;
 import org.apache.iotdb.tsfile.qp.optimizer.PhysicalOptimizer;
 import org.apache.iotdb.tsfile.qp.optimizer.RemoveNotOptimizer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 
 /**
@@ -73,6 +76,48 @@ public class QueryProcessor {
         } else {
             queryPlans.addAll(new PhysicalOptimizer(columnNames).optimize(null, paths, in, start, end));
         }
+        // merge query plan
+        Map<List<String>, List<TSQueryPlan>> pathMap = new HashMap<>();
+        for(TSQueryPlan tsQueryPlan : queryPlans){
+            if(pathMap.containsKey(tsQueryPlan.getPaths())){
+                pathMap.get(tsQueryPlan.getPaths()).add(tsQueryPlan);
+            }
+            else{
+                List<TSQueryPlan> plans = new ArrayList<>();
+                plans.add(tsQueryPlan);
+                pathMap.put(tsQueryPlan.getPaths(), plans);
+            }
+        }
+
+        queryPlans.clear();
+
+        for(List<TSQueryPlan> plans : pathMap.values()){
+            TSQueryPlan mergePlan = null;
+            for(TSQueryPlan plan : plans){
+                if(mergePlan == null){
+                    mergePlan = plan;
+                }
+                else{
+                    FilterOperator timeFilterOperator = new FilterOperator(SQLConstant.KW_OR);
+                    List<FilterOperator> timeFilterChildren = new ArrayList<>();
+                    timeFilterChildren.add(mergePlan.getTimeFilterOperator());
+                    timeFilterChildren.add(plan.getTimeFilterOperator());
+                    timeFilterOperator.setChildrenList(timeFilterChildren);
+                    mergePlan.setTimeFilterOperator(timeFilterOperator);
+
+                    FilterOperator valueFilterOperator = new FilterOperator(SQLConstant.KW_OR);
+                    List<FilterOperator> valueFilterChildren = new ArrayList<>();
+                    valueFilterChildren.add(mergePlan.getValueFilterOperator());
+                    valueFilterChildren.add(plan.getValueFilterOperator());
+                    valueFilterOperator.setChildrenList(valueFilterChildren);
+                    mergePlan.setValueFilterOperator(valueFilterOperator);
+                }
+            }
+            queryPlans.add(mergePlan);
+        }
+        //
+
+
         return queryPlans;
     }
 

spark-tsfile/src/main/java/org/apache/iotdb/tsfile/qp/common/TSQueryPlan.java

@@ -44,6 +44,14 @@ public class TSQueryPlan {
         return timeFilterOperator;
     }
 
+    public void setTimeFilterOperator(FilterOperator timeFilterOperator) {
+        this.timeFilterOperator = timeFilterOperator;
+    }
+
+    public void setValueFilterOperator(FilterOperator valueFilterOperator) {
+        this.valueFilterOperator = valueFilterOperator;
+    }
+
     public FilterOperator getValueFilterOperator() {
         return valueFilterOperator;
     }

spark-tsfile/src/main/scala/org/apache/iotdb/tsfile/NewConverter.scala

@@ -478,7 +478,7 @@ object NewConverter {
       case SQLConstant.LESSTHAN =>
         val basicOperator = node.asInstanceOf[BasicOperator]
         if (QueryConstant.RESERVED_TIME.equals(basicOperator.getSeriesPath.toLowerCase())) {
-          filter = new GlobalTimeExpression(TimeFilter.eq(java.lang.Long.parseLong(basicOperator.getSeriesValue)))
+          filter = new GlobalTimeExpression(TimeFilter.lt(java.lang.Long.parseLong(basicOperator.getSeriesValue)))
         } else {
           filter = constructExpression(schema, basicOperator.getSeriesPath, basicOperator.getSeriesValue, FilterTypes.Lt, device_name)
         }
@@ -487,7 +487,7 @@ object NewConverter {
       case SQLConstant.LESSTHANOREQUALTO =>
         val basicOperator = node.asInstanceOf[BasicOperator]
         if (QueryConstant.RESERVED_TIME.equals(basicOperator.getSeriesPath.toLowerCase())) {
-          filter = new GlobalTimeExpression(TimeFilter.eq(java.lang.Long.parseLong(basicOperator.getSeriesValue)))
+          filter = new GlobalTimeExpression(TimeFilter.ltEq(java.lang.Long.parseLong(basicOperator.getSeriesValue)))
         } else {
           filter = constructExpression(schema, basicOperator.getSeriesPath, basicOperator.getSeriesValue, FilterTypes.LtEq, device_name)
         }
@@ -496,7 +496,7 @@ object NewConverter {
       case SQLConstant.GREATERTHAN =>
         val basicOperator = node.asInstanceOf[BasicOperator]
         if (QueryConstant.RESERVED_TIME.equals(basicOperator.getSeriesPath.toLowerCase())) {
-          filter = new GlobalTimeExpression(TimeFilter.eq(java.lang.Long.parseLong(basicOperator.getSeriesValue)))
+          filter = new GlobalTimeExpression(TimeFilter.gt(java.lang.Long.parseLong(basicOperator.getSeriesValue)))
         } else {
           filter = constructExpression(schema, basicOperator.getSeriesPath, basicOperator.getSeriesValue, FilterTypes.Gt, device_name)
         }
@@ -505,7 +505,7 @@ object NewConverter {
       case SQLConstant.GREATERTHANOREQUALTO =>
         val basicOperator = node.asInstanceOf[BasicOperator]
         if (QueryConstant.RESERVED_TIME.equals(basicOperator.getSeriesPath.toLowerCase())) {
-          filter = new GlobalTimeExpression(TimeFilter.eq(java.lang.Long.parseLong(basicOperator.getSeriesValue)))
+          filter = new GlobalTimeExpression(TimeFilter.gtEq(java.lang.Long.parseLong(basicOperator.getSeriesValue)))
         } else {
           filter = constructExpression(schema, basicOperator.getSeriesPath, basicOperator.getSeriesValue, FilterTypes.GtEq, device_name)
         }