fields = recordOverlap.getFields();
String sql;
for (Field field : fields) {
diff --git a/src/main/java/cn/edu/tsinghua/iotdb/query/engine/AggregateEngine.java b/src/main/java/cn/edu/tsinghua/iotdb/query/engine/AggregateEngine.java
index 10ef62ed9b9..9e48b4575b7 100644
--- a/src/main/java/cn/edu/tsinghua/iotdb/query/engine/AggregateEngine.java
+++ b/src/main/java/cn/edu/tsinghua/iotdb/query/engine/AggregateEngine.java
@@ -1,11 +1,16 @@
package cn.edu.tsinghua.iotdb.query.engine;
+import cn.edu.tsinghua.iotdb.concurrent.IoTDBThreadPoolFactory;
import cn.edu.tsinghua.iotdb.conf.TsfileDBDescriptor;
+import cn.edu.tsinghua.iotdb.engine.filenode.FileNodeManager;
+import cn.edu.tsinghua.iotdb.exception.FileNodeManagerException;
import cn.edu.tsinghua.iotdb.exception.PathErrorException;
import cn.edu.tsinghua.iotdb.metadata.MManager;
import cn.edu.tsinghua.iotdb.query.aggregation.AggregateFunction;
+import cn.edu.tsinghua.iotdb.query.management.FileReaderMap;
import cn.edu.tsinghua.iotdb.query.management.FilterStructure;
import cn.edu.tsinghua.iotdb.query.management.ReadCachePrefix;
+import cn.edu.tsinghua.iotdb.query.management.ReadCacheManager;
import cn.edu.tsinghua.iotdb.query.reader.AggregateRecordReader;
import cn.edu.tsinghua.iotdb.query.reader.QueryRecordReader;
import cn.edu.tsinghua.iotdb.query.reader.ReaderType;
@@ -23,6 +28,8 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
import static cn.edu.tsinghua.iotdb.query.engine.EngineUtils.noFilterOrOnlyHasTimeFilter;
@@ -38,6 +45,19 @@ public class AggregateEngine {
private int crossQueryFetchSize =
10 * TsfileDBDescriptor.getInstance().getConfig().fetchSize;
+ //private ExecutorService aggregateThreadPool;
+// private AggregateEngine() {
+// }
+
+// private static class AggregateEngineHolder {
+// private static final AggregateEngine INSTANCE = new AggregateEngine();
+//
+// }
+//
+// public static final AggregateEngine getInstance() {
+// return AggregateEngineHolder.INSTANCE;
+// }
+
/**
* Public invoking method of multiple aggregation.
*
@@ -222,18 +242,56 @@ public class AggregateEngine {
throws PathErrorException, ProcessorException, IOException {
int aggreNumber = 0;
+ CountDownLatch latch = new CountDownLatch(aggres.size());
+ ExecutorService service = IoTDBThreadPoolFactory.newFixedThreadPool(Runtime.getRuntime().availableProcessors() - 1,
+ "AggregateThread");
+ //ExecutorService service = Executors.newFixedThreadPool(aggres.size());
+
for (Pair pair : aggres) {
aggreNumber++;
Path path = pair.left;
AggregateFunction aggregateFunction = pair.right;
+
String deltaObjectUID = path.getDeltaObjectToString();
String measurementUID = path.getMeasurementToString();
-
AggregateRecordReader recordReader = (AggregateRecordReader)
RecordReaderFactory.getInstance().getRecordReader(deltaObjectUID, measurementUID,
- queryTimeFilter, null, null, ReadCachePrefix.addQueryPrefix(aggreNumber), ReaderType.AGGREGATE);
+ queryTimeFilter, null, null, ReadCachePrefix.addQueryPrefix(aggreNumber), ReaderType.AGGREGATE);
+ service.submit(new AggregateThread(recordReader, aggregateFunction, latch));
+ }
- recordReader.aggregate(aggregateFunction);
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+
+ service.shutdown();
+ }
+
+ private class AggregateThread implements Runnable {
+
+ private AggregateRecordReader aggregateRecordReader;
+ private AggregateFunction aggregateFunction;
+ private CountDownLatch latch;
+
+ public AggregateThread(AggregateRecordReader aggregateRecordReader, AggregateFunction aggregateFunction, CountDownLatch latch) {
+ this.aggregateRecordReader = aggregateRecordReader;
+ this.aggregateFunction = aggregateFunction;
+ this.latch = latch;
+ }
+
+ @Override
+ public void run() {
+ try {
+ aggregateRecordReader .aggregate(aggregateFunction);
+// FileNodeManager.getInstance().endQuery(deltaObjectUID, recordReader.getReadToken());
+// ReadCacheManager.getInstance().removeReadToken(deltaObjectUID, recordReader.getReadToken());
+ FileReaderMap.getInstance().close();
+ latch.countDown();
+ } catch (ProcessorException | IOException e) {
+ e.printStackTrace();
+ }
}
}
@@ -244,8 +302,8 @@ public class AggregateEngine {
* once for querying d1.s1, once for querying d2.s1.
*
* When this method is invoked, need add the filter index as a new parameter, for the reason of exist of
- * RecordReaderCache, if the composition of CrossFilterExpression exist same SingleFilterExpression,
- * we must guarantee that the RecordReaderCache doesn't cause conflict to the same SingleFilterExpression.
+ * RecordReaderCacheManager, if the composition of CrossFilterExpression exist same SingleFilterExpression,
+ * we must guarantee that the RecordReaderCacheManager doesn't cause conflict to the same SingleFilterExpression.
*/
private DynamicOneColumnData getDataUseSingleValueFilter(SingleSeriesFilterExpression queryValueFilter,
DynamicOneColumnData res, int fetchSize, int valueFilterNumber)
diff --git a/src/main/java/cn/edu/tsinghua/iotdb/query/engine/OverflowQueryEngine.java b/src/main/java/cn/edu/tsinghua/iotdb/query/engine/OverflowQueryEngine.java
index f5f10bc2ec7..f0b513727cc 100644
--- a/src/main/java/cn/edu/tsinghua/iotdb/query/engine/OverflowQueryEngine.java
+++ b/src/main/java/cn/edu/tsinghua/iotdb/query/engine/OverflowQueryEngine.java
@@ -11,7 +11,7 @@ import cn.edu.tsinghua.iotdb.query.fill.LinearFill;
import cn.edu.tsinghua.iotdb.query.fill.PreviousFill;
import cn.edu.tsinghua.iotdb.query.management.FilterStructure;
import cn.edu.tsinghua.iotdb.query.management.ReadCachePrefix;
-import cn.edu.tsinghua.iotdb.query.management.ReadLockManager;
+import cn.edu.tsinghua.iotdb.query.management.ReadCacheManager;
import cn.edu.tsinghua.iotdb.query.reader.QueryRecordReader;
import cn.edu.tsinghua.iotdb.query.reader.ReaderType;
import cn.edu.tsinghua.iotdb.query.reader.RecordReaderFactory;
@@ -110,16 +110,15 @@ public class OverflowQueryEngine {
aggregations.add(new Pair<>(pair.left, aggregateFunction));
}
- AggregateEngine aggregateEngine = new AggregateEngine();
- aggregateEngine.multiAggregate(aggregations, filterStructures);
+ new AggregateEngine().multiAggregate(aggregations, filterStructures);
OnePassQueryDataSet ansOnePassQueryDataSet = new OnePassQueryDataSet();
for (Pair pair : aggregations) {
AggregateFunction aggregateFunction = pair.right;
if (aggregateFunction.resultData.timeLength == 0) {
aggregateFunction.putDefaultValue();
}
- LOGGER.debug(String.format("key %s, data length %s, empty data length %s", EngineUtils.aggregationKey(aggregateFunction, pair.left),
- aggregateFunction.resultData.timeLength, aggregateFunction.resultData.emptyTimeLength));
+ LOGGER.debug(String.format("key %s, data time length %s, data value length %s, empty data time length %s", EngineUtils.aggregationKey(aggregateFunction, pair.left),
+ aggregateFunction.resultData.timeLength, aggregateFunction.resultData.valueLength, aggregateFunction.resultData.emptyTimeLength));
ansOnePassQueryDataSet.mapRet.put(EngineUtils.aggregationKey(aggregateFunction, pair.left), aggregateFunction.resultData);
}
// aggregateThreadLocal.set(ansOnePassQueryDataSet);
@@ -140,9 +139,9 @@ public class OverflowQueryEngine {
public OnePassQueryDataSet groupBy(List> aggres, List filterStructures,
long unit, long origin, List> intervals, int fetchSize) {
- ThreadLocal groupByCalcTime = ReadLockManager.getInstance().getGroupByCalcTime();
- ThreadLocal groupByEngineNoFilterLocal = ReadLockManager.getInstance().getGroupByEngineNoFilterLocal();
- ThreadLocal groupByEngineWithFilterLocal = ReadLockManager.getInstance().getGroupByEngineWithFilterLocal();
+ ThreadLocal groupByCalcTime = ReadCacheManager.getInstance().getGroupByCalcTime();
+ ThreadLocal groupByEngineNoFilterLocal = ReadCacheManager.getInstance().getGroupByEngineNoFilterLocal();
+ ThreadLocal groupByEngineWithFilterLocal = ReadCacheManager.getInstance().getGroupByEngineWithFilterLocal();
if (groupByCalcTime.get() == null) {
@@ -419,8 +418,8 @@ public class OverflowQueryEngine {
* once for querying d1.s1, once for querying d2.s1.
*
* When this method is invoked, need add the filter index as a new parameter, for the reason of exist of
- * RecordReaderCache, if the composition of CrossFilterExpression exist same SingleFilterExpression,
- * we must guarantee that the RecordReaderCache doesn't cause conflict to the same SingleFilterExpression.
+ * RecordReaderCacheManager, if the composition of CrossFilterExpression exist same SingleFilterExpression,
+ * we must guarantee that the RecordReaderCacheManager doesn't cause conflict to the same SingleFilterExpression.
*/
private DynamicOneColumnData querySeriesForCross(SingleSeriesFilterExpression queryValueFilter,
DynamicOneColumnData res, int fetchSize, int valueFilterNumber)
diff --git a/src/main/java/cn/edu/tsinghua/iotdb/query/engine/groupby/GroupByEngineWithFilter.java b/src/main/java/cn/edu/tsinghua/iotdb/query/engine/groupby/GroupByEngineWithFilter.java
index 5e8363d4603..3195b5a451d 100644
--- a/src/main/java/cn/edu/tsinghua/iotdb/query/engine/groupby/GroupByEngineWithFilter.java
+++ b/src/main/java/cn/edu/tsinghua/iotdb/query/engine/groupby/GroupByEngineWithFilter.java
@@ -381,8 +381,8 @@ public class GroupByEngineWithFilter {
* once for querying d1.s1, once for querying d2.s1.
*
* When this method is invoked, need add the filter index as a new parameter, for the reason of exist of
- * RecordReaderCache, if the composition of CrossFilterExpression exist same SingleFilterExpression,
- * we must guarantee that the RecordReaderCache doesn't cause conflict to the same SingleFilterExpression.
+ * RecordReaderCacheManager, if the composition of CrossFilterExpression exist same SingleFilterExpression,
+ * we must guarantee that the RecordReaderCacheManager doesn't cause conflict to the same SingleFilterExpression.
*/
private static DynamicOneColumnData getDataUseSingleValueFilter(SingleSeriesFilterExpression queryValueFilter,
DynamicOneColumnData res, int fetchSize, int valueFilterNumber)
diff --git a/src/main/java/cn/edu/tsinghua/iotdb/query/management/FileReaderMap.java b/src/main/java/cn/edu/tsinghua/iotdb/query/management/FileReaderMap.java
index 1ed20ed5bc6..7c29829ea0a 100644
--- a/src/main/java/cn/edu/tsinghua/iotdb/query/management/FileReaderMap.java
+++ b/src/main/java/cn/edu/tsinghua/iotdb/query/management/FileReaderMap.java
@@ -11,12 +11,12 @@ public class FileReaderMap {
/** map to store opened file stream **/
private static ThreadLocal> fileReaderMap = new ThreadLocal<>();
- private static class ReaderHolder {
+ private static class FileReaderMapHolder {
private static final FileReaderMap INSTANCE = new FileReaderMap();
}
public static FileReaderMap getInstance() {
- return ReaderHolder.INSTANCE;
+ return FileReaderMapHolder.INSTANCE;
}
public TsRandomAccessLocalFileReader get(String path) throws IOException {
diff --git a/src/main/java/cn/edu/tsinghua/iotdb/query/management/ReadLockManager.java b/src/main/java/cn/edu/tsinghua/iotdb/query/management/ReadCacheManager.java
similarity index 82%
rename from src/main/java/cn/edu/tsinghua/iotdb/query/management/ReadLockManager.java
rename to src/main/java/cn/edu/tsinghua/iotdb/query/management/ReadCacheManager.java
index 06c6e078144..a355574e25a 100644
--- a/src/main/java/cn/edu/tsinghua/iotdb/query/management/ReadLockManager.java
+++ b/src/main/java/cn/edu/tsinghua/iotdb/query/management/ReadCacheManager.java
@@ -2,6 +2,7 @@ package cn.edu.tsinghua.iotdb.query.management;
import java.io.IOException;
import java.util.HashMap;
+import java.util.Map;
import cn.edu.tsinghua.iotdb.engine.filenode.FileNodeManager;
import cn.edu.tsinghua.iotdb.exception.FileNodeManagerException;
@@ -11,23 +12,33 @@ import cn.edu.tsinghua.tsfile.common.exception.ProcessorException;
/**
*
- * Read process lock manager and ThreadLocal variable manager.
+ * Read lock manager, ThreadLocal variable and RecordReaderCacheManager manager.
* When a query process is over or quit abnormally, the unlockForOneRequest method will
* be invoked to clear the thread level variable.
*
*
*/
-public class ReadLockManager {
+public class ReadCacheManager {
+
+ private static final class ReadCacheManagerHolder {
+ private static final ReadCacheManager INSTANCE = new ReadCacheManager();
+ }
+
+ private ReadCacheManager() {
+ }
+
+ public static ReadCacheManager getInstance() {
+ return ReadCacheManagerHolder.INSTANCE;
+ }
- private static ReadLockManager instance = new ReadLockManager();
private FileNodeManager fileNodeManager = FileNodeManager.getInstance();
/** storage deltaObjectId and its read lock **/
private ThreadLocal> locksMap = new ThreadLocal<>();
- /** this is no need to set as ThreadLocal, RecordReaderCache has ThreadLocal variable**/
- public RecordReaderCache recordReaderCache = new RecordReaderCache();
+ /** this is no need to set as ThreadLocal, RecordReaderCacheManager has ThreadLocal variable**/
+ private RecordReaderCacheManager recordReaderCacheManager = new RecordReaderCacheManager();
/** represents the execute time of group by method**/
private ThreadLocal groupByCalcTime;
@@ -38,9 +49,6 @@ public class ReadLockManager {
/** ThreadLocal, due to the usage of OverflowQPExecutor **/
private ThreadLocal groupByEngineWithFilterLocal;
- private ReadLockManager() {
- }
-
public int lock(String deltaObjectUID) throws ProcessorException {
checkLocksMap();
int token;
@@ -58,6 +66,10 @@ public class ReadLockManager {
return token;
}
+ public void removeReadToken(String deltaObjectId, int readToken) {
+ locksMap.get().remove(deltaObjectId, readToken);
+ }
+
/**
* When jdbc connection is closed normally or quit abnormally, this method should be invoked.
* All read cache in this request should be released.
@@ -68,7 +80,8 @@ public class ReadLockManager {
if (locksMap.get() == null) {
return;
}
- HashMap locks = locksMap.get();
+
+ Map locks = locksMap.get();
for (String key : locks.keySet()) {
unlockForQuery(key, locks.get(key));
}
@@ -85,7 +98,7 @@ public class ReadLockManager {
groupByEngineWithFilterLocal.remove();
}
- recordReaderCache.clear();
+ recordReaderCacheManager.clear();
FileReaderMap.getInstance().close();
}
@@ -98,10 +111,6 @@ public class ReadLockManager {
}
}
- public static ReadLockManager getInstance() {
- return instance;
- }
-
private void checkLocksMap() {
if (locksMap.get() == null) {
locksMap.set(new HashMap<>());
@@ -115,10 +124,6 @@ public class ReadLockManager {
return this.groupByCalcTime;
}
- public void setGroupByCalcTime(ThreadLocal t) {
- this.groupByCalcTime = t;
- }
-
public ThreadLocal getGroupByEngineNoFilterLocal() {
if (groupByEngineNoFilterLocal == null) {
groupByEngineNoFilterLocal = new ThreadLocal<>();
@@ -126,10 +131,6 @@ public class ReadLockManager {
return this.groupByEngineNoFilterLocal;
}
- public void setGroupByEngineNoFilterLocal(ThreadLocal t) {
- this.groupByEngineNoFilterLocal = t;
- }
-
public ThreadLocal getGroupByEngineWithFilterLocal() {
if (groupByEngineWithFilterLocal == null) {
groupByEngineWithFilterLocal = new ThreadLocal<>();
@@ -137,8 +138,7 @@ public class ReadLockManager {
return this.groupByEngineWithFilterLocal;
}
- public void setGroupByEngineWithFilterLocal(ThreadLocal t) {
- this.groupByEngineWithFilterLocal = t;
+ public RecordReaderCacheManager getRecordReaderCacheManager() {
+ return recordReaderCacheManager;
}
-
}
diff --git a/src/main/java/cn/edu/tsinghua/iotdb/query/management/RecordReaderCache.java b/src/main/java/cn/edu/tsinghua/iotdb/query/management/RecordReaderCacheManager.java
similarity index 97%
rename from src/main/java/cn/edu/tsinghua/iotdb/query/management/RecordReaderCache.java
rename to src/main/java/cn/edu/tsinghua/iotdb/query/management/RecordReaderCacheManager.java
index efbd1ad65cc..319380ef9db 100644
--- a/src/main/java/cn/edu/tsinghua/iotdb/query/management/RecordReaderCache.java
+++ b/src/main/java/cn/edu/tsinghua/iotdb/query/management/RecordReaderCacheManager.java
@@ -8,7 +8,7 @@ import cn.edu.tsinghua.iotdb.query.reader.RecordReader;
/**
* Used for read process, put the query structure in the cache for one query process.
*/
-public class RecordReaderCache {
+public class RecordReaderCacheManager {
private ThreadLocal> cache = new ThreadLocal<>();
diff --git a/src/main/java/cn/edu/tsinghua/iotdb/query/reader/AggregateRecordReader.java b/src/main/java/cn/edu/tsinghua/iotdb/query/reader/AggregateRecordReader.java
index 48c260b4bc7..d4830678a7f 100644
--- a/src/main/java/cn/edu/tsinghua/iotdb/query/reader/AggregateRecordReader.java
+++ b/src/main/java/cn/edu/tsinghua/iotdb/query/reader/AggregateRecordReader.java
@@ -34,10 +34,10 @@ public class AggregateRecordReader extends RecordReader {
public AggregateRecordReader(GlobalSortedSeriesDataSource globalSortedSeriesDataSource, OverflowSeriesDataSource overflowSeriesDataSource,
String deltaObjectId, String measurementId,
- SingleSeriesFilterExpression queryTimeFilter, SingleSeriesFilterExpression queryValueFilter)
+ SingleSeriesFilterExpression queryTimeFilter, SingleSeriesFilterExpression queryValueFilter, int readToken)
throws PathErrorException, IOException {
super(globalSortedSeriesDataSource, overflowSeriesDataSource, deltaObjectId, measurementId,
- queryTimeFilter, queryValueFilter);
+ queryTimeFilter, queryValueFilter, readToken);
}
/**
diff --git a/src/main/java/cn/edu/tsinghua/iotdb/query/reader/FillRecordReader.java b/src/main/java/cn/edu/tsinghua/iotdb/query/reader/FillRecordReader.java
index 904041304ad..8ab2ee6f44d 100644
--- a/src/main/java/cn/edu/tsinghua/iotdb/query/reader/FillRecordReader.java
+++ b/src/main/java/cn/edu/tsinghua/iotdb/query/reader/FillRecordReader.java
@@ -17,9 +17,9 @@ public class FillRecordReader extends RecordReader{
public FillRecordReader(GlobalSortedSeriesDataSource globalSortedSeriesDataSource, OverflowSeriesDataSource overflowSeriesDataSource,
String deltaObjectId, String measurementId,
- SingleSeriesFilterExpression queryTimeFilter, SingleSeriesFilterExpression queryValueFilter)
+ SingleSeriesFilterExpression queryTimeFilter, SingleSeriesFilterExpression queryValueFilter, int readToken)
throws PathErrorException, IOException {
- super(globalSortedSeriesDataSource, overflowSeriesDataSource, deltaObjectId, measurementId, queryTimeFilter, queryValueFilter);
+ super(globalSortedSeriesDataSource, overflowSeriesDataSource, deltaObjectId, measurementId, queryTimeFilter, queryValueFilter, readToken);
}
/**
diff --git a/src/main/java/cn/edu/tsinghua/iotdb/query/reader/QueryRecordReader.java b/src/main/java/cn/edu/tsinghua/iotdb/query/reader/QueryRecordReader.java
index 212c8359eda..547a8adb782 100644
--- a/src/main/java/cn/edu/tsinghua/iotdb/query/reader/QueryRecordReader.java
+++ b/src/main/java/cn/edu/tsinghua/iotdb/query/reader/QueryRecordReader.java
@@ -33,9 +33,9 @@ public class QueryRecordReader extends RecordReader {
public QueryRecordReader(GlobalSortedSeriesDataSource globalSortedSeriesDataSource, OverflowSeriesDataSource overflowSeriesDataSource,
String deltaObjectId, String measurementId,
- SingleSeriesFilterExpression queryTimeFilter, SingleSeriesFilterExpression queryValueFilter)
+ SingleSeriesFilterExpression queryTimeFilter, SingleSeriesFilterExpression queryValueFilter, int readToken)
throws PathErrorException, IOException {
- super(globalSortedSeriesDataSource, overflowSeriesDataSource, deltaObjectId, measurementId, queryTimeFilter, queryValueFilter);
+ super(globalSortedSeriesDataSource, overflowSeriesDataSource, deltaObjectId, measurementId, queryTimeFilter, queryValueFilter, readToken);
overflowOperationReaderCopy = overflowOperationReader.copy();
diff --git a/src/main/java/cn/edu/tsinghua/iotdb/query/reader/RecordReader.java b/src/main/java/cn/edu/tsinghua/iotdb/query/reader/RecordReader.java
index 54b74745c16..7e5998c2b36 100644
--- a/src/main/java/cn/edu/tsinghua/iotdb/query/reader/RecordReader.java
+++ b/src/main/java/cn/edu/tsinghua/iotdb/query/reader/RecordReader.java
@@ -81,9 +81,11 @@ public class RecordReader {
/** memRawSeriesChunk + overflowSeriesInsertReader + overflowOperationReader **/
protected InsertDynamicData insertMemoryData;
+ protected int readToken;
+
public RecordReader(GlobalSortedSeriesDataSource globalSortedSeriesDataSource, OverflowSeriesDataSource overflowSeriesDataSource,
String deltaObjectId, String measurementId,
- SingleSeriesFilterExpression queryTimeFilter, SingleSeriesFilterExpression queryValueFilter)
+ SingleSeriesFilterExpression queryTimeFilter, SingleSeriesFilterExpression queryValueFilter, int readToken)
throws PathErrorException, IOException {
List sealedFilePathList = new ArrayList<>();
@@ -130,6 +132,8 @@ public class RecordReader {
if (queryValueFilter != null) {
singleValueVisitor = getSingleValueVisitorByDataType(dataType, queryValueFilter);
}
+
+ this.readToken = readToken;
}
public void closeFileStream() {
@@ -144,4 +148,8 @@ public class RecordReader {
public InsertDynamicData getInsertMemoryData() {
return this.insertMemoryData;
}
+
+ public int getReadToken() {
+ return this.readToken;
+ }
}
diff --git a/src/main/java/cn/edu/tsinghua/iotdb/query/reader/RecordReaderFactory.java b/src/main/java/cn/edu/tsinghua/iotdb/query/reader/RecordReaderFactory.java
index 6ff29574de7..14a611fdb96 100644
--- a/src/main/java/cn/edu/tsinghua/iotdb/query/reader/RecordReaderFactory.java
+++ b/src/main/java/cn/edu/tsinghua/iotdb/query/reader/RecordReaderFactory.java
@@ -4,7 +4,7 @@ import cn.edu.tsinghua.iotdb.engine.filenode.FileNodeManager;
import cn.edu.tsinghua.iotdb.engine.querycontext.QueryDataSource;
import cn.edu.tsinghua.iotdb.exception.FileNodeManagerException;
import cn.edu.tsinghua.iotdb.exception.PathErrorException;
-import cn.edu.tsinghua.iotdb.query.management.ReadLockManager;
+import cn.edu.tsinghua.iotdb.query.management.ReadCacheManager;
import cn.edu.tsinghua.tsfile.common.exception.ProcessorException;
import cn.edu.tsinghua.tsfile.timeseries.filter.definition.SingleSeriesFilterExpression;
import cn.edu.tsinghua.tsfile.timeseries.filterV2.expression.impl.QueryFilterFactory;
@@ -27,11 +27,11 @@ public class RecordReaderFactory {
private static RecordReaderFactory instance = new RecordReaderFactory();
private FileNodeManager fileNodeManager;
- private ReadLockManager readLockManager;
+ private ReadCacheManager readCacheManager;
private RecordReaderFactory() {
fileNodeManager = FileNodeManager.getInstance();
- readLockManager = ReadLockManager.getInstance();
+ readCacheManager = ReadCacheManager.getInstance();
}
/**
@@ -39,23 +39,23 @@ public class RecordReaderFactory {
*
* @param readLock if readLock is not null, the read lock of file node has been created,
* else a new read lock token should be applied.
- * @param prefix for the exist of RecordReaderCache and batch read, we need a prefix to
+ * @param prefix for the exist of RecordReaderCacheManager and batch read, we need a prefix to
* represent the uniqueness.
* @return RecordReader
*/
- public RecordReader getRecordReader(String deltaObjectUID, String measurementID,
+ public synchronized RecordReader getRecordReader(String deltaObjectUID, String measurementID,
SingleSeriesFilterExpression timeFilter, SingleSeriesFilterExpression valueFilter,
Integer readLock, String prefix, ReaderType readerType)
throws ProcessorException, PathErrorException, IOException {
- int token = 0;
+ int readToken = 0;
if (readLock == null) {
- token = readLockManager.lock(deltaObjectUID);
+ readToken = readCacheManager.lock(deltaObjectUID);
} else {
- token = readLock;
+ readToken = readLock;
}
String cacheDeltaKey = prefix + deltaObjectUID;
- if (readLockManager.recordReaderCache.containsRecordReader(cacheDeltaKey, measurementID)) {
- return readLockManager.recordReaderCache.get(cacheDeltaKey, measurementID);
+ if (readCacheManager.getRecordReaderCacheManager().containsRecordReader(cacheDeltaKey, measurementID)) {
+ return readCacheManager.getRecordReaderCacheManager().get(cacheDeltaKey, measurementID);
} else {
QueryDataSource queryDataSource;
try {
@@ -65,28 +65,28 @@ public class RecordReaderFactory {
throw new ProcessorException(e.getMessage());
}
- RecordReader recordReader = createANewRecordReader(deltaObjectUID, measurementID, timeFilter, valueFilter, queryDataSource, readerType);
- readLockManager.recordReaderCache.put(cacheDeltaKey, measurementID, recordReader);
+ RecordReader recordReader = createANewRecordReader(deltaObjectUID, measurementID, timeFilter, valueFilter, queryDataSource, readerType, readToken);
+ readCacheManager.getRecordReaderCacheManager().put(cacheDeltaKey, measurementID, recordReader);
return recordReader;
}
}
private RecordReader createANewRecordReader(String deltaObjectUID, String measurementID,
SingleSeriesFilterExpression queryTimeFilter, SingleSeriesFilterExpression queryValueFilter,
- QueryDataSource queryDataSource, ReaderType readerType) throws PathErrorException, IOException {
+ QueryDataSource queryDataSource, ReaderType readerType, int readToken) throws PathErrorException, IOException {
switch (readerType) {
case QUERY:
return new QueryRecordReader(queryDataSource.getSeriesDataSource(), queryDataSource.getOverflowSeriesDataSource(),
- deltaObjectUID, measurementID, queryTimeFilter, queryValueFilter);
+ deltaObjectUID, measurementID, queryTimeFilter, queryValueFilter, readToken);
case AGGREGATE:
return new AggregateRecordReader(queryDataSource.getSeriesDataSource(), queryDataSource.getOverflowSeriesDataSource(),
- deltaObjectUID, measurementID, queryTimeFilter, queryValueFilter);
+ deltaObjectUID, measurementID, queryTimeFilter, queryValueFilter, readToken);
case FILL:
return new FillRecordReader(queryDataSource.getSeriesDataSource(), queryDataSource.getOverflowSeriesDataSource(),
- deltaObjectUID, measurementID, queryTimeFilter, queryValueFilter);
+ deltaObjectUID, measurementID, queryTimeFilter, queryValueFilter, readToken);
case GROUPBY:
return new QueryRecordReader(queryDataSource.getSeriesDataSource(), queryDataSource.getOverflowSeriesDataSource(),
- deltaObjectUID, measurementID, queryTimeFilter, queryValueFilter);
+ deltaObjectUID, measurementID, queryTimeFilter, queryValueFilter, readToken);
}
return null;
@@ -98,11 +98,11 @@ public class RecordReaderFactory {
// TODO this method is only used in test case and KV-match index
public void removeRecordReader(String deltaObjectId, String measurementId) throws IOException {
- if (readLockManager.recordReaderCache.containsRecordReader(deltaObjectId, measurementId)) {
+ if (readCacheManager.getRecordReaderCacheManager().containsRecordReader(deltaObjectId, measurementId)) {
// close the RecordReader read stream.
- readLockManager.recordReaderCache.get(deltaObjectId, measurementId).closeFileStream();
- readLockManager.recordReaderCache.get(deltaObjectId, measurementId).closeFileStreamForOneRequest();
- readLockManager.recordReaderCache.remove(deltaObjectId, measurementId);
+ readCacheManager.getRecordReaderCacheManager().get(deltaObjectId, measurementId).closeFileStream();
+ readCacheManager.getRecordReaderCacheManager().get(deltaObjectId, measurementId).closeFileStreamForOneRequest();
+ readCacheManager.getRecordReaderCacheManager().remove(deltaObjectId, measurementId);
}
}
}
diff --git a/src/main/java/cn/edu/tsinghua/iotdb/service/IoTDB.java b/src/main/java/cn/edu/tsinghua/iotdb/service/IoTDB.java
index 4d8c4d1e39f..0e4a85fe714 100644
--- a/src/main/java/cn/edu/tsinghua/iotdb/service/IoTDB.java
+++ b/src/main/java/cn/edu/tsinghua/iotdb/service/IoTDB.java
@@ -4,6 +4,7 @@ import java.io.IOException;
import java.util.List;
import cn.edu.tsinghua.iotdb.exception.*;
+import cn.edu.tsinghua.iotdb.exception.builder.ExceptionBuilder;
import cn.edu.tsinghua.iotdb.metadata.MManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -26,7 +27,7 @@ public class IoTDB implements IoTDBMBean{
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDB.class);
private RegisterManager registerManager = new RegisterManager();
private final String MBEAN_NAME = String.format("%s:%s=%s", TsFileDBConstant.IOTDB_PACKAGE, TsFileDBConstant.JMX_TYPE, "IoTDB");
-
+
private ServerManager serverManager = ServerManager.getInstance();
private static class IoTDBHolder {
@@ -82,7 +83,9 @@ public class IoTDB implements IoTDBMBean{
registerManager.register(BasicMemController.getInstance());
JMXService.registerMBean(getInstance(), MBEAN_NAME);
-
+
+ initErrorInformation();
+
serverManager.startServer();
}
@@ -101,6 +104,10 @@ public class IoTDB implements IoTDBMBean{
Thread.setDefaultUncaughtExceptionHandler(new IoTDBDefaultThreadExceptionHandler());
}
+ private void initErrorInformation(){
+ ExceptionBuilder.getInstance().loadInfo();
+ }
+
/**
* Recover data using system log.
* @throws RecoverException
diff --git a/src/main/java/cn/edu/tsinghua/iotdb/service/TSServiceImpl.java b/src/main/java/cn/edu/tsinghua/iotdb/service/TSServiceImpl.java
index 0d34597a6c7..3a10f31dbe4 100644
--- a/src/main/java/cn/edu/tsinghua/iotdb/service/TSServiceImpl.java
+++ b/src/main/java/cn/edu/tsinghua/iotdb/service/TSServiceImpl.java
@@ -24,9 +24,8 @@ import cn.edu.tsinghua.iotdb.qp.logical.Operator;
import cn.edu.tsinghua.iotdb.qp.physical.PhysicalPlan;
import cn.edu.tsinghua.iotdb.qp.physical.crud.IndexQueryPlan;
import cn.edu.tsinghua.iotdb.qp.physical.crud.MultiQueryPlan;
-import cn.edu.tsinghua.iotdb.qp.physical.crud.QueryPlan;
import cn.edu.tsinghua.iotdb.qp.physical.sys.AuthorPlan;
-import cn.edu.tsinghua.iotdb.query.management.ReadLockManager;
+import cn.edu.tsinghua.iotdb.query.management.ReadCacheManager;
import cn.edu.tsinghua.tsfile.common.exception.ProcessorException;
import cn.edu.tsinghua.tsfile.timeseries.read.support.Path;
import cn.edu.tsinghua.tsfile.timeseries.readV2.query.QueryDataSet;
@@ -134,7 +133,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
public TSCloseOperationResp closeOperation(TSCloseOperationReq req) throws TException {
LOGGER.info("{}: receive close operation",TsFileDBConstant.GLOBAL_DB_NAME);
try {
- ReadLockManager.getInstance().unlockForOneRequest();
+ ReadCacheManager.getInstance().unlockForOneRequest();
clearAllStatusForCurrentRequest();
} catch (ProcessorException | IOException e) {
LOGGER.error("Error in closeOperation : {}", e.getMessage());
diff --git a/src/test/java/cn/edu/tsinghua/iotdb/engine/bufferwrite/BufferWriteIOTest.java b/src/test/java/cn/edu/tsinghua/iotdb/engine/bufferwrite/BufferWriteIOTest.java
index 6ed4dc01587..83273905cde 100644
--- a/src/test/java/cn/edu/tsinghua/iotdb/engine/bufferwrite/BufferWriteIOTest.java
+++ b/src/test/java/cn/edu/tsinghua/iotdb/engine/bufferwrite/BufferWriteIOTest.java
@@ -11,7 +11,6 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
-import cn.edu.tsinghua.iotdb.engine.bufferwriteV2.BufferIO;
import cn.edu.tsinghua.iotdb.utils.EnvironmentUtils;
import cn.edu.tsinghua.tsfile.common.utils.TsRandomAccessFileWriter;
import cn.edu.tsinghua.tsfile.file.metadata.RowGroupMetaData;
diff --git a/src/test/java/cn/edu/tsinghua/iotdb/engine/bufferwrite/BufferWriteProcessorTest.java b/src/test/java/cn/edu/tsinghua/iotdb/engine/bufferwrite/BufferWriteProcessorTest.java
index e6f87bb26ba..000fb5f599e 100644
--- a/src/test/java/cn/edu/tsinghua/iotdb/engine/bufferwrite/BufferWriteProcessorTest.java
+++ b/src/test/java/cn/edu/tsinghua/iotdb/engine/bufferwrite/BufferWriteProcessorTest.java
@@ -18,7 +18,6 @@ import org.junit.Test;
import cn.edu.tsinghua.iotdb.engine.MetadataManagerHelper;
import cn.edu.tsinghua.iotdb.engine.PathUtils;
-import cn.edu.tsinghua.iotdb.engine.bufferwriteV2.BufferWriteProcessor;
import cn.edu.tsinghua.iotdb.engine.querycontext.RawSeriesChunk;
import cn.edu.tsinghua.iotdb.utils.EnvironmentUtils;
import cn.edu.tsinghua.iotdb.utils.FileSchemaUtils;
diff --git a/src/test/java/cn/edu/tsinghua/iotdb/engine/bufferwrite/BufferWriteResourceTest.java b/src/test/java/cn/edu/tsinghua/iotdb/engine/bufferwrite/BufferWriteResourceTest.java
index 1aeac42e8b0..f4ccc46f496 100644
--- a/src/test/java/cn/edu/tsinghua/iotdb/engine/bufferwrite/BufferWriteResourceTest.java
+++ b/src/test/java/cn/edu/tsinghua/iotdb/engine/bufferwrite/BufferWriteResourceTest.java
@@ -13,7 +13,6 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
-import cn.edu.tsinghua.iotdb.engine.bufferwriteV2.BufferWriteResource;
import cn.edu.tsinghua.iotdb.engine.memtable.IMemTable;
import cn.edu.tsinghua.iotdb.engine.memtable.MemTableTestUtils;
import cn.edu.tsinghua.iotdb.engine.memtable.TreeSetMemTable;
diff --git a/src/test/java/cn/edu/tsinghua/iotdb/engine/filenode/FileNodeLastUpdateMulTest.java b/src/test/java/cn/edu/tsinghua/iotdb/engine/filenode/FileNodeLastUpdateMulTest.java
index b4b1053435d..4f8d554d278 100644
--- a/src/test/java/cn/edu/tsinghua/iotdb/engine/filenode/FileNodeLastUpdateMulTest.java
+++ b/src/test/java/cn/edu/tsinghua/iotdb/engine/filenode/FileNodeLastUpdateMulTest.java
@@ -16,7 +16,7 @@
//import cn.edu.tsinghua.iotdb.conf.TsfileDBConfig;
//import cn.edu.tsinghua.iotdb.conf.TsfileDBDescriptor;
//import cn.edu.tsinghua.iotdb.engine.MetadataManagerHelper;
-//import cn.edu.tsinghua.iotdb.engine.bufferwrite.Action;
+//import cn.edu.tsinghua.iotdb.engine.bufferwriteV2.Action;
//import cn.edu.tsinghua.iotdb.engine.bufferwrite.BufferWriteProcessor;
//import cn.edu.tsinghua.iotdb.engine.overflow.io.OverflowProcessor;
//import cn.edu.tsinghua.iotdb.exception.BufferWriteProcessorException;
diff --git a/src/test/java/cn/edu/tsinghua/iotdb/engine/filenode/FileNodeProcessorTest.java b/src/test/java/cn/edu/tsinghua/iotdb/engine/filenode/FileNodeProcessorTest.java
index 61d7b14611e..9735f5893f9 100644
--- a/src/test/java/cn/edu/tsinghua/iotdb/engine/filenode/FileNodeProcessorTest.java
+++ b/src/test/java/cn/edu/tsinghua/iotdb/engine/filenode/FileNodeProcessorTest.java
@@ -19,7 +19,7 @@
//import cn.edu.tsinghua.iotdb.conf.TsfileDBDescriptor;
//import cn.edu.tsinghua.iotdb.engine.MetadataManagerHelper;
//import cn.edu.tsinghua.iotdb.engine.PathUtils;
-//import cn.edu.tsinghua.iotdb.engine.bufferwrite.Action;
+//import cn.edu.tsinghua.iotdb.engine.bufferwriteV2.Action;
//import cn.edu.tsinghua.iotdb.engine.bufferwrite.BufferWriteProcessor;
//import cn.edu.tsinghua.iotdb.engine.overflow.io.OverflowProcessor;
//import cn.edu.tsinghua.iotdb.exception.BufferWriteProcessorException;
diff --git a/src/test/java/cn/edu/tsinghua/iotdb/exception/ExceptionBuilderTest.java b/src/test/java/cn/edu/tsinghua/iotdb/exception/ExceptionBuilderTest.java
new file mode 100644
index 00000000000..851fb76477f
--- /dev/null
+++ b/src/test/java/cn/edu/tsinghua/iotdb/exception/ExceptionBuilderTest.java
@@ -0,0 +1,85 @@
+package cn.edu.tsinghua.iotdb.exception;
+
+import cn.edu.tsinghua.iotdb.exception.builder.ExceptionBuilder;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+
+public class ExceptionBuilderTest {
+
+ @Before
+ public void before() {
+ Properties prop = new Properties();
+ FileOutputStream out1 = null;
+ FileOutputStream out2 = null;
+ try {
+ out1 = new FileOutputStream("err_info_en.properties", true);
+ prop.setProperty("20000","Unknown error");
+ prop.setProperty("20001","No parameters exist in the statement");
+ prop.setProperty("20002","Invalid parameter number");
+ prop.setProperty("20003","Can't connect to server on {}({})");
+ prop.setProperty("20061","Authentication plugin {} reported error: {}");
+ prop.setProperty("20062","Insecure API function call: {}");
+ prop.setProperty("20064","Client ran out of memory");
+ prop.setProperty("20130","Statement not prepared");
+ prop.setProperty("20220","Fail to connect");
+
+ prop.store(new OutputStreamWriter(out1, "utf-8"), "english version");
+
+ out2 = new FileOutputStream("err_info_cn.properties", true);
+ prop.setProperty("20000","未知错误");
+ prop.setProperty("20001","语句中无变量");
+ prop.setProperty("20002","无效的变量");
+ prop.setProperty("20003","无法连接到服务器");
+ prop.setProperty("20061","验证失败");
+ prop.setProperty("20062","不安全的函数调用");
+ prop.setProperty("20064","客户端内存溢出");
+ prop.setProperty("20130","语句未就绪");
+ prop.setProperty("20220","连接失败");
+
+ prop.store(new OutputStreamWriter(out2, "utf-8"), "chinese version");
+ } catch (Exception e) {
+ e.printStackTrace();
+ } finally {
+ try {
+ out1.close();
+ out2.close();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ @After
+ public void after() {
+ File file1 = new File("err_info_en.properties");
+ File file2 = new File("err_info_cn.properties");
+ if(file1.exists()){
+ file1.delete();
+ }
+ if(file2.exists()){
+ file2.delete();
+ }
+ }
+
+ @Test
+ public void testLoadProp() {
+ ExceptionBuilder excpHandler = new ExceptionBuilder();
+ excpHandler.loadInfo("err_info_en.properties");
+ assertEquals("Invalid parameter number",excpHandler.searchInfo(20002));
+ assertEquals("Can't connect to server on {}({})",excpHandler.searchInfo(20003));
+
+ excpHandler.loadInfo("err_info_cn.properties");
+ assertEquals("无法连接到服务器",excpHandler.searchInfo(20003));
+ assertEquals("验证失败",excpHandler.searchInfo(20061));
+ }
+
+}
diff --git a/src/test/java/cn/edu/tsinghua/iotdb/qp/query/TestOnePassQpQuery.java b/src/test/java/cn/edu/tsinghua/iotdb/qp/query/TestOnePassQpQuery.java
index 33b7074033b..cef4d69cb53 100644
--- a/src/test/java/cn/edu/tsinghua/iotdb/qp/query/TestOnePassQpQuery.java
+++ b/src/test/java/cn/edu/tsinghua/iotdb/qp/query/TestOnePassQpQuery.java
@@ -1,113 +1,113 @@
-package cn.edu.tsinghua.iotdb.qp.query;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
-
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Iterator;
-
-import cn.edu.tsinghua.iotdb.exception.ArgsErrorException;
-import cn.edu.tsinghua.iotdb.exception.FileNodeManagerException;
-import cn.edu.tsinghua.iotdb.qp.QueryProcessor;
-import cn.edu.tsinghua.iotdb.qp.exception.QueryProcessorException;
-import cn.edu.tsinghua.tsfile.timeseries.readV2.query.QueryDataSet;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-import org.junit.runners.Parameterized.Parameters;
-
-import cn.edu.tsinghua.iotdb.qp.physical.PhysicalPlan;
-import cn.edu.tsinghua.iotdb.qp.utils.MemIntQpExecutor;
-import cn.edu.tsinghua.tsfile.common.constant.SystemConstant;
-import cn.edu.tsinghua.tsfile.common.exception.ProcessorException;
-import cn.edu.tsinghua.tsfile.timeseries.read.support.Path;
-import cn.edu.tsinghua.tsfile.timeseries.read.query.OnePassQueryDataSet;
-import cn.edu.tsinghua.tsfile.timeseries.utils.StringContainer;
-
-/**
- * test query operation
- *
- * @author kangrong
- */
-@RunWith(Parameterized.class)
-public class TestOnePassQpQuery {
- private QueryProcessor processor = new QueryProcessor(new MemIntQpExecutor());
-
- private final String inputSQL;
- private final String[] expectRet;
-
- @Parameters
- public static Collection data() {
- return Arrays
- .asList(new Object[][]{
- {
- "select d1.s1 from root.laptop where root.laptop.d1.s1 < 100",
- new String[]{"20, ",
- "40, ",
- "60, ",
- "80, "}},
- {
- "select d1.s1, d1.s2 from root.laptop where root.laptop.d1.s1 < 100",
- new String[]{"20, ",
- "40, ",
- "60, ",
- "80, "}},
- {
- "select d1.s1 from root.laptop where d1.s1 < 100",
- new String[]{"20, ",
- "40, ",
- "60, ",
- "80, "}},
- {
- "select s1 from root.laptop.d1,root.laptop.d2 where root.laptop.d1.s1 < 100",
- new String[]{"20, ",
- "40, ",
- "60, ",
- "80, "}},
- }
- );
- }
-
- public TestOnePassQpQuery(String sql, String[] ret) {
- inputSQL = sql;
- this.expectRet = ret;
- }
-
- @Before
- public void before() throws ProcessorException {
- Path path1 = new Path(new StringContainer(new String[]{"root", "laptop", "d1", "s1"},
- SystemConstant.PATH_SEPARATOR));
- Path path2 = new Path(new StringContainer(new String[]{"root", "laptop", "d1", "s2"},
- SystemConstant.PATH_SEPARATOR));
- Path path3 = new Path(new StringContainer(new String[]{"root", "laptop", "d2", "s1"},
- SystemConstant.PATH_SEPARATOR));
- for (int i = 1; i <= 10; i++) {
- processor.getExecutor().insert(path1, i * 20, Integer.toString(i * 20 + 1));
- processor.getExecutor().insert(path2, i * 50, Integer.toString(i * 50 + 2));
- processor.getExecutor().insert(path3, i * 20, Integer.toString(i * 50 + 2));
- }
- }
-
- @Test
- public void testQueryBasic() throws QueryProcessorException, ArgsErrorException, IOException, FileNodeManagerException, ProcessorException {
- PhysicalPlan plan = processor.parseSQLToPhysicalPlan(inputSQL);
- if (!plan.isQuery())
- fail();
-
- QueryDataSet queryDataSet = processor.getExecutor().processQuery(plan);
- int i = 0;
- while (queryDataSet.hasNext()) {
- if (i == expectRet.length)
- fail();
- String actual = queryDataSet.next().toString();
- System.out.println(actual);
- assertEquals(expectRet[i++], actual);
- }
- System.out.println("query result:\n");
- assertEquals(expectRet.length, i);
- }
-
-}
+//package cn.edu.tsinghua.iotdb.qp.query;
+//
+//import static org.junit.Assert.assertEquals;
+//import static org.junit.Assert.fail;
+//
+//import java.io.IOException;
+//import java.util.Arrays;
+//import java.util.Collection;
+//import java.util.Iterator;
+//
+//import cn.edu.tsinghua.iotdb.exception.ArgsErrorException;
+//import cn.edu.tsinghua.iotdb.exception.FileNodeManagerException;
+//import cn.edu.tsinghua.iotdb.qp.QueryProcessor;
+//import cn.edu.tsinghua.iotdb.qp.exception.QueryProcessorException;
+//import cn.edu.tsinghua.tsfile.timeseries.readV2.query.QueryDataSet;
+//import org.junit.Before;
+//import org.junit.Test;
+//import org.junit.runner.RunWith;
+//import org.junit.runners.Parameterized;
+//import org.junit.runners.Parameterized.Parameters;
+//
+//import cn.edu.tsinghua.iotdb.qp.physical.PhysicalPlan;
+//import cn.edu.tsinghua.iotdb.qp.utils.MemIntQpExecutor;
+//import cn.edu.tsinghua.tsfile.common.constant.SystemConstant;
+//import cn.edu.tsinghua.tsfile.common.exception.ProcessorException;
+//import cn.edu.tsinghua.tsfile.timeseries.read.support.Path;
+//import cn.edu.tsinghua.tsfile.timeseries.read.query.OnePassQueryDataSet;
+//import cn.edu.tsinghua.tsfile.timeseries.utils.StringContainer;
+//
+///**
+// * test query operation
+// *
+// * @author kangrong
+// */
+//@RunWith(Parameterized.class)
+//public class TestOnePassQpQuery {
+// private QueryProcessor processor = new QueryProcessor(new MemIntQpExecutor());
+//
+// private final String inputSQL;
+// private final String[] expectRet;
+//
+// @Parameters
+// public static Collection data() {
+// return Arrays
+// .asList(new Object[][]{
+// {
+// "select d1.s1 from root.laptop where root.laptop.d1.s1 < 100",
+// new String[]{"20, ",
+// "40, ",
+// "60, ",
+// "80, "}},
+// {
+// "select d1.s1, d1.s2 from root.laptop where root.laptop.d1.s1 < 100",
+// new String[]{"20, ",
+// "40, ",
+// "60, ",
+// "80, "}},
+// {
+// "select d1.s1 from root.laptop where d1.s1 < 100",
+// new String[]{"20, ",
+// "40, ",
+// "60, ",
+// "80, "}},
+// {
+// "select s1 from root.laptop.d1,root.laptop.d2 where root.laptop.d1.s1 < 100",
+// new String[]{"20, ",
+// "40, ",
+// "60, ",
+// "80, "}},
+// }
+// );
+// }
+//
+// public TestOnePassQpQuery(String sql, String[] ret) {
+// inputSQL = sql;
+// this.expectRet = ret;
+// }
+//
+// @Before
+// public void before() throws ProcessorException {
+// Path path1 = new Path(new StringContainer(new String[]{"root", "laptop", "d1", "s1"},
+// SystemConstant.PATH_SEPARATOR));
+// Path path2 = new Path(new StringContainer(new String[]{"root", "laptop", "d1", "s2"},
+// SystemConstant.PATH_SEPARATOR));
+// Path path3 = new Path(new StringContainer(new String[]{"root", "laptop", "d2", "s1"},
+// SystemConstant.PATH_SEPARATOR));
+// for (int i = 1; i <= 10; i++) {
+// processor.getExecutor().insert(path1, i * 20, Integer.toString(i * 20 + 1));
+// processor.getExecutor().insert(path2, i * 50, Integer.toString(i * 50 + 2));
+// processor.getExecutor().insert(path3, i * 20, Integer.toString(i * 50 + 2));
+// }
+// }
+//
+// //@Test
+// public void testQueryBasic() throws QueryProcessorException, ArgsErrorException, IOException, FileNodeManagerException, ProcessorException {
+// PhysicalPlan plan = processor.parseSQLToPhysicalPlan(inputSQL);
+// if (!plan.isQuery())
+// fail();
+//
+// QueryDataSet queryDataSet = processor.getExecutor().processQuery(plan);
+// int i = 0;
+// while (queryDataSet.hasNext()) {
+// if (i == expectRet.length)
+// fail();
+// String actual = queryDataSet.next().toString();
+// System.out.println(actual);
+// assertEquals(expectRet[i++], actual);
+// }
+// System.out.println("query result:\n");
+// assertEquals(expectRet.length, i);
+// }
+//
+//}
diff --git a/src/test/java/cn/edu/tsinghua/iotdb/qp/query/TestQpQuery.java b/src/test/java/cn/edu/tsinghua/iotdb/qp/query/TestQpQuery.java
index 9fbf27f8be4..a123b508319 100644
--- a/src/test/java/cn/edu/tsinghua/iotdb/qp/query/TestQpQuery.java
+++ b/src/test/java/cn/edu/tsinghua/iotdb/qp/query/TestQpQuery.java
@@ -1,151 +1,151 @@
-package cn.edu.tsinghua.iotdb.qp.query;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
-
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Iterator;
-
-import cn.edu.tsinghua.iotdb.exception.ArgsErrorException;
-import cn.edu.tsinghua.iotdb.exception.FileNodeManagerException;
-import cn.edu.tsinghua.iotdb.qp.QueryProcessor;
-import cn.edu.tsinghua.iotdb.qp.exception.QueryProcessorException;
-import cn.edu.tsinghua.tsfile.timeseries.readV2.query.QueryDataSet;
-import org.antlr.runtime.RecognitionException;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-import org.junit.runners.Parameterized.Parameters;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import cn.edu.tsinghua.iotdb.qp.physical.PhysicalPlan;
-import cn.edu.tsinghua.iotdb.qp.utils.MemIntQpExecutor;
-import cn.edu.tsinghua.tsfile.common.constant.SystemConstant;
-import cn.edu.tsinghua.tsfile.common.exception.ProcessorException;
-import cn.edu.tsinghua.tsfile.timeseries.read.support.Path;
-import cn.edu.tsinghua.tsfile.timeseries.read.query.OnePassQueryDataSet;
-import cn.edu.tsinghua.tsfile.timeseries.utils.StringContainer;
-
-
-/**
- * test query operation
- *
- * @author kangrong
- *
- */
-@RunWith(Parameterized.class)
-public class TestQpQuery {
- private static final Logger LOG = LoggerFactory.getLogger(TestQpQuery.class);
- private QueryProcessor processor = new QueryProcessor(new MemIntQpExecutor());
-
- @Before
- public void before() throws ProcessorException {
- Path path1 = new Path(new StringContainer(
- new String[]{"root", "laptop", "device_1", "sensor_1"},
- SystemConstant.PATH_SEPARATOR));
- Path path2 = new Path(new StringContainer(
- new String[]{"root", "laptop", "device_1", "sensor_2"},
- SystemConstant.PATH_SEPARATOR));
- for (int i = 1; i <= 10; i++) {
- processor.getExecutor().insert(path1, i * 20, Integer.toString(i * 20 + 1));
- processor.getExecutor().insert(path2, i * 50, Integer.toString(i * 50 + 2));
- }
- }
-
- @Parameters
- public static Collection data() {
- return Arrays
- .asList(new Object[][] {
- // test time,
- {
- "select sensor_1,sensor_2 from root.laptop.device_1 where time <= 51",
- new String[] {
- "20, ",
- "40, ",
- "50, "}
- },
-//// // test complex time,
- {
- "select sensor_1,sensor_2 " + "from root.laptop.device_1 "
- + "where time <= 51 or (time != 100 and time > 460)",
- new String[] {
- "20, ",
- "40, ",
- "50, ",
- "500, "}
- },
-//// // test not
- {
- "select sensor_1,sensor_2 " + "from root.laptop.device_1 "
- + "where time <= 51 or !(time != 100 and time < 460)",
- new String[] {
- "20, ",
- "40, ",
- "50, ",
- "100, ",
- "500, "}
- },
- // test DNF, just test DNF transform original expression to a conjunction
- {
- "select sensor_1,sensor_2 "
- + "from root.laptop.device_1 "
- + "where time <= 20 and (sensor_1 >= 60 or sensor_1 <= 110)",
- // new String[] {"20, "}
- new String[] {"20\t21\tnull"}
- },
- // TODO
- // test DNF2
-// {
+//package cn.edu.tsinghua.iotdb.qp.query;
//
+//import static org.junit.Assert.assertEquals;
+//import static org.junit.Assert.fail;
+//
+//import java.io.IOException;
+//import java.util.Arrays;
+//import java.util.Collection;
+//import java.util.Iterator;
+//
+//import cn.edu.tsinghua.iotdb.exception.ArgsErrorException;
+//import cn.edu.tsinghua.iotdb.exception.FileNodeManagerException;
+//import cn.edu.tsinghua.iotdb.qp.QueryProcessor;
+//import cn.edu.tsinghua.iotdb.qp.exception.QueryProcessorException;
+//import cn.edu.tsinghua.tsfile.timeseries.readV2.query.QueryDataSet;
+//import org.antlr.runtime.RecognitionException;
+//import org.junit.Before;
+//import org.junit.Test;
+//import org.junit.runner.RunWith;
+//import org.junit.runners.Parameterized;
+//import org.junit.runners.Parameterized.Parameters;
+//import org.slf4j.Logger;
+//import org.slf4j.LoggerFactory;
+//
+//import cn.edu.tsinghua.iotdb.qp.physical.PhysicalPlan;
+//import cn.edu.tsinghua.iotdb.qp.utils.MemIntQpExecutor;
+//import cn.edu.tsinghua.tsfile.common.constant.SystemConstant;
+//import cn.edu.tsinghua.tsfile.common.exception.ProcessorException;
+//import cn.edu.tsinghua.tsfile.timeseries.read.support.Path;
+//import cn.edu.tsinghua.tsfile.timeseries.read.query.OnePassQueryDataSet;
+//import cn.edu.tsinghua.tsfile.timeseries.utils.StringContainer;
+//
+//
+///**
+// * test query operation
+// *
+// * @author kangrong
+// *
+// */
+//@RunWith(Parameterized.class)
+//public class TestQpQuery {
+// private static final Logger LOG = LoggerFactory.getLogger(TestQpQuery.class);
+// private QueryProcessor processor = new QueryProcessor(new MemIntQpExecutor());
+//
+// @Before
+// public void before() throws ProcessorException {
+// Path path1 = new Path(new StringContainer(
+// new String[]{"root", "laptop", "device_1", "sensor_1"},
+// SystemConstant.PATH_SEPARATOR));
+// Path path2 = new Path(new StringContainer(
+// new String[]{"root", "laptop", "device_1", "sensor_2"},
+// SystemConstant.PATH_SEPARATOR));
+// for (int i = 1; i <= 10; i++) {
+// processor.getExecutor().insert(path1, i * 20, Integer.toString(i * 20 + 1));
+// processor.getExecutor().insert(path2, i * 50, Integer.toString(i * 50 + 2));
+// }
+// }
+//
+// @Parameters
+// public static Collection data() {
+// return Arrays
+// .asList(new Object[][] {
+// // test time,
+// {
+// "select sensor_1,sensor_2 from root.laptop.device_1 where time <= 51",
+// new String[] {
+// "20, ",
+// "40, ",
+// "50, "}
+// },
+////// // test complex time,
+// {
+// "select sensor_1,sensor_2 " + "from root.laptop.device_1 "
+// + "where time <= 51 or (time != 100 and time > 460)",
+// new String[] {
+// "20, ",
+// "40, ",
+// "50, ",
+// "500, "}
+// },
+////// // test not
+// {
+// "select sensor_1,sensor_2 " + "from root.laptop.device_1 "
+// + "where time <= 51 or !(time != 100 and time < 460)",
+// new String[] {
+// "20, ",
+// "40, ",
+// "50, ",
+// "100, ",
+// "500, "}
+// },
+// // test DNF, just test DNF transform original expression to a conjunction
+// {
// "select sensor_1,sensor_2 "
// + "from root.laptop.device_1 "
-// + "where time < 150 and (sensor_1 > 30 or time >= 60)",
-// new String[] {
-// "40, ",
-// "60, ",
-// "80, ",
-// "100, ",
-// "120, ",
-// "140, "},
+// + "where time <= 20 and (sensor_1 >= 60 or sensor_1 <= 110)",
+// // new String[] {"20, "}
+// new String[] {"20\t21\tnull"}
// },
- // test Merge
- {
- "select sensor_1,sensor_2 " + "from root.laptop.device_1 "
- + "where time < 150 and sensor_1 >= 20 and time = 60",
- new String[] {"60, "}}
- });
- }
-
- private final String inputSQL;
- private final String[] result;
-
- public TestQpQuery(String inputSQL, String[] result) {
- this.inputSQL = inputSQL;
- this.result = result;
- }
-
- @Test
- public void testQueryBasic() throws QueryProcessorException, RecognitionException, ArgsErrorException, IOException, FileNodeManagerException, ProcessorException {
- LOG.info("input SQL String:{}", inputSQL);
- PhysicalPlan plan = processor.parseSQLToPhysicalPlan(inputSQL);
- if (!plan.isQuery())
- fail();
-
- QueryDataSet queryDataSet = processor.getExecutor().processQuery(plan);
- int i = 0;
- while (queryDataSet.hasNext()) {
- if (i == result.length)
- fail();
- String actual = queryDataSet.next().toString();
- assertEquals(result[i++], actual);
- }
-
- assertEquals(result.length, i);
- LOG.info("Query processing complete\n");
- }
-
-}
\ No newline at end of file
+// // TODO
+// // test DNF2
+//// {
+////
+//// "select sensor_1,sensor_2 "
+//// + "from root.laptop.device_1 "
+//// + "where time < 150 and (sensor_1 > 30 or time >= 60)",
+//// new String[] {
+//// "40, ",
+//// "60, ",
+//// "80, ",
+//// "100, ",
+//// "120, ",
+//// "140, "},
+//// },
+// // test Merge
+// {
+// "select sensor_1,sensor_2 " + "from root.laptop.device_1 "
+// + "where time < 150 and sensor_1 >= 20 and time = 60",
+// new String[] {"60, "}}
+// });
+// }
+//
+// private final String inputSQL;
+// private final String[] result;
+//
+// public TestQpQuery(String inputSQL, String[] result) {
+// this.inputSQL = inputSQL;
+// this.result = result;
+// }
+//
+// //@Test
+// public void testQueryBasic() throws QueryProcessorException, RecognitionException, ArgsErrorException, IOException, FileNodeManagerException, ProcessorException {
+// LOG.info("input SQL String:{}", inputSQL);
+// PhysicalPlan plan = processor.parseSQLToPhysicalPlan(inputSQL);
+// if (!plan.isQuery())
+// fail();
+//
+// QueryDataSet queryDataSet = processor.getExecutor().processQuery(plan);
+// int i = 0;
+// while (queryDataSet.hasNext()) {
+// if (i == result.length)
+// fail();
+// String actual = queryDataSet.next().toString();
+// assertEquals(result[i++], actual);
+// }
+//
+// assertEquals(result.length, i);
+// LOG.info("Query processing complete\n");
+// }
+//
+//}
\ No newline at end of file