[iotdb] branch master updated: Fix Query Deadlock (#4704)

[jackietien]

This is an automated email from the ASF dual-hosted git repository.

jackietien pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new 9dca629  Fix Query Deadlock (#4704)
9dca629 is described below

commit 9dca629f65ae63f415cce3a1431403bfeedc97ea
Author: Jackie Tien <jackiet...@foxmail.com>
AuthorDate: Tue Jan 4 20:39:23 2022 +0800

    Fix Query Deadlock (#4704)
---
 .../dataset/RawQueryDataSetWithoutValueFilter.java      | 17 +++++++++++++++--
 1 file changed, 15 insertions(+), 2 deletions(-)

diff --git 
a/server/src/main/java/org/apache/iotdb/db/query/dataset/RawQueryDataSetWithoutValueFilter.java
 
b/server/src/main/java/org/apache/iotdb/db/query/dataset/RawQueryDataSetWithoutValueFilter.java
index b4f37aa..d0852be 100644
--- 
a/server/src/main/java/org/apache/iotdb/db/query/dataset/RawQueryDataSetWithoutValueFilter.java
+++ 
b/server/src/main/java/org/apache/iotdb/db/query/dataset/RawQueryDataSetWithoutValueFilter.java
@@ -101,10 +101,18 @@ public class RawQueryDataSetWithoutValueFilter extends 
QueryDataSet
             }
             blockingQueue.put(batchData);
 
+            // has limit clause
             if (batchDataLengthList != null) {
               batchDataLengthList[seriesIndex] += batchData.length();
               if (batchDataLengthList[seriesIndex] >= fetchLimit) {
-                break;
+                // the queue has enough space to hold SignalBatchData, just 
break the while loop
+                if (blockingQueue.remainingCapacity() > 0) {
+                  break;
+                } else { // otherwise, exit without putting SignalBatchData, 
main thread will submit
+                  // a new task again, then it will put SignalBatchData 
successfully
+                  reader.setManagedByQueryManager(false);
+                  return;
+                }
               }
             }
             // if the queue also has free space, just submit another itself
@@ -180,6 +188,9 @@ public class RawQueryDataSetWithoutValueFilter extends 
QueryDataSet
 
   private final long queryId;
 
+  // this field record the original value of offset clause, won't change 
during the query execution
+  protected final int originalRowOffset;
+
   private static final RawQueryReadTaskPoolManager TASK_POOL_MANAGER =
       RawQueryReadTaskPoolManager.getInstance();
 
@@ -199,6 +210,7 @@ public class RawQueryDataSetWithoutValueFilter extends 
QueryDataSet
         queryPlan.getDeduplicatedDataTypes(),
         queryPlan.isAscending());
     this.rowLimit = queryPlan.getRowLimit();
+    this.originalRowOffset = queryPlan.getRowOffset();
     this.rowOffset = queryPlan.getRowOffset();
     this.withoutAnyNull = queryPlan.isWithoutAnyNull();
     this.withoutAllNull = queryPlan.isWithoutAllNull();
@@ -232,6 +244,7 @@ public class RawQueryDataSetWithoutValueFilter extends 
QueryDataSet
    */
   public RawQueryDataSetWithoutValueFilter(long queryId) {
     this.queryId = queryId;
+    this.originalRowOffset = 0;
     blockingQueueArray = new BlockingQueue[0];
     timeHeap = new TimeSelector(0, ascending);
   }
@@ -263,7 +276,7 @@ public class RawQueryDataSetWithoutValueFilter extends 
QueryDataSet
         paths.get(seriesIndex).getFullPath(),
         batchDataLengthList,
         seriesIndex,
-        rowLimit + rowOffset);
+        rowLimit + originalRowOffset);
   }
 
   /**