This is an automated email from the ASF dual-hosted git repository.
chaow pushed a commit to branch rel/0.12
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/0.12 by this push:
new 94a7274 [IOTDB-2009] fix incorrect previous filling (#4386) (#4413)
94a7274 is described below
commit 94a727466ba25fa9d8dea37f5afe990fc4468cde
Author: Zhong Wang <[email protected]>
AuthorDate: Fri Nov 19 10:56:16 2021 +0800
[IOTDB-2009] fix incorrect previous filling (#4386) (#4413)
---
.../cluster/query/fill/ClusterFillExecutor.java | 3 ++-
.../cluster/query/fill/ClusterLinearFill.java | 4 +++-
.../cluster/query/fill/ClusterPreviousFill.java | 25 +++++++++++----------
.../handlers/caller/PreviousFillHandler.java | 26 +++++++++++++++++++---
.../iotdb/db/query/executor/FillQueryExecutor.java | 3 ++-
.../apache/iotdb/db/query/executor/fill/IFill.java | 3 ++-
.../iotdb/db/query/executor/fill/PreviousFill.java | 3 ++-
7 files changed, 47 insertions(+), 20 deletions(-)
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/fill/ClusterFillExecutor.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/fill/ClusterFillExecutor.java
index b364e99..e2c4328 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/query/fill/ClusterFillExecutor.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/fill/ClusterFillExecutor.java
@@ -58,7 +58,8 @@ public class ClusterFillExecutor extends FillQueryExecutor {
TSDataType dataType,
long queryTime,
Set<String> deviceMeasurements,
- QueryContext context) {
+ QueryContext context)
+ throws QueryProcessException, StorageEngineException {
if (fill instanceof LinearFill) {
IFill clusterFill = new ClusterLinearFill((LinearFill) fill,
metaGroupMember);
clusterFill.configureFill(path, dataType, queryTime, deviceMeasurements,
context);
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/fill/ClusterLinearFill.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/fill/ClusterLinearFill.java
index ed1cd7a..34b2706 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/query/fill/ClusterLinearFill.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/fill/ClusterLinearFill.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.cluster.query.fill;
import org.apache.iotdb.cluster.query.aggregate.ClusterAggregator;
import org.apache.iotdb.cluster.server.member.MetaGroupMember;
import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.qp.constant.SQLConstant;
import org.apache.iotdb.db.query.aggregation.AggregateResult;
import org.apache.iotdb.db.query.executor.fill.LinearFill;
@@ -48,7 +49,8 @@ public class ClusterLinearFill extends LinearFill {
}
@Override
- protected TimeValuePair calculatePrecedingPoint() {
+ protected TimeValuePair calculatePrecedingPoint()
+ throws QueryProcessException, StorageEngineException {
// calculate the preceding point can be viewed as a previous fill
ClusterPreviousFill clusterPreviousFill =
new ClusterPreviousFill(dataType, queryTime, beforeRange,
metaGroupMember);
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/fill/ClusterPreviousFill.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/fill/ClusterPreviousFill.java
index 9af7082..b6f9543 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/query/fill/ClusterPreviousFill.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/fill/ClusterPreviousFill.java
@@ -78,14 +78,11 @@ public class ClusterPreviousFill extends PreviousFill {
TSDataType dataType,
long queryTime,
Set<String> deviceMeasurements,
- QueryContext context) {
- try {
- fillResult =
- performPreviousFill(
- path, dataType, queryTime, getBeforeRange(), deviceMeasurements,
context);
- } catch (StorageEngineException e) {
- logger.error("Failed to configure previous fill for Path {}", path, e);
- }
+ QueryContext context)
+ throws QueryProcessException, StorageEngineException {
+ fillResult =
+ performPreviousFill(
+ path, dataType, queryTime, getBeforeRange(), deviceMeasurements,
context);
}
@Override
@@ -100,7 +97,7 @@ public class ClusterPreviousFill extends PreviousFill {
long beforeRange,
Set<String> deviceMeasurements,
QueryContext context)
- throws StorageEngineException {
+ throws StorageEngineException, QueryProcessException {
// make sure the partition table is new
try {
metaGroupMember.syncLeaderWithConsistencyCheck(false);
@@ -131,10 +128,14 @@ public class ClusterPreviousFill extends PreviousFill {
}
fillService.shutdown();
try {
- fillService.awaitTermination(RaftServer.getReadOperationTimeoutMS(),
TimeUnit.MILLISECONDS);
+ boolean terminated =
+ fillService.awaitTermination(
+ RaftServer.getReadOperationTimeoutMS(), TimeUnit.MILLISECONDS);
+ if (!terminated) {
+ logger.warn("Executor service termination timed out");
+ }
} catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- logger.error("Unexpected interruption when waiting for fill pool to
stop", e);
+ throw new QueryProcessException(e.getMessage());
}
return handler.getResult();
}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/server/handlers/caller/PreviousFillHandler.java
b/cluster/src/main/java/org/apache/iotdb/cluster/server/handlers/caller/PreviousFillHandler.java
index 3b4d073..738eecf 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/server/handlers/caller/PreviousFillHandler.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/server/handlers/caller/PreviousFillHandler.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.cluster.server.handlers.caller;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.utils.SerializeUtils;
import org.apache.iotdb.tsfile.read.TimeValuePair;
@@ -27,6 +28,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -36,9 +39,11 @@ public class PreviousFillHandler implements
AsyncMethodCallback<ByteBuffer> {
private static final long MAX_WAIT_MIN = 3;
private CountDownLatch latch;
private TimeValuePair result = new TimeValuePair(Long.MIN_VALUE, null);
+ private List<Exception> exceptions;
public PreviousFillHandler(CountDownLatch latch) {
this.latch = latch;
+ this.exceptions = new ArrayList<>();
}
@Override
@@ -62,19 +67,34 @@ public class PreviousFillHandler implements
AsyncMethodCallback<ByteBuffer> {
@Override
public synchronized void onError(Exception exception) {
logger.error("Cannot get previous fill result", exception);
+ this.exceptions.add(exception);
latch.countDown();
}
- public TimeValuePair getResult() {
+ public TimeValuePair getResult() throws QueryProcessException {
+ if (!exceptions.isEmpty()) {
+ QueryProcessException e =
+ new QueryProcessException(
+ "Exception happened when performing previous fill. "
+ + "See the suppressed exceptions for causes.");
+ for (Exception exception : exceptions) {
+ e.addSuppressed(exception);
+ }
+ throw e;
+ }
+
try {
if (!latch.await(MAX_WAIT_MIN, TimeUnit.MINUTES)) {
logger.warn(
"Not all nodes returned previous fill result when timed out,
remaining {}",
latch.getCount());
+ throw new QueryProcessException(
+ "Failed to get the previous fill result since "
+ + latch.getCount()
+ + " nodes didn't respond");
}
} catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- logger.error("Unexpected interruption when waiting for the result of
previous fill");
+ throw new QueryProcessException(e.getMessage());
}
return result;
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/executor/FillQueryExecutor.java
b/server/src/main/java/org/apache/iotdb/db/query/executor/FillQueryExecutor.java
index f78e1ae..2ad3df0 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/executor/FillQueryExecutor.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/executor/FillQueryExecutor.java
@@ -137,7 +137,8 @@ public class FillQueryExecutor {
TSDataType dataType,
long queryTime,
Set<String> deviceMeasurements,
- QueryContext context) {
+ QueryContext context)
+ throws QueryProcessException, StorageEngineException {
fill.configureFill(path, dataType, queryTime, deviceMeasurements, context);
return fill;
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/executor/fill/IFill.java
b/server/src/main/java/org/apache/iotdb/db/query/executor/fill/IFill.java
index a8472a4..975a988 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/fill/IFill.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/fill/IFill.java
@@ -48,7 +48,8 @@ public abstract class IFill {
TSDataType dataType,
long queryTime,
Set<String> deviceMeasurements,
- QueryContext context);
+ QueryContext context)
+ throws QueryProcessException, StorageEngineException;
public abstract TimeValuePair getFillResult()
throws IOException, QueryProcessException, StorageEngineException;
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/executor/fill/PreviousFill.java
b/server/src/main/java/org/apache/iotdb/db/query/executor/fill/PreviousFill.java
index c6a9916..1cfd0c9 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/executor/fill/PreviousFill.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/executor/fill/PreviousFill.java
@@ -87,7 +87,8 @@ public class PreviousFill extends IFill {
TSDataType dataType,
long queryTime,
Set<String> sensors,
- QueryContext context) {
+ QueryContext context)
+ throws QueryProcessException, StorageEngineException {
this.seriesPath = path;
this.dataType = dataType;
this.context = context;
