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 1b377c049d8 Change Query Data Source init way from lock to tryLock &
Add more log and remove useless synchronize
1b377c049d8 is described below
commit 1b377c049d80a5e83520746c0952a83f5f03bac0
Author: Jackie Tien <[email protected]>
AuthorDate: Fri Aug 22 17:11:11 2025 +0800
Change Query Data Source init way from lock to tryLock & Add more log and
remove useless synchronize
---
.../queryengine/execution/driver/DataDriver.java | 44 ++-
.../fragment/FakedFragmentInstanceContext.java | 5 +-
.../fragment/FragmentInstanceContext.java | 244 ++++++++----
.../fragment/FragmentInstanceExecution.java | 4 +-
.../queryengine/metric/QueryResourceMetricSet.java | 26 +-
.../FragmentInstanceStatisticsDrawer.java | 11 +
.../db/storageengine/dataregion/DataRegion.java | 413 ++++++++++++++-------
.../dataregion/IDataRegionForQuery.java | 29 +-
.../dataregion/VirtualDataRegion.java | 36 +-
.../memtable/AbstractWritableMemChunk.java | 11 +
.../dataregion/memtable/TsFileProcessor.java | 65 +++-
.../dataregion/tsfile/TsFileManager.java | 43 ++-
.../db/queryengine/execution/DataDriverTest.java | 15 +-
.../src/main/thrift/datanode.thrift | 1 +
14 files changed, 685 insertions(+), 262 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/DataDriver.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/DataDriver.java
index a225793437f..743a05d3777 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/DataDriver.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/DataDriver.java
@@ -32,6 +32,7 @@ import javax.annotation.concurrent.NotThreadSafe;
import java.util.List;
import static
org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet.QUERY_RESOURCE_INIT;
+import static
org.apache.iotdb.db.storageengine.dataregion.VirtualDataRegion.UNFINISHED_QUERY_DATA_SOURCE;
/**
* One {@link DataDriver} is responsible for one {@link FragmentInstance}
which is for data query,
@@ -40,7 +41,7 @@ import static
org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet.QUE
@NotThreadSafe
public class DataDriver extends Driver {
- private boolean init;
+ private boolean init = false;
// Unit : Byte
private final long estimatedMemorySize;
@@ -55,7 +56,13 @@ public class DataDriver extends Driver {
protected boolean init(SettableFuture<?> blockedFuture) {
if (!init) {
try {
- initialize();
+ if (!initialize()) { // failed to init this time, but now exception
thrown, possibly failed
+ // to acquire lock within the specific time
+ blockedFuture.set(null);
+ return false;
+ } else {
+ return true;
+ }
} catch (Throwable t) {
LOGGER.error(
"Failed to do the initialization for driver {} ",
driverContext.getDriverTaskID(), t);
@@ -78,8 +85,7 @@ public class DataDriver extends Driver {
* org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource} is
null after
* initialization, IllegalStateException will be thrown
*/
- private void initialize() throws QueryProcessException {
- long startTime = System.nanoTime();
+ private boolean initialize() throws QueryProcessException {
try {
List<DataSourceOperator> sourceOperators =
((DataDriverContext) driverContext).getSourceOperators();
@@ -90,20 +96,30 @@ public class DataDriver extends Driver {
// for some reasons, we may get null QueryDataSource here.
// And it's safe for us to throw this exception here in such case.
throw new IllegalStateException("QueryDataSource should never be
null!");
+ } else if (dataSource == UNFINISHED_QUERY_DATA_SOURCE) {
+ // init query data source timeout. Maybe failed to acquire the read
lock within the
+ // specified time
+ // do nothing, wait for next try
+ } else {
+ sourceOperators.forEach(
+ sourceOperator -> {
+ // Construct QueryDataSource for source operator
+ sourceOperator.initQueryDataSource(dataSource.clone());
+ });
+ this.init = true;
}
- sourceOperators.forEach(
- sourceOperator -> {
- // Construct QueryDataSource for source operator
- sourceOperator.initQueryDataSource(dataSource.clone());
- });
+ } else {
+ this.init = true;
}
-
- this.init = true;
} finally {
- ((DataDriverContext) driverContext).clearSourceOperators();
- QUERY_EXECUTION_METRICS.recordExecutionCost(
- QUERY_RESOURCE_INIT, System.nanoTime() - startTime);
+ if (this.init) {
+ ((DataDriverContext) driverContext).clearSourceOperators();
+ QUERY_EXECUTION_METRICS.recordExecutionCost(
+ QUERY_RESOURCE_INIT,
+
driverContext.getFragmentInstanceContext().getInitQueryDataSourceCost());
+ }
}
+ return this.init;
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FakedFragmentInstanceContext.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FakedFragmentInstanceContext.java
index ea2757345dd..26bfb723917 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FakedFragmentInstanceContext.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FakedFragmentInstanceContext.java
@@ -48,7 +48,7 @@ public class FakedFragmentInstanceContext extends
FragmentInstanceContext {
public void initQueryDataSource(IFullPath sourcePath) throws
QueryProcessException {
- dataRegion.readLock();
+ dataRegion.tryReadLock(Long.MAX_VALUE);
try {
this.sharedQueryDataSource =
dataRegion.query(
@@ -56,7 +56,8 @@ public class FakedFragmentInstanceContext extends
FragmentInstanceContext {
sourcePath.getDeviceId(),
this,
getGlobalTimeFilter(),
- null);
+ null,
+ Long.MAX_VALUE);
// used files should be added before mergeLock is unlocked, or they may
be deleted by
// running merge
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java
index 50c9afd7ff0..40b16de4ad0 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java
@@ -35,6 +35,7 @@ import org.apache.iotdb.db.queryengine.common.QueryId;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
import org.apache.iotdb.db.queryengine.metric.DriverSchedulerMetricSet;
import org.apache.iotdb.db.queryengine.metric.QueryRelatedResourceMetricSet;
+import org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet;
import org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet;
import
org.apache.iotdb.db.queryengine.plan.planner.memory.MemoryReservationManager;
import
org.apache.iotdb.db.queryengine.plan.planner.memory.ThreadSafeMemoryReservationManager;
@@ -78,13 +79,14 @@ import java.util.stream.Collectors;
import static
org.apache.iotdb.db.queryengine.metric.DriverSchedulerMetricSet.BLOCK_QUEUED_TIME;
import static
org.apache.iotdb.db.queryengine.metric.DriverSchedulerMetricSet.READY_QUEUED_TIME;
import static
org.apache.iotdb.db.storageengine.dataregion.VirtualDataRegion.EMPTY_QUERY_DATA_SOURCE;
+import static
org.apache.iotdb.db.storageengine.dataregion.VirtualDataRegion.UNFINISHED_QUERY_DATA_SOURCE;
import static org.apache.iotdb.db.utils.ErrorHandlingUtils.getRootCause;
import static org.apache.iotdb.rpc.TSStatusCode.DATE_OUT_OF_RANGE;
public class FragmentInstanceContext extends QueryContext {
private static final Logger LOGGER =
LoggerFactory.getLogger(FragmentInstanceContext.class);
- private static final IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
+ private static final IoTDBConfig CONFIG =
IoTDBDescriptor.getInstance().getConfig();
private static final long END_TIME_INITIAL_VALUE = -1L;
// wait over 5s for driver to close is abnormal
private static final long LONG_WAIT_DURATION = 5_000_000_000L;
@@ -147,6 +149,7 @@ public class FragmentInstanceContext extends QueryContext {
private TFetchFragmentInstanceStatisticsResp fragmentInstanceStatistics =
null;
private long initQueryDataSourceCost = 0;
+ private int initQueryDataSourceRetryCount = 0;
private final AtomicLong readyQueueTime = new AtomicLong(0);
private final AtomicLong blockQueueTime = new AtomicLong(0);
private long unclosedSeqFileNum = 0;
@@ -345,7 +348,7 @@ public class FragmentInstanceContext extends QueryContext {
PatternTreeMap<ModEntry, PatternTreeMapFactory.ModsSerializer>
allMods =
loadAllModificationsFromDisk(resource);
atomicReference.set(allMods);
- if (cachedModEntriesSize.get() >=
config.getModsCacheSizeLimitPerFI()) {
+ if (cachedModEntriesSize.get() >=
CONFIG.getModsCacheSizeLimitPerFI()) {
return null;
}
long memCost =
@@ -353,7 +356,7 @@ public class FragmentInstanceContext extends QueryContext {
+
RamUsageEstimator.SHALLOW_SIZE_OF_CONCURRENT_HASHMAP_ENTRY;
long alreadyUsedMemoryForCachedModEntries =
cachedModEntriesSize.get();
while (alreadyUsedMemoryForCachedModEntries + memCost
- < config.getModsCacheSizeLimitPerFI()) {
+ < CONFIG.getModsCacheSizeLimitPerFI()) {
if (cachedModEntriesSize.compareAndSet(
alreadyUsedMemoryForCachedModEntries,
alreadyUsedMemoryForCachedModEntries + memCost)) {
@@ -456,10 +459,6 @@ public class FragmentInstanceContext extends QueryContext {
return dataNodeQueryContext;
}
- public void setDataNodeQueryContext(DataNodeQueryContext
dataNodeQueryContext) {
- this.dataNodeQueryContext = dataNodeQueryContext;
- }
-
public FragmentInstanceInfo getInstanceInfo() {
FragmentInstanceState state = stateMachine.getState();
long endTime = getEndTime();
@@ -559,11 +558,11 @@ public class FragmentInstanceContext extends QueryContext
{
memoryReservationManager.releaseAllReservedMemory();
}
- public void initQueryDataSource(List<IFullPath> sourcePaths) throws
QueryProcessException {
+ public boolean initQueryDataSource(List<IFullPath> sourcePaths) throws
QueryProcessException {
long startTime = System.nanoTime();
if (sourcePaths == null || sourcePaths.isEmpty()) {
this.sharedQueryDataSource = EMPTY_QUERY_DATA_SOURCE;
- return;
+ return true;
}
IDeviceID singleDeviceId = null;
@@ -586,82 +585,135 @@ public class FragmentInstanceContext extends
QueryContext {
}
}
- dataRegion.readLock();
- try {
-
- this.sharedQueryDataSource =
- dataRegion.query(
- sourcePaths,
- // when all the selected series are under the same device, the
QueryDataSource will be
- // filtered according to timeIndex
- singleDeviceId,
- this,
- // time filter may be stateful, so we need to copy it
- globalTimeFilter != null ? globalTimeFilter.copy() : null,
- timePartitions);
-
- // used files should be added before mergeLock is unlocked, or they may
be deleted by
- // running merge
- if (sharedQueryDataSource != null) {
- closedFilePaths = new HashSet<>();
- unClosedFilePaths = new HashSet<>();
- addUsedFilesForQuery((QueryDataSource) sharedQueryDataSource);
- ((QueryDataSource)
sharedQueryDataSource).setSingleDevice(singleDeviceId != null);
+ long waitForLockTime = CONFIG.getDriverTaskExecutionTimeSliceInMs();
+ long startAcquireLockTime = System.nanoTime();
+ if (dataRegion.tryReadLock(waitForLockTime)) {
+ try {
+ // minus already consumed time
+ waitForLockTime -= (System.nanoTime() - startAcquireLockTime) /
1_000_000;
+
+ // no remaining time slice
+ if (waitForLockTime <= 0) {
+ return false;
+ }
+
+ this.sharedQueryDataSource =
+ dataRegion.query(
+ sourcePaths,
+ // when all the selected series are under the same device, the
QueryDataSource will
+ // be
+ // filtered according to timeIndex
+ singleDeviceId,
+ this,
+ // time filter may be stateful, so we need to copy it
+ globalTimeFilter != null ? globalTimeFilter.copy() : null,
+ timePartitions,
+ waitForLockTime);
+
+ // used files should be added before mergeLock is unlocked, or they
may be deleted by
+ // running merge
+ if (sharedQueryDataSource != null) {
+ closedFilePaths = new HashSet<>();
+ unClosedFilePaths = new HashSet<>();
+ addUsedFilesForQuery((QueryDataSource) sharedQueryDataSource);
+ ((QueryDataSource)
sharedQueryDataSource).setSingleDevice(singleDeviceId != null);
+ return true;
+ } else {
+ // failed to acquire lock within the specific time
+ return false;
+ }
+ } finally {
+ addInitQueryDataSourceCost(System.nanoTime() - startTime);
+ dataRegion.readUnlock();
}
- } finally {
- setInitQueryDataSourceCost(System.nanoTime() - startTime);
- dataRegion.readUnlock();
+ } else {
+ addInitQueryDataSourceCost(System.nanoTime() - startTime);
+ return false;
}
}
- public void initRegionScanQueryDataSource(Map<IDeviceID, DeviceContext>
devicePathsToContext)
- throws QueryProcessException {
+ public boolean initRegionScanQueryDataSource(Map<IDeviceID, DeviceContext>
devicePathsToContext) {
long startTime = System.nanoTime();
if (devicePathsToContext == null) {
- return;
+ return true;
}
- dataRegion.readLock();
- try {
- this.sharedQueryDataSource =
- dataRegion.queryForDeviceRegionScan(
- devicePathsToContext,
- this,
- globalTimeFilter != null ? globalTimeFilter.copy() : null,
- timePartitions);
-
- if (sharedQueryDataSource != null) {
- closedFilePaths = new HashSet<>();
- unClosedFilePaths = new HashSet<>();
- addUsedFilesForRegionQuery((QueryDataSourceForRegionScan)
sharedQueryDataSource);
+
+ long waitForLockTime = CONFIG.getDriverTaskExecutionTimeSliceInMs();
+ if (dataRegion.tryReadLock(waitForLockTime)) {
+ try {
+ // minus already consumed time
+ waitForLockTime -= (System.nanoTime() - startTime) / 1_000_000;
+
+ // no remaining time slice
+ if (waitForLockTime <= 0) {
+ return false;
+ }
+ this.sharedQueryDataSource =
+ dataRegion.queryForDeviceRegionScan(
+ devicePathsToContext,
+ this,
+ globalTimeFilter != null ? globalTimeFilter.copy() : null,
+ timePartitions,
+ waitForLockTime);
+
+ if (sharedQueryDataSource != null) {
+ closedFilePaths = new HashSet<>();
+ unClosedFilePaths = new HashSet<>();
+ addUsedFilesForRegionQuery((QueryDataSourceForRegionScan)
sharedQueryDataSource);
+ return true;
+ } else {
+ // failed to acquire lock within the specific time
+ return false;
+ }
+ } finally {
+ addInitQueryDataSourceCost(System.nanoTime() - startTime);
+ dataRegion.readUnlock();
}
- } finally {
- setInitQueryDataSourceCost(System.nanoTime() - startTime);
- dataRegion.readUnlock();
+ } else {
+ addInitQueryDataSourceCost(System.nanoTime() - startTime);
+ return false;
}
}
- public void initRegionScanQueryDataSource(List<IFullPath> pathList) throws
QueryProcessException {
+ public boolean initRegionScanQueryDataSource(List<IFullPath> pathList) {
long startTime = System.nanoTime();
if (pathList == null) {
- return;
+ return true;
}
- dataRegion.readLock();
- try {
- this.sharedQueryDataSource =
- dataRegion.queryForSeriesRegionScan(
- pathList,
- this,
- globalTimeFilter != null ? globalTimeFilter.copy() : null,
- timePartitions);
-
- if (sharedQueryDataSource != null) {
- closedFilePaths = new HashSet<>();
- unClosedFilePaths = new HashSet<>();
- addUsedFilesForRegionQuery((QueryDataSourceForRegionScan)
sharedQueryDataSource);
+ long waitForLockTime = CONFIG.getDriverTaskExecutionTimeSliceInMs();
+ if (dataRegion.tryReadLock(waitForLockTime)) {
+ // minus already consumed time
+ waitForLockTime -= (System.nanoTime() - startTime) / 1_000_000;
+
+ // no remaining time slice
+ if (waitForLockTime <= 0) {
+ return false;
+ }
+ try {
+ this.sharedQueryDataSource =
+ dataRegion.queryForSeriesRegionScan(
+ pathList,
+ this,
+ globalTimeFilter != null ? globalTimeFilter.copy() : null,
+ timePartitions,
+ waitForLockTime);
+
+ if (sharedQueryDataSource != null) {
+ closedFilePaths = new HashSet<>();
+ unClosedFilePaths = new HashSet<>();
+ addUsedFilesForRegionQuery((QueryDataSourceForRegionScan)
sharedQueryDataSource);
+ return true;
+ } else {
+ // failed to acquire lock within the specific time
+ return false;
+ }
+ } finally {
+ addInitQueryDataSourceCost(System.nanoTime() - startTime);
+ dataRegion.readUnlock();
}
- } finally {
- setInitQueryDataSourceCost(System.nanoTime() - startTime);
- dataRegion.readUnlock();
+ } else {
+ addInitQueryDataSourceCost(System.nanoTime() - startTime);
+ return false;
}
}
@@ -669,17 +721,26 @@ public class FragmentInstanceContext extends QueryContext
{
if (sharedQueryDataSource == null) {
switch (queryDataSourceType) {
case SERIES_SCAN:
- initQueryDataSource(sourcePaths);
- // Friendly for gc
- sourcePaths = null;
+ if (initQueryDataSource(sourcePaths)) {
+ // Friendly for gc
+ sourcePaths = null;
+ } else {
+ return getUnfinishedQueryDataSource();
+ }
break;
case DEVICE_REGION_SCAN:
- initRegionScanQueryDataSource(devicePathsToContext);
- devicePathsToContext = null;
+ if (initRegionScanQueryDataSource(devicePathsToContext)) {
+ devicePathsToContext = null;
+ } else {
+ return getUnfinishedQueryDataSource();
+ }
break;
case TIME_SERIES_REGION_SCAN:
- initRegionScanQueryDataSource(sourcePaths);
- sourcePaths = null;
+ if (initRegionScanQueryDataSource(sourcePaths)) {
+ sourcePaths = null;
+ } else {
+ return getUnfinishedQueryDataSource();
+ }
break;
default:
throw new QueryProcessException(
@@ -689,6 +750,18 @@ public class FragmentInstanceContext extends QueryContext {
return sharedQueryDataSource;
}
+ private IQueryDataSource getUnfinishedQueryDataSource() {
+ increaseInitQueryDataSourceRetryCount();
+ // record warn log every 10 times retry
+ if (initQueryDataSourceRetryCount % 10 == 0) {
+ LOGGER.warn(
+ "Failed to acquire the read lock of DataRegion-{} for {} times",
+ dataRegion == null ? "UNKNOWN" : dataRegion.getDataRegionId(),
+ initQueryDataSourceRetryCount);
+ }
+ return UNFINISHED_QUERY_DATA_SOURCE;
+ }
+
/** Lock and check if tsFileResource is deleted */
private boolean processTsFileResource(TsFileResource tsFileResource, boolean
isClosed) {
addFilePathToMap(tsFileResource, isClosed);
@@ -866,6 +939,9 @@ public class FragmentInstanceContext extends QueryContext {
QueryRelatedResourceMetricSet.getInstance().updateFragmentInstanceTime(durationTime);
+ QueryResourceMetricSet.getInstance()
+ .recordInitQueryResourceRetryCount(getInitQueryDataSourceRetryCount());
+
SeriesScanCostMetricSet.getInstance()
.recordBloomFilterMetrics(
getQueryStatistics().getLoadBloomFilterFromCacheCount().get(),
@@ -949,7 +1025,7 @@ public class FragmentInstanceContext extends QueryContext {
.updatePageReaderMemoryUsage(getQueryStatistics().getPageReaderMaxUsedMemorySize().get());
}
- private synchronized void releaseDataNodeQueryContext() {
+ private void releaseDataNodeQueryContext() {
if (dataNodeQueryContextMap == null) {
// this process is in fetch schema, nothing need to release
return;
@@ -986,14 +1062,22 @@ public class FragmentInstanceContext extends
QueryContext {
return fragmentInstanceStatistics;
}
- public void setInitQueryDataSourceCost(long initQueryDataSourceCost) {
- this.initQueryDataSourceCost = initQueryDataSourceCost;
+ public void addInitQueryDataSourceCost(long initQueryDataSourceCost) {
+ this.initQueryDataSourceCost += initQueryDataSourceCost;
}
public long getInitQueryDataSourceCost() {
return initQueryDataSourceCost;
}
+ public void increaseInitQueryDataSourceRetryCount() {
+ this.initQueryDataSourceRetryCount++;
+ }
+
+ public int getInitQueryDataSourceRetryCount() {
+ return initQueryDataSourceRetryCount;
+ }
+
public void addReadyQueuedTime(long time) {
readyQueueTime.addAndGet(time);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecution.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecution.java
index 92471853f60..a3d7a887352 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecution.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecution.java
@@ -30,7 +30,6 @@ import
org.apache.iotdb.db.queryengine.execution.exchange.MPPDataExchangeManager
import org.apache.iotdb.db.queryengine.execution.exchange.sink.ISink;
import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
import org.apache.iotdb.db.queryengine.execution.schedule.IDriverScheduler;
-import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
import org.apache.iotdb.db.storageengine.dataregion.IDataRegionForQuery;
import org.apache.iotdb.db.storageengine.dataregion.VirtualDataRegion;
import org.apache.iotdb.db.utils.SetThreadName;
@@ -153,7 +152,7 @@ public class FragmentInstanceExecution {
// We don't need to output the region having ExplainAnalyzeOperator only.
return false;
}
- statistics.setDataRegion(((DataRegion)
context.getDataRegion()).getDataRegionId());
+ statistics.setDataRegion(context.getDataRegion().getDataRegionId());
statistics.setIp(CONFIG.getInternalAddress() + ":" +
CONFIG.getInternalPort());
statistics.setStartTimeInMS(context.getStartTime());
statistics.setEndTimeInMS(
@@ -163,6 +162,7 @@ public class FragmentInstanceExecution {
statistics.setReadyQueuedTime(context.getReadyQueueTime());
statistics.setInitDataQuerySourceCost(context.getInitQueryDataSourceCost());
+
statistics.setInitDataQuerySourceRetryCount(context.getInitQueryDataSourceRetryCount());
statistics.setSeqClosednNum(context.getClosedSeqFileNum());
statistics.setSeqUnclosedNum(context.getUnclosedSeqFileNum());
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/QueryResourceMetricSet.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/QueryResourceMetricSet.java
index 401fc2e0c8f..241486750a4 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/QueryResourceMetricSet.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/metric/QueryResourceMetricSet.java
@@ -41,10 +41,12 @@ public class QueryResourceMetricSet implements IMetricSet {
public static final String UNSEQUENCE_TSFILE = "unsequence_tsfile";
public static final String FLUSHING_MEMTABLE = "flushing_memtable";
public static final String WORKING_MEMTABLE = "working_memtable";
+ public static final String INIT_QUERY_RESOURCE_RETRY_COUNT = "retry_count";
private Histogram sequenceTsFileHistogram =
DoNothingMetricManager.DO_NOTHING_HISTOGRAM;
private Histogram unsequenceTsFileHistogram =
DoNothingMetricManager.DO_NOTHING_HISTOGRAM;
private Histogram flushingMemTableHistogram =
DoNothingMetricManager.DO_NOTHING_HISTOGRAM;
private Histogram workingMemTableHistogram =
DoNothingMetricManager.DO_NOTHING_HISTOGRAM;
+ private Histogram retryCountHistogram =
DoNothingMetricManager.DO_NOTHING_HISTOGRAM;
public void recordQueryResourceNum(String type, int count) {
switch (type) {
@@ -60,11 +62,22 @@ public class QueryResourceMetricSet implements IMetricSet {
case WORKING_MEMTABLE:
workingMemTableHistogram.update(count);
break;
+ case INIT_QUERY_RESOURCE_RETRY_COUNT:
+ if (count > 0) {
+ retryCountHistogram.update(count);
+ }
+ break;
default:
break;
}
}
+ public void recordInitQueryResourceRetryCount(int count) {
+ if (count > 0) {
+ retryCountHistogram.update(count);
+ }
+ }
+
@Override
public void bindTo(AbstractMetricService metricService) {
sequenceTsFileHistogram =
@@ -91,11 +104,22 @@ public class QueryResourceMetricSet implements IMetricSet {
MetricLevel.IMPORTANT,
Tag.TYPE.toString(),
WORKING_MEMTABLE);
+ retryCountHistogram =
+ metricService.getOrCreateHistogram(
+ Metric.QUERY_RESOURCE.toString(),
+ MetricLevel.IMPORTANT,
+ Tag.TYPE.toString(),
+ INIT_QUERY_RESOURCE_RETRY_COUNT);
}
@Override
public void unbindFrom(AbstractMetricService metricService) {
- Arrays.asList(SEQUENCE_TSFILE, UNSEQUENCE_TSFILE, FLUSHING_MEMTABLE,
WORKING_MEMTABLE)
+ Arrays.asList(
+ SEQUENCE_TSFILE,
+ UNSEQUENCE_TSFILE,
+ FLUSHING_MEMTABLE,
+ WORKING_MEMTABLE,
+ INIT_QUERY_RESOURCE_RETRY_COUNT)
.forEach(
type ->
metricService.remove(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsDrawer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsDrawer.java
index 0141a86e627..b9f1a55b5bd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsDrawer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsDrawer.java
@@ -113,6 +113,17 @@ public class FragmentInstanceStatisticsDrawer {
String.format(
"Cost of initDataQuerySource: %.3f ms",
statistics.getInitDataQuerySourceCost() * NS_TO_MS_FACTOR));
+
+ if (statistics.isSetInitDataQuerySourceRetryCount()
+ && statistics.getInitDataQuerySourceRetryCount() > 0) {
+ addLine(
+ singleFragmentInstanceArea,
+ 1,
+ String.format(
+ "Retry count of initDataQuerySource: %d",
+ statistics.getInitDataQuerySourceRetryCount()));
+ }
+
addLine(
singleFragmentInstanceArea,
1,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index 04f959be5b4..fc7977b8c12 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -266,11 +266,6 @@ public class DataRegion implements IDataRegionForQuery {
/** closeStorageGroupCondition is used to wait for all currently closing
TsFiles to be done. */
private final Object closeStorageGroupCondition = new Object();
- /**
- * Avoid some tsfileResource is changed (e.g., from unsealed to sealed) when
a read is executed.
- */
- private final ReadWriteLock closeQueryLock = new ReentrantReadWriteLock();
-
/** time partition id in the database -> {@link TsFileProcessor} for this
time partition. */
private final TreeMap<Long, TsFileProcessor> workSequenceTsFileProcessors =
new TreeMap<>();
@@ -2073,8 +2068,8 @@ public class DataRegion implements IDataRegionForQuery {
}
}
- /** used for query engine */
@Override
+ @TestOnly
public QueryDataSource query(
List<IFullPath> pathList,
IDeviceID singleDeviceId,
@@ -2082,31 +2077,148 @@ public class DataRegion implements IDataRegionForQuery
{
Filter globalTimeFilter,
List<Long> timePartitions)
throws QueryProcessException {
- try {
- List<TsFileResource> seqResources =
- getFileResourceListForQuery(
- tsFileManager.getTsFileList(true, timePartitions,
globalTimeFilter),
- pathList,
- singleDeviceId,
- context,
- globalTimeFilter,
- true);
- List<TsFileResource> unseqResources =
- getFileResourceListForQuery(
- tsFileManager.getTsFileList(false, timePartitions,
globalTimeFilter),
- pathList,
- singleDeviceId,
- context,
- globalTimeFilter,
- false);
+ return query(
+ pathList, singleDeviceId, context, globalTimeFilter, timePartitions,
Long.MAX_VALUE);
+ }
+
+ /** used for query engine */
+ @Override
+ public QueryDataSource query(
+ List<IFullPath> pathList,
+ IDeviceID singleDeviceId,
+ QueryContext context,
+ Filter globalTimeFilter,
+ List<Long> timePartitions,
+ long waitForLockTimeInMs)
+ throws QueryProcessException {
+
+ Pair<List<TsFileResource>, List<TsFileResource>> pair =
+ tsFileManager.getAllTsFileListForQuery(timePartitions,
globalTimeFilter);
+
+ List<TsFileResource> seqTsFileResouceList = pair.left;
+ List<TsFileResource> unSeqTsFileResouceList = pair.right;
+
+ List<TsFileProcessor> needToUnLockList = new ArrayList<>();
+
+ boolean success =
+ tryGetFLushLock(
+ waitForLockTimeInMs,
+ singleDeviceId,
+ globalTimeFilter,
+ context.isDebug(),
+ seqTsFileResouceList,
+ unSeqTsFileResouceList,
+ needToUnLockList);
+
+ if (success) {
+ try {
+ List<TsFileResource> satisfiedSeqResourceList =
+ getFileResourceListForQuery(
+ seqTsFileResouceList, pathList, singleDeviceId, context,
globalTimeFilter, true);
+
+ List<TsFileResource> satisfiedUnSeqResourceList =
+ getFileResourceListForQuery(
+ unSeqTsFileResouceList, pathList, singleDeviceId, context,
globalTimeFilter, false);
+
+ QUERY_RESOURCE_METRIC_SET.recordQueryResourceNum(
+ SEQUENCE_TSFILE, satisfiedSeqResourceList.size());
+ QUERY_RESOURCE_METRIC_SET.recordQueryResourceNum(
+ UNSEQUENCE_TSFILE, satisfiedUnSeqResourceList.size());
+ return new QueryDataSource(
+ satisfiedSeqResourceList, satisfiedUnSeqResourceList,
databaseName);
+ } catch (MetadataException e) {
+ throw new QueryProcessException(e);
+ } finally {
+ clearAlreadyLockedList(needToUnLockList);
+ }
+ } else {
+ // means that failed to acquire lock within the specific time
+ return null;
+ }
+ }
+
+ /**
+ * try to get flush lock for each unclosed satisfied tsfile
+ *
+ * @return true if lock successfully, otherwise false if return false,
needToUnLockList will
+ * always be empty because this method is responsible for unlocking all
the already-acquiring
+ * lock if return true, the caller is responsible for unlocking all the
already-acquiring lock
+ * in needToUnLockList
+ */
+ private boolean tryGetFLushLock(
+ long waitTimeInMs,
+ IDeviceID singleDeviceId,
+ Filter globalTimeFilter,
+ boolean isDebug,
+ List<TsFileResource> seqResources,
+ List<TsFileResource> unSeqResources,
+ List<TsFileProcessor> needToUnLockList) {
+ // deal with seq resources
+ for (TsFileResource tsFileResource : seqResources) {
+ // only need to acquire flush lock for those unclosed and satisfied
tsfile
+ if (!tsFileResource.isClosed()
+ && tsFileResource.isSatisfied(singleDeviceId, globalTimeFilter,
true, isDebug)) {
+ TsFileProcessor tsFileProcessor = tsFileResource.getProcessor();
+ try {
+ long startTime = System.nanoTime();
+ if (tsFileProcessor.tryReadLock(waitTimeInMs)) {
+ // minus already consumed time
+ waitTimeInMs -= (System.nanoTime() - startTime) / 1_000_000;
- QUERY_RESOURCE_METRIC_SET.recordQueryResourceNum(SEQUENCE_TSFILE,
seqResources.size());
- QUERY_RESOURCE_METRIC_SET.recordQueryResourceNum(UNSEQUENCE_TSFILE,
unseqResources.size());
+ needToUnLockList.add(tsFileProcessor);
- return new QueryDataSource(seqResources, unseqResources, databaseName);
- } catch (MetadataException e) {
- throw new QueryProcessException(e);
+ // no remaining time slice
+ if (waitTimeInMs <= 0) {
+ clearAlreadyLockedList(needToUnLockList);
+ return false;
+ }
+ } else {
+ clearAlreadyLockedList(needToUnLockList);
+ return false;
+ }
+ } catch (InterruptedException e) {
+ clearAlreadyLockedList(needToUnLockList);
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
}
+ // deal with unSeq resources
+ for (TsFileResource tsFileResource : unSeqResources) {
+ if (!tsFileResource.isClosed()
+ && tsFileResource.isSatisfied(singleDeviceId, globalTimeFilter,
false, isDebug)) {
+ TsFileProcessor tsFileProcessor = tsFileResource.getProcessor();
+ try {
+ long startTime = System.nanoTime();
+ if (tsFileProcessor.tryReadLock(waitTimeInMs)) {
+ // minus already consumed time
+ waitTimeInMs -= (System.nanoTime() - startTime) / 1_000_000;
+
+ needToUnLockList.add(tsFileProcessor);
+ // no remaining time slice
+ if (waitTimeInMs <= 0) {
+ clearAlreadyLockedList(needToUnLockList);
+ return false;
+ }
+ } else {
+ clearAlreadyLockedList(needToUnLockList);
+ return false;
+ }
+ } catch (InterruptedException e) {
+ clearAlreadyLockedList(needToUnLockList);
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ private void clearAlreadyLockedList(List<TsFileProcessor> needToUnLockList) {
+ for (TsFileProcessor processor : needToUnLockList) {
+ processor.readUnLock();
+ }
+ needToUnLockList.clear();
}
@Override
@@ -2114,31 +2226,51 @@ public class DataRegion implements IDataRegionForQuery {
List<IFullPath> pathList,
QueryContext queryContext,
Filter globalTimeFilter,
- List<Long> timePartitions)
- throws QueryProcessException {
- try {
- List<IFileScanHandle> seqFileScanHandles =
- getFileHandleListForQuery(
- tsFileManager.getTsFileList(true, timePartitions,
globalTimeFilter),
- pathList,
- queryContext,
- globalTimeFilter,
- true);
- List<IFileScanHandle> unseqFileScanHandles =
- getFileHandleListForQuery(
- tsFileManager.getTsFileList(false, timePartitions,
globalTimeFilter),
- pathList,
- queryContext,
- globalTimeFilter,
- false);
-
- QUERY_RESOURCE_METRIC_SET.recordQueryResourceNum(SEQUENCE_TSFILE,
seqFileScanHandles.size());
- QUERY_RESOURCE_METRIC_SET.recordQueryResourceNum(
- UNSEQUENCE_TSFILE, unseqFileScanHandles.size());
+ List<Long> timePartitions,
+ long waitForLockTimeInMs) {
+ Pair<List<TsFileResource>, List<TsFileResource>> pair =
+ tsFileManager.getAllTsFileListForQuery(timePartitions,
globalTimeFilter);
+
+ List<TsFileResource> seqTsFileResouceList = pair.left;
+ List<TsFileResource> unSeqTsFileResouceList = pair.right;
+
+ List<TsFileProcessor> needToUnLockList = new ArrayList<>();
+
+ boolean success =
+ tryGetFLushLock(
+ waitForLockTimeInMs,
+ null,
+ globalTimeFilter,
+ queryContext.isDebug(),
+ seqTsFileResouceList,
+ unSeqTsFileResouceList,
+ needToUnLockList);
+
+ if (success) {
+ try {
+ List<IFileScanHandle> seqFileScanHandles =
+ getFileHandleListForQuery(
+ seqTsFileResouceList, pathList, queryContext,
globalTimeFilter, true);
+
+ List<IFileScanHandle> unSeqFileScanHandles =
+ getFileHandleListForQuery(
+ tsFileManager.getTsFileList(false, timePartitions,
globalTimeFilter),
+ pathList,
+ queryContext,
+ globalTimeFilter,
+ false);
- return new QueryDataSourceForRegionScan(seqFileScanHandles,
unseqFileScanHandles);
- } catch (MetadataException e) {
- throw new QueryProcessException(e);
+ QUERY_RESOURCE_METRIC_SET.recordQueryResourceNum(
+ SEQUENCE_TSFILE, seqFileScanHandles.size());
+ QUERY_RESOURCE_METRIC_SET.recordQueryResourceNum(
+ UNSEQUENCE_TSFILE, unSeqFileScanHandles.size());
+ return new QueryDataSourceForRegionScan(seqFileScanHandles,
unSeqFileScanHandles);
+ } finally {
+ clearAlreadyLockedList(needToUnLockList);
+ }
+ } else {
+ // means that failed to acquire lock within the specific time
+ return null;
}
}
@@ -2147,25 +2279,19 @@ public class DataRegion implements IDataRegionForQuery {
List<IFullPath> partialPaths,
QueryContext context,
Filter globalTimeFilter,
- boolean isSeq)
- throws MetadataException {
+ boolean isSeq) {
List<IFileScanHandle> fileScanHandles = new ArrayList<>();
for (TsFileResource tsFileResource : tsFileResources) {
if (!tsFileResource.isSatisfied(null, globalTimeFilter, isSeq,
context.isDebug())) {
continue;
}
- closeQueryLock.readLock().lock();
- try {
- if (tsFileResource.isClosed()) {
- fileScanHandles.add(new ClosedFileScanHandleImpl(tsFileResource,
context));
- } else {
- tsFileResource
- .getProcessor()
- .queryForSeriesRegionScan(partialPaths, context,
fileScanHandles);
- }
- } finally {
- closeQueryLock.readLock().unlock();
+ if (tsFileResource.isClosed()) {
+ fileScanHandles.add(new ClosedFileScanHandleImpl(tsFileResource,
context));
+ } else {
+ tsFileResource
+ .getProcessor()
+ .queryForSeriesRegionScanWithoutLock(partialPaths, context,
fileScanHandles);
}
}
return fileScanHandles;
@@ -2176,31 +2302,52 @@ public class DataRegion implements IDataRegionForQuery {
Map<IDeviceID, DeviceContext> devicePathToAligned,
QueryContext queryContext,
Filter globalTimeFilter,
- List<Long> timePartitions)
- throws QueryProcessException {
- try {
- List<IFileScanHandle> seqFileScanHandles =
- getFileHandleListForQuery(
- tsFileManager.getTsFileList(true, timePartitions,
globalTimeFilter),
- devicePathToAligned,
- queryContext,
- globalTimeFilter,
- true);
- List<IFileScanHandle> unseqFileScanHandles =
- getFileHandleListForQuery(
- tsFileManager.getTsFileList(false, timePartitions,
globalTimeFilter),
- devicePathToAligned,
- queryContext,
- globalTimeFilter,
- false);
+ List<Long> timePartitions,
+ long waitForLockTimeInMs) {
- QUERY_RESOURCE_METRIC_SET.recordQueryResourceNum(SEQUENCE_TSFILE,
seqFileScanHandles.size());
- QUERY_RESOURCE_METRIC_SET.recordQueryResourceNum(
- UNSEQUENCE_TSFILE, unseqFileScanHandles.size());
+ Pair<List<TsFileResource>, List<TsFileResource>> pair =
+ tsFileManager.getAllTsFileListForQuery(timePartitions,
globalTimeFilter);
- return new QueryDataSourceForRegionScan(seqFileScanHandles,
unseqFileScanHandles);
- } catch (MetadataException e) {
- throw new QueryProcessException(e);
+ List<TsFileResource> seqTsFileResouceList = pair.left;
+ List<TsFileResource> unSeqTsFileResouceList = pair.right;
+
+ List<TsFileProcessor> needToUnLockList = new ArrayList<>();
+
+ boolean success =
+ tryGetFLushLock(
+ waitForLockTimeInMs,
+ null,
+ globalTimeFilter,
+ queryContext.isDebug(),
+ seqTsFileResouceList,
+ unSeqTsFileResouceList,
+ needToUnLockList);
+
+ if (success) {
+ try {
+ List<IFileScanHandle> seqFileScanHandles =
+ getFileHandleListForQuery(
+ seqTsFileResouceList, devicePathToAligned, queryContext,
globalTimeFilter, true);
+
+ List<IFileScanHandle> unSeqFileScanHandles =
+ getFileHandleListForQuery(
+ tsFileManager.getTsFileList(false, timePartitions,
globalTimeFilter),
+ devicePathToAligned,
+ queryContext,
+ globalTimeFilter,
+ false);
+
+ QUERY_RESOURCE_METRIC_SET.recordQueryResourceNum(
+ SEQUENCE_TSFILE, seqFileScanHandles.size());
+ QUERY_RESOURCE_METRIC_SET.recordQueryResourceNum(
+ UNSEQUENCE_TSFILE, unSeqFileScanHandles.size());
+ return new QueryDataSourceForRegionScan(seqFileScanHandles,
unSeqFileScanHandles);
+ } finally {
+ clearAlreadyLockedList(needToUnLockList);
+ }
+ } else {
+ // means that failed to acquire lock within the specific time
+ return null;
}
}
@@ -2209,25 +2356,19 @@ public class DataRegion implements IDataRegionForQuery {
Map<IDeviceID, DeviceContext> devicePathsToContext,
QueryContext context,
Filter globalTimeFilter,
- boolean isSeq)
- throws MetadataException {
+ boolean isSeq) {
List<IFileScanHandle> fileScanHandles = new ArrayList<>();
for (TsFileResource tsFileResource : tsFileResources) {
if (!tsFileResource.isSatisfied(null, globalTimeFilter, isSeq,
context.isDebug())) {
continue;
}
- closeQueryLock.readLock().lock();
- try {
- if (tsFileResource.isClosed()) {
- fileScanHandles.add(new ClosedFileScanHandleImpl(tsFileResource,
context));
- } else {
- tsFileResource
- .getProcessor()
- .queryForDeviceRegionScan(devicePathsToContext, context,
fileScanHandles);
- }
- } finally {
- closeQueryLock.readLock().unlock();
+ if (tsFileResource.isClosed()) {
+ fileScanHandles.add(new ClosedFileScanHandleImpl(tsFileResource,
context));
+ } else {
+ tsFileResource
+ .getProcessor()
+ .queryForDeviceRegionScanWithoutLock(devicePathsToContext,
context, fileScanHandles);
}
}
return fileScanHandles;
@@ -2235,11 +2376,45 @@ public class DataRegion implements IDataRegionForQuery {
/** lock the read lock of the insert lock */
@Override
- public void readLock() {
- // apply read lock for SG insert lock to prevent inconsistent with
concurrently writing memtable
- insertLock.readLock().lock();
+ public boolean tryReadLock(long waitMillis) {
+ try {
+ // apply read lock for SG insert lock to prevent inconsistent with
concurrently writing
+ // memtable
+ long startTime = System.nanoTime();
+ if (insertLock.readLock().tryLock(waitMillis, TimeUnit.MILLISECONDS)) {
+ // minus already consumed time
+ waitMillis -= (System.nanoTime() - startTime) / 1_000_000;
+ // no remaining time slice
+ if (waitMillis <= 0) {
+ insertLock.readLock().unlock();
+ return false;
+ }
+ return tryGetTsFileManagerReadLock(waitMillis);
+ } else {
+ return false;
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+
+ private boolean tryGetTsFileManagerReadLock(long waitMillis) {
// apply read lock for TsFileResource list
- tsFileManager.readLock();
+ try {
+ if (tsFileManager.tryReadLock(waitMillis)) {
+ return true;
+ } else {
+ // failed to acquire tsFileManager read lock, we also need to unlock
the insertLock
+ insertLock.readLock().unlock();
+ return false;
+ }
+ } catch (InterruptedException e) {
+ // failed to acquire tsFileManager read lock, we also need to unlock the
insertLock
+ insertLock.readLock().unlock();
+ Thread.currentThread().interrupt();
+ return false;
+ }
}
/** unlock the read lock of insert lock */
@@ -2274,34 +2449,22 @@ public class DataRegion implements IDataRegionForQuery {
boolean isSeq)
throws MetadataException {
- if (context.isDebug()) {
- DEBUG_LOGGER.info(
- "Path: {}, get tsfile list: {} isSeq: {} time filter: {}",
- pathList,
- tsFileResources,
- isSeq,
- (globalTimeFilter == null ? "null" : globalTimeFilter));
- }
-
List<TsFileResource> tsfileResourcesForQuery = new ArrayList<>();
for (TsFileResource tsFileResource : tsFileResources) {
if (!tsFileResource.isSatisfied(singleDeviceId, globalTimeFilter, isSeq,
context.isDebug())) {
continue;
}
- closeQueryLock.readLock().lock();
try {
if (tsFileResource.isClosed()) {
tsfileResourcesForQuery.add(tsFileResource);
} else {
tsFileResource
.getProcessor()
- .query(pathList, context, tsfileResourcesForQuery,
globalTimeFilter);
+ .queryWithoutLock(pathList, context, tsfileResourcesForQuery,
globalTimeFilter);
}
} catch (IOException e) {
throw new MetadataException(e);
- } finally {
- closeQueryLock.readLock().unlock();
}
}
return tsfileResourcesForQuery;
@@ -2657,17 +2820,17 @@ public class DataRegion implements IDataRegionForQuery {
if (tsFileProcessor == null) {
sealedTsFiles.add(tsFileResource);
} else {
- tsFileProcessor.getFlushQueryLock().writeLock().lock();
+ tsFileProcessor.writeLock();
if (tsFileResource.isClosed()) {
sealedTsFiles.add(tsFileResource);
- tsFileProcessor.getFlushQueryLock().writeLock().unlock();
+ tsFileProcessor.writeUnlock();
} else {
try {
if (!tsFileProcessor.deleteDataInMemory(deletion)) {
sealedTsFiles.add(tsFileResource);
} // else do nothing
} finally {
- tsFileProcessor.getFlushQueryLock().writeLock().unlock();
+ tsFileProcessor.writeUnlock();
}
}
}
@@ -2819,9 +2982,9 @@ public class DataRegion implements IDataRegionForQuery {
isValidateTsFileFailed =
!TsFileValidator.getInstance().validateTsFile(tsFileProcessor.getTsFileResource());
}
- closeQueryLock.writeLock().lock();
+ tsFileProcessor.writeLock();
try {
- tsFileProcessor.close();
+ tsFileProcessor.closeWithoutSettingResourceStatus();
if (isEmptyFile) {
tsFileProcessor.getTsFileResource().remove();
} else if (isValidateTsFileFailed) {
@@ -2830,10 +2993,11 @@ public class DataRegion implements IDataRegionForQuery {
renameAndHandleError(
tsFilePath + RESOURCE_SUFFIX, tsFilePath + RESOURCE_SUFFIX +
BROKEN_SUFFIX);
} else {
+
tsFileProcessor.getTsFileResource().setStatus(TsFileResourceStatus.NORMAL);
tsFileResourceManager.registerSealedTsFileResource(tsFileProcessor.getTsFileResource());
}
} finally {
- closeQueryLock.writeLock().unlock();
+ tsFileProcessor.writeUnlock();
}
if (isEmptyFile || isValidateTsFileFailed) {
tsFileManager.remove(tsFileProcessor.getTsFileResource(),
tsFileProcessor.isSequence());
@@ -3607,6 +3771,7 @@ public class DataRegion implements IDataRegionForQuery {
return tsFileManager.getTsFileList(false);
}
+ @Override
public String getDataRegionId() {
return dataRegionId;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/IDataRegionForQuery.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/IDataRegionForQuery.java
index c94a62c8566..108c75d50ad 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/IDataRegionForQuery.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/IDataRegionForQuery.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.storageengine.dataregion;
import org.apache.iotdb.commons.path.IFullPath;
+import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.queryengine.common.DeviceContext;
import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
@@ -34,12 +35,12 @@ import java.util.Map;
/** It's an interface that storage engine must provide for query engine */
public interface IDataRegionForQuery {
- /** lock the read lock for thread-safe */
- void readLock();
+ boolean tryReadLock(long waitMillis);
void readUnlock();
/** Get satisfied QueryDataSource from DataRegion for seriesScan */
+ @TestOnly
QueryDataSource query(
List<IFullPath> pathList,
IDeviceID singleDeviceId,
@@ -48,21 +49,37 @@ public interface IDataRegionForQuery {
List<Long> timePartitions)
throws QueryProcessException;
+ /**
+ * Get satisfied QueryDataSource from DataRegion for seriesScan
+ *
+ * @return null means that failed to acquire lock within the specific time
+ */
+ QueryDataSource query(
+ List<IFullPath> pathList,
+ IDeviceID singleDeviceId,
+ QueryContext context,
+ Filter globalTimeFilter,
+ List<Long> timePartitions,
+ long waitForLockTimeInMs)
+ throws QueryProcessException;
+
/** Get satisfied QueryDataSource from DataRegion for regionScan */
IQueryDataSource queryForDeviceRegionScan(
Map<IDeviceID, DeviceContext> devicePathsToContext,
QueryContext queryContext,
Filter globalTimeFilter,
- List<Long> timePartitions)
- throws QueryProcessException;
+ List<Long> timePartitions,
+ long waitForLockTimeInMs);
IQueryDataSource queryForSeriesRegionScan(
List<IFullPath> pathList,
QueryContext queryContext,
Filter globalTimeFilter,
- List<Long> timePartitions)
- throws QueryProcessException;
+ List<Long> timePartitions,
+ long waitForLockTimeInMs);
/** Get database name of this DataRegion */
String getDatabaseName();
+
+ String getDataRegionId();
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/VirtualDataRegion.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/VirtualDataRegion.java
index a104ee32e68..5194d99b71d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/VirtualDataRegion.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/VirtualDataRegion.java
@@ -41,9 +41,14 @@ public class VirtualDataRegion implements
IDataRegionForQuery {
private static final String VIRTUAL_DB_NAME = "root.__virtual";
+ private static final String VIRTUAL_DATA_REGION_ID = "virtual_data_region";
+
public static final QueryDataSource EMPTY_QUERY_DATA_SOURCE =
new QueryDataSource(Collections.emptyList(), Collections.emptyList());
+ public static final QueryDataSource UNFINISHED_QUERY_DATA_SOURCE =
+ new QueryDataSource(Collections.emptyList(), Collections.emptyList());
+
private static final QueryDataSourceForRegionScan
EMPTY_REGION_QUERY_DATA_SOURCE =
new QueryDataSourceForRegionScan(Collections.emptyList(),
Collections.emptyList());
@@ -52,8 +57,9 @@ public class VirtualDataRegion implements IDataRegionForQuery
{
}
@Override
- public void readLock() {
- // do nothing, because itself is thread-safe already
+ public boolean tryReadLock(long waitMillis) {
+ // do nothing, always return true
+ return true;
}
@Override
@@ -69,6 +75,19 @@ public class VirtualDataRegion implements
IDataRegionForQuery {
Filter globalTimeFilter,
List<Long> timePartitions)
throws QueryProcessException {
+ return query(
+ pathList, singleDeviceId, context, globalTimeFilter, timePartitions,
Long.MAX_VALUE);
+ }
+
+ @Override
+ public QueryDataSource query(
+ List<IFullPath> pathList,
+ IDeviceID singleDeviceId,
+ QueryContext context,
+ Filter globalTimeFilter,
+ List<Long> timePartitions,
+ long waitForLockTimeInMs)
+ throws QueryProcessException {
return EMPTY_QUERY_DATA_SOURCE;
}
@@ -77,8 +96,8 @@ public class VirtualDataRegion implements IDataRegionForQuery
{
Map<IDeviceID, DeviceContext> devicePathsToContext,
QueryContext queryContext,
Filter globalTimeFilter,
- List<Long> timePartitions)
- throws QueryProcessException {
+ List<Long> timePartitions,
+ long waitForLockTimeInMs) {
return EMPTY_REGION_QUERY_DATA_SOURCE;
}
@@ -87,8 +106,8 @@ public class VirtualDataRegion implements
IDataRegionForQuery {
List<IFullPath> pathList,
QueryContext queryContext,
Filter globalTimeFilter,
- List<Long> timePartitions)
- throws QueryProcessException {
+ List<Long> timePartitions,
+ long waitForLockTimeInMs) {
return EMPTY_REGION_QUERY_DATA_SOURCE;
}
@@ -97,6 +116,11 @@ public class VirtualDataRegion implements
IDataRegionForQuery {
return VIRTUAL_DB_NAME;
}
+ @Override
+ public String getDataRegionId() {
+ return VIRTUAL_DATA_REGION_ID;
+ }
+
private static class InstanceHolder {
private InstanceHolder() {}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java
index ac71e3da6bb..268a9213b3c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java
@@ -33,12 +33,16 @@ import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BitMap;
import org.apache.tsfile.write.chunk.IChunkWriter;
import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
public abstract class AbstractWritableMemChunk implements IWritableMemChunk {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(AbstractWritableMemChunk.class);
+
protected static long RETRY_INTERVAL_MS = 100L;
protected static long MAX_WAIT_QUERY_MS = 60 * 1000L;
@@ -54,11 +58,18 @@ public abstract class AbstractWritableMemChunk implements
IWritableMemChunk {
protected void maybeReleaseTvList(TVList tvList) {
long startTimeInMs = System.currentTimeMillis();
boolean succeed = false;
+ int retryCount = 0;
while (!succeed) {
try {
tryReleaseTvList(tvList);
succeed = true;
} catch (MemoryNotEnoughException ex) {
+ // print log every 5 seconds
+ if (retryCount % 50 == 0) {
+ LOGGER.warn(
+ "Failed to transfer tvlist memory owner to query engine, {}",
ex.getMessage());
+ }
+ retryCount++;
long waitQueryInMs = System.currentTimeMillis() - startTimeInMs;
if (waitQueryInMs > MAX_WAIT_QUERY_MS) {
// Abort first query in the list. When all queries in the list have
been aborted,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index e90fc264e79..fc1f8190de8 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -1788,10 +1788,10 @@ public class TsFileProcessor {
}
/** Close this tsfile */
- public void close() throws TsFileProcessorException {
+ public void closeWithoutSettingResourceStatus() throws
TsFileProcessorException {
try {
// When closing resource file, its corresponding mod file is also closed.
- tsFileResource.close();
+ tsFileResource.closeWithoutSettingStatus();
} catch (IOException e) {
throw new TsFileProcessorException(e);
}
@@ -1966,7 +1966,7 @@ public class TsFileProcessor {
return new ArrayList<>(alignedChunkMetadataForOneDevice);
}
- public void queryForSeriesRegionScan(
+ public void queryForSeriesRegionScanWithoutLock(
List<IFullPath> pathList,
QueryContext queryContext,
List<IFileScanHandle> fileScanHandlesForQuery) {
@@ -1975,13 +1975,11 @@ public class TsFileProcessor {
Map<IDeviceID, Map<String, List<IChunkHandle>>>
deviceToMemChunkHandleMap = new HashMap<>();
Map<IDeviceID, Map<String, List<IChunkMetadata>>>
deviceToChunkMetadataListMap =
new HashMap<>();
- flushQueryLock.readLock().lock();
try {
for (IFullPath seriesPath : pathList) {
Map<String, List<IChunkMetadata>> measurementToChunkMetaList = new
HashMap<>();
Map<String, List<IChunkHandle>> measurementToChunkHandleList = new
HashMap<>();
- // TODO Tien change the way
long timeLowerBound =
getQueryTimeLowerBound(seriesPath.getDeviceId());
for (IMemTable flushingMemTable : flushingMemTables) {
if (flushingMemTable.isSignalMemTable()) {
@@ -2032,9 +2030,6 @@ public class TsFileProcessor {
QUERY_RESOURCE_METRICS.recordQueryResourceNum(FLUSHING_MEMTABLE,
flushingMemTables.size());
QUERY_RESOURCE_METRICS.recordQueryResourceNum(
WORKING_MEMTABLE, workMemTable != null ? 1 : 0);
-
- flushQueryLock.readLock().unlock();
- logFlushQueryReadUnlocked();
}
if (!deviceToMemChunkHandleMap.isEmpty() ||
!deviceToChunkMetadataListMap.isEmpty()) {
fileScanHandlesForQuery.add(
@@ -2051,7 +2046,7 @@ public class TsFileProcessor {
* Construct IFileScanHandle for data in memtable and the other ones in
flushing memtables. Then
* get the related ChunkMetadata of data on disk.
*/
- public void queryForDeviceRegionScan(
+ public void queryForDeviceRegionScanWithoutLock(
Map<IDeviceID, DeviceContext> devicePathsToContext,
QueryContext queryContext,
List<IFileScanHandle> fileScanHandlesForQuery) {
@@ -2060,7 +2055,6 @@ public class TsFileProcessor {
Map<IDeviceID, Map<String, List<IChunkHandle>>>
deviceToMemChunkHandleMap = new HashMap<>();
Map<IDeviceID, Map<String, List<IChunkMetadata>>>
deviceToChunkMetadataListMap =
new HashMap<>();
- flushQueryLock.readLock().lock();
try {
for (Map.Entry<IDeviceID, DeviceContext> entry :
devicePathsToContext.entrySet()) {
IDeviceID deviceID = entry.getKey();
@@ -2114,9 +2108,6 @@ public class TsFileProcessor {
QUERY_RESOURCE_METRICS.recordQueryResourceNum(FLUSHING_MEMTABLE,
flushingMemTables.size());
QUERY_RESOURCE_METRICS.recordQueryResourceNum(
WORKING_MEMTABLE, workMemTable != null ? 1 : 0);
-
- flushQueryLock.readLock().unlock();
- logFlushQueryReadUnlocked();
}
if (!deviceToMemChunkHandleMap.isEmpty() ||
!deviceToChunkMetadataListMap.isEmpty()) {
@@ -2137,18 +2128,39 @@ public class TsFileProcessor {
*
* @param seriesPaths selected paths
*/
+ public void queryWithoutLock(
+ List<IFullPath> seriesPaths,
+ QueryContext context,
+ List<TsFileResource> tsfileResourcesForQuery,
+ Filter globalTimeFilter)
+ throws IOException {
+ query(seriesPaths, context, tsfileResourcesForQuery, globalTimeFilter,
false);
+ }
+
+ @TestOnly
public void query(
List<IFullPath> seriesPaths,
QueryContext context,
List<TsFileResource> tsfileResourcesForQuery,
Filter globalTimeFilter)
throws IOException {
+ query(seriesPaths, context, tsfileResourcesForQuery, globalTimeFilter,
true);
+ }
+
+ private void query(
+ List<IFullPath> seriesPaths,
+ QueryContext context,
+ List<TsFileResource> tsfileResourcesForQuery,
+ Filter globalTimeFilter,
+ boolean needLock)
+ throws IOException {
long startTime = System.nanoTime();
try {
Map<IFullPath, List<IChunkMetadata>> pathToChunkMetadataListMap = new
HashMap<>();
Map<IFullPath, List<ReadOnlyMemChunk>> pathToReadOnlyMemChunkMap = new
HashMap<>();
-
- flushQueryLock.readLock().lock();
+ if (needLock) {
+ flushQueryLock.readLock().lock();
+ }
try {
for (IFullPath seriesPath : seriesPaths) {
List<ReadOnlyMemChunk> readOnlyMemChunks = new ArrayList<>();
@@ -2193,9 +2205,10 @@ public class TsFileProcessor {
QUERY_RESOURCE_METRICS.recordQueryResourceNum(FLUSHING_MEMTABLE,
flushingMemTables.size());
QUERY_RESOURCE_METRICS.recordQueryResourceNum(
WORKING_MEMTABLE, workMemTable != null ? 1 : 0);
-
- flushQueryLock.readLock().unlock();
- logFlushQueryReadUnlocked();
+ if (needLock) {
+ flushQueryLock.readLock().unlock();
+ logFlushQueryReadUnlocked();
+ }
}
if (!pathToReadOnlyMemChunkMap.isEmpty() ||
!pathToChunkMetadataListMap.isEmpty()) {
@@ -2314,8 +2327,20 @@ public class TsFileProcessor {
getWriter().getSchema().getTableSchemaMap().computeIfAbsent(tableName,
tableSchemaFunction);
}
- public ReadWriteLock getFlushQueryLock() {
- return flushQueryLock;
+ public void writeLock() {
+ flushQueryLock.writeLock().lock();
+ }
+
+ public void writeUnlock() {
+ flushQueryLock.writeLock().unlock();
+ }
+
+ public boolean tryReadLock(long waitInMs) throws InterruptedException {
+ return flushQueryLock.readLock().tryLock(waitInMs, TimeUnit.MILLISECONDS);
+ }
+
+ public void readUnLock() {
+ flushQueryLock.readLock().unlock();
}
private void logFlushQueryWriteLocked() {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileManager.java
index 15c2365ebdb..b7c1ba2c14f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileManager.java
@@ -27,6 +27,7 @@ import
org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeInd
import org.apache.iotdb.db.storageengine.rescon.memory.TsFileResourceManager;
import org.apache.tsfile.read.filter.basic.Filter;
+import org.apache.tsfile.utils.Pair;
import java.io.IOException;
import java.util.ArrayList;
@@ -36,6 +37,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -99,6 +101,41 @@ public class TsFileManager {
}
}
+ /**
+ * don't need to acquire lock again, caller should guarantee the lock has
been acquired
+ *
+ * @return left is seq resource list, right is unSeq resource list
+ */
+ public Pair<List<TsFileResource>, List<TsFileResource>>
getAllTsFileListForQuery(
+ List<Long> timePartitions, Filter timeFilter) {
+ List<TsFileResource> seq = new ArrayList<>();
+ List<TsFileResource> unSeq = new ArrayList<>();
+ if (timePartitions == null) {
+ for (Map.Entry<Long, TsFileResourceList> entry :
sequenceFiles.entrySet()) {
+ if (TimePartitionUtils.satisfyTimePartition(timeFilter,
entry.getKey())) {
+ seq.addAll(entry.getValue().getArrayList());
+ }
+ }
+ for (Map.Entry<Long, TsFileResourceList> entry :
unsequenceFiles.entrySet()) {
+ if (TimePartitionUtils.satisfyTimePartition(timeFilter,
entry.getKey())) {
+ unSeq.addAll(entry.getValue().getArrayList());
+ }
+ }
+ } else {
+ for (Long timePartitionId : timePartitions) {
+ TsFileResourceList tsFileResources =
sequenceFiles.get(timePartitionId);
+ if (tsFileResources != null) {
+ seq.addAll(tsFileResources.getArrayList());
+ }
+ tsFileResources = unsequenceFiles.get(timePartitionId);
+ if (tsFileResources != null) {
+ unSeq.addAll(tsFileResources.getArrayList());
+ }
+ }
+ }
+ return new Pair<>(seq, unSeq);
+ }
+
public List<TsFileResource> getTsFileListSnapshot(long timePartition,
boolean sequence) {
readLock();
try {
@@ -364,12 +401,14 @@ public class TsFileManager {
}
}
- public void getModFileManagement() {}
-
public void readLock() {
resourceListLock.readLock().lock();
}
+ public boolean tryReadLock(long waitMillis) throws InterruptedException {
+ return resourceListLock.readLock().tryLock(waitMillis,
TimeUnit.MILLISECONDS);
+ }
+
public void readUnlock() {
resourceListLock.readLock().unlock();
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/DataDriverTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/DataDriverTest.java
index 68f1f9c61f0..0af5098d4ba 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/DataDriverTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/DataDriverTest.java
@@ -76,6 +76,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.eq;
public class DataDriverTest {
@@ -93,6 +94,7 @@ public class DataDriverTest {
@Before
public void setUp() throws MetadataException, IOException,
WriteProcessException {
+
IoTDBDescriptor.getInstance().getConfig().setDriverTaskExecutionTimeSliceInMs(10000);
SeriesReaderTestUtil.setUp(
measurementSchemas, deviceIds, seqResources, unSeqResources,
DATA_DRIVER_TEST_SG);
}
@@ -120,6 +122,7 @@ public class DataDriverTest {
FragmentInstanceStateMachine stateMachine =
new FragmentInstanceStateMachine(instanceId,
instanceNotificationExecutor);
DataRegion dataRegion = Mockito.mock(DataRegion.class);
+ Mockito.when(dataRegion.tryReadLock(Mockito.anyLong())).thenReturn(true);
FragmentInstanceContext fragmentInstanceContext =
createFragmentInstanceContext(instanceId, stateMachine);
fragmentInstanceContext.setDataRegion(dataRegion);
@@ -180,14 +183,16 @@ public class DataDriverTest {
LimitOperator limitOperator =
new LimitOperator(driverContext.getOperatorContexts().get(3), 250,
timeJoinOperator);
+ fragmentInstanceContext.setSourcePaths(driverContext.getPaths());
String deviceId = DATA_DRIVER_TEST_SG + ".device0";
Mockito.when(
dataRegion.query(
- driverContext.getPaths(),
- IDeviceID.Factory.DEFAULT_FACTORY.create(deviceId),
- fragmentInstanceContext,
- null,
- null))
+ eq(driverContext.getPaths()),
+ eq(IDeviceID.Factory.DEFAULT_FACTORY.create(deviceId)),
+ eq(fragmentInstanceContext),
+ Mockito.isNull(),
+ Mockito.isNull(),
+ Mockito.anyLong()))
.thenReturn(new QueryDataSource(seqResources, unSeqResources));
fragmentInstanceContext.initQueryDataSource(driverContext.getPaths());
fragmentInstanceContext.initializeNumOfDrivers(1);
diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
index 290987d0feb..5be8808db50 100644
--- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
+++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
@@ -737,6 +737,7 @@ struct TFetchFragmentInstanceStatisticsResp {
14: optional i64 blockQueuedTime
15: optional string ip
16: optional string state
+ 17: optional i32 initDataQuerySourceRetryCount
}
/**
