You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by hu...@apache.org on 2023/05/14 13:05:00 UTC
[iotdb] branch research/core-sketch updated: add core-sketch to iotdb (#9847)
This is an automated email from the ASF dual-hosted git repository.
hui pushed a commit to branch research/core-sketch
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/research/core-sketch by this push:
new 62034c0091b add core-sketch to iotdb (#9847)
62034c0091b is described below
commit 62034c0091b0753e5c14cb8cb5bb18f76e23f4ea
Author: Haoquan Guan <ho...@outlook.com>
AuthorDate: Sun May 14 21:04:53 2023 +0800
add core-sketch to iotdb (#9847)
---
client-py/release.sh | 0
cluster/src/assembly/resources/sbin/add-node.bat | 0
cluster/src/assembly/resources/sbin/add-node.sh | 0
cluster/src/assembly/resources/sbin/nodetool.bat | 0
cluster/src/assembly/resources/sbin/nodetool.sh | 0
.../src/assembly/resources/sbin/remove-node.bat | 0
cluster/src/assembly/resources/sbin/remove-node.sh | 0
cluster/src/assembly/resources/sbin/start-node.bat | 0
cluster/src/assembly/resources/sbin/start-node.sh | 0
code-coverage/copy-code-coverage-sources.sh | 0
mvnw.sh | 0
.../resources/conf/iotdb-engine.properties | 2 +
server/src/assembly/resources/conf/iotdb-env.bat | 4 +-
server/src/assembly/resources/conf/iotdb-env.sh | 0
.../src/assembly/resources/sbin/start-server.bat | 0
server/src/assembly/resources/sbin/start-server.sh | 0
server/src/assembly/resources/sbin/stop-server.bat | 0
server/src/assembly/resources/sbin/stop-server.sh | 0
.../assembly/resources/tools/start-WalChecker.sh | 0
.../assembly/resources/tools/start-sync-client.bat | 0
.../assembly/resources/tools/start-sync-client.sh | 0
.../assembly/resources/tools/stop-sync-client.bat | 0
.../assembly/resources/tools/stop-sync-client.sh | 0
.../tsfileToolSet/print-tsfile-resource-files.sh | 0
.../tools/tsfileToolSet/print-tsfile-sketch.sh | 0
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 10 +
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 5 +
.../storagegroup/VirtualStorageGroupProcessor.java | 0
.../apache/iotdb/db/qp/constant/SQLConstant.java | 8 +-
.../qp/logical/crud/AggregationQueryOperator.java | 6 +
.../db/query/aggregation/AggregationType.java | 20 +-
.../query/aggregation/impl/MADCoreAggrResult.java | 447 +++++++++++++++++++++
.../db/query/aggregation/impl/MADDDAggrResult.java | 323 +++++++++++++++
.../db/query/aggregation/impl/MADQSAggrResult.java | 311 ++++++++++++++
.../db/query/factory/AggregateResultFactory.java | 18 +
.../iotdb/db/query/pool/QueryTaskManager.java | 1 +
.../org/apache/iotdb/db/utils/SchemaUtils.java | 3 +
.../apache/iotdb/db/utils/TypeInferenceUtils.java | 3 +
.../org/apache/iotdb/session/InsertCsvDataIT.java | 22 +-
.../org/apache/iotdb/spark/tsfile/qp/Executor.java | 0
.../iotdb/spark/tsfile/qp/QueryProcessor.java | 0
.../spark/tsfile/qp/common/BasicOperator.java | 0
.../spark/tsfile/qp/common/FilterOperator.java | 0
.../iotdb/spark/tsfile/qp/common/Operator.java | 0
.../iotdb/spark/tsfile/qp/common/SQLConstant.java | 0
.../iotdb/spark/tsfile/qp/common/SingleQuery.java | 0
.../iotdb/spark/tsfile/qp/common/TSQueryPlan.java | 0
.../qp/exception/BasicOperatorException.java | 0
.../tsfile/qp/exception/DNFOptimizeException.java | 0
.../qp/exception/LogicalOptimizeException.java | 0
.../tsfile/qp/exception/MergeFilterException.java | 0
.../qp/exception/QueryOperatorException.java | 0
.../qp/exception/QueryProcessorException.java | 0
.../tsfile/qp/exception/RemoveNotException.java | 0
.../tsfile/qp/optimizer/DNFFilterOptimizer.java | 0
.../tsfile/qp/optimizer/IFilterOptimizer.java | 0
.../qp/optimizer/MergeSingleFilterOptimizer.java | 0
.../tsfile/qp/optimizer/PhysicalOptimizer.java | 0
.../tsfile/qp/optimizer/RemoveNotOptimizer.java | 0
.../apache/iotdb/spark/tsfile/DefaultSource.scala | 0
.../apache/iotdb/spark/tsfile/WideConverter.scala | 0
.../org/apache/iotdb/spark/tsfile/package.scala | 0
test/e2e/cases/cli/cleanup.sh | 0
test/e2e/cases/cli/run.sh | 0
thrift-sync/src/main/thrift/sync.thrift | 0
tools/check_sigs.sh | 0
tools/common.sh | 0
tools/download_staged_release.sh | 0
.../tsfile/read/filter/basic/BinaryFilter.java | 0
.../iotdb/tsfile/read/filter/basic/Filter.java | 0
.../tsfile/read/filter/basic/UnaryFilter.java | 0
.../tsfile/read/filter/operator/AndFilter.java | 0
.../iotdb/tsfile/read/filter/operator/Eq.java | 0
.../iotdb/tsfile/read/filter/operator/Gt.java | 0
.../iotdb/tsfile/read/filter/operator/GtEq.java | 0
.../iotdb/tsfile/read/filter/operator/Lt.java | 0
.../iotdb/tsfile/read/filter/operator/LtEq.java | 0
.../iotdb/tsfile/read/filter/operator/NotEq.java | 0
.../tsfile/read/filter/operator/NotFilter.java | 0
.../tsfile/read/filter/operator/OrFilter.java | 0
.../org/apache/iotdb/tsfile/utils/CORESketch.java | 414 +++++++++++++++++++
.../org/apache/iotdb/tsfile/utils/DDSketch.java | 174 ++++++++
.../org/apache/iotdb/tsfile/utils/FileHelper.java | 111 +++++
.../java/org/apache/iotdb/tsfile/utils/Mad.java | 33 ++
.../java/org/apache/iotdb/tsfile/utils/Params.java | 9 +
.../org/apache/iotdb/tsfile/utils/TPSketch.java | 447 +++++++++++++++++++++
.../org/apache/iotdb/tsfile/read/ReadTest.java | 0
.../iotdb/tsfile/read/TimePlainEncodeReadTest.java | 0
.../TsFileGeneratorForSeriesReaderByTimestamp.java | 0
.../apache/iotdb/tsfile/utils/FileGenerator.java | 0
.../apache/iotdb/tsfile/utils/FileUtilsTest.java | 0
.../apache/iotdb/tsfile/utils/RecordUtilsTest.java | 0
.../iotdb/tsfile/utils/StringContainerTest.java | 0
.../iotdb/tsfile/utils/TsFileGeneratorForTest.java | 0
.../org/apache/iotdb/tsfile/write/PerfTest.java | 0
.../org/apache/iotdb/tsfile/write/WriteTest.java | 0
.../iotdb/tsfile/write/writer/PageWriterTest.java | 0
97 files changed, 2355 insertions(+), 16 deletions(-)
diff --git a/client-py/release.sh b/client-py/release.sh
old mode 100755
new mode 100644
diff --git a/cluster/src/assembly/resources/sbin/add-node.bat b/cluster/src/assembly/resources/sbin/add-node.bat
old mode 100755
new mode 100644
diff --git a/cluster/src/assembly/resources/sbin/add-node.sh b/cluster/src/assembly/resources/sbin/add-node.sh
old mode 100755
new mode 100644
diff --git a/cluster/src/assembly/resources/sbin/nodetool.bat b/cluster/src/assembly/resources/sbin/nodetool.bat
old mode 100755
new mode 100644
diff --git a/cluster/src/assembly/resources/sbin/nodetool.sh b/cluster/src/assembly/resources/sbin/nodetool.sh
old mode 100755
new mode 100644
diff --git a/cluster/src/assembly/resources/sbin/remove-node.bat b/cluster/src/assembly/resources/sbin/remove-node.bat
old mode 100755
new mode 100644
diff --git a/cluster/src/assembly/resources/sbin/remove-node.sh b/cluster/src/assembly/resources/sbin/remove-node.sh
old mode 100755
new mode 100644
diff --git a/cluster/src/assembly/resources/sbin/start-node.bat b/cluster/src/assembly/resources/sbin/start-node.bat
old mode 100755
new mode 100644
diff --git a/cluster/src/assembly/resources/sbin/start-node.sh b/cluster/src/assembly/resources/sbin/start-node.sh
old mode 100755
new mode 100644
diff --git a/code-coverage/copy-code-coverage-sources.sh b/code-coverage/copy-code-coverage-sources.sh
old mode 100755
new mode 100644
diff --git a/mvnw.sh b/mvnw.sh
old mode 100755
new mode 100644
diff --git a/server/src/assembly/resources/conf/iotdb-engine.properties b/server/src/assembly/resources/conf/iotdb-engine.properties
index 679996bf472..e99fb75db38 100644
--- a/server/src/assembly/resources/conf/iotdb-engine.properties
+++ b/server/src/assembly/resources/conf/iotdb-engine.properties
@@ -20,6 +20,8 @@
####################
### RPC Configuration
####################
+mad_query_thread=64
+concurrent_query_thread=24
query_timeout_threshold=6000000
no_update=false
diff --git a/server/src/assembly/resources/conf/iotdb-env.bat b/server/src/assembly/resources/conf/iotdb-env.bat
index 91db2cfd474..f1515d987fd 100644
--- a/server/src/assembly/resources/conf/iotdb-env.bat
+++ b/server/src/assembly/resources/conf/iotdb-env.bat
@@ -63,14 +63,14 @@ set /a half_=%system_memory_in_mb%/2
set /a oneThird_=%system_memory_in_mb%/3
set /a quarter_=%half_%/2
-if %half_% GTR 2048 set half_=2048
+@rem if %half_% GTR 2048 set half_=2048
if %quarter_% GTR 65536 set quarter_=65536
if %half_% GTR %quarter_% (
set max_heap_size_in_mb=%half_%
) else set max_heap_size_in_mb=%quarter_%
-set max_heap_size_in_mb=%oneThird_%
+set max_heap_size_in_mb=%half_%
set MAX_HEAP_SIZE=%max_heap_size_in_mb%M
set max_sensible_yg_per_core_in_mb=100
diff --git a/server/src/assembly/resources/conf/iotdb-env.sh b/server/src/assembly/resources/conf/iotdb-env.sh
old mode 100755
new mode 100644
diff --git a/server/src/assembly/resources/sbin/start-server.bat b/server/src/assembly/resources/sbin/start-server.bat
old mode 100755
new mode 100644
diff --git a/server/src/assembly/resources/sbin/start-server.sh b/server/src/assembly/resources/sbin/start-server.sh
old mode 100755
new mode 100644
diff --git a/server/src/assembly/resources/sbin/stop-server.bat b/server/src/assembly/resources/sbin/stop-server.bat
old mode 100755
new mode 100644
diff --git a/server/src/assembly/resources/sbin/stop-server.sh b/server/src/assembly/resources/sbin/stop-server.sh
old mode 100755
new mode 100644
diff --git a/server/src/assembly/resources/tools/start-WalChecker.sh b/server/src/assembly/resources/tools/start-WalChecker.sh
old mode 100755
new mode 100644
diff --git a/server/src/assembly/resources/tools/start-sync-client.bat b/server/src/assembly/resources/tools/start-sync-client.bat
old mode 100755
new mode 100644
diff --git a/server/src/assembly/resources/tools/start-sync-client.sh b/server/src/assembly/resources/tools/start-sync-client.sh
old mode 100755
new mode 100644
diff --git a/server/src/assembly/resources/tools/stop-sync-client.bat b/server/src/assembly/resources/tools/stop-sync-client.bat
old mode 100755
new mode 100644
diff --git a/server/src/assembly/resources/tools/stop-sync-client.sh b/server/src/assembly/resources/tools/stop-sync-client.sh
old mode 100755
new mode 100644
diff --git a/server/src/assembly/resources/tools/tsfileToolSet/print-tsfile-resource-files.sh b/server/src/assembly/resources/tools/tsfileToolSet/print-tsfile-resource-files.sh
old mode 100755
new mode 100644
diff --git a/server/src/assembly/resources/tools/tsfileToolSet/print-tsfile-sketch.sh b/server/src/assembly/resources/tools/tsfileToolSet/print-tsfile-sketch.sh
old mode 100755
new mode 100644
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index a9dff4239ce..e587e026c33 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -262,6 +262,8 @@ public class IoTDBConfig {
/** How many threads can concurrently execute query statement. When <= 0, use CPU core number. */
private int concurrentQueryThread = 16;
+ private int MADQueryThread = 32;
+
/**
* How many threads can concurrently read data for raw data query. When <= 0, use CPU core number.
*/
@@ -1201,6 +1203,14 @@ public class IoTDBConfig {
this.concurrentQueryThread = concurrentQueryThread;
}
+ public int getMADQueryThread() {
+ return MADQueryThread;
+ }
+
+ public void setMADQueryThread(int t) {
+ this.MADQueryThread = t;
+ }
+
public int getConcurrentSubRawQueryThread() {
return concurrentSubRawQueryThread;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 22d25a223aa..ca061f6cac4 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -461,6 +461,11 @@ public class IoTDBDescriptor {
properties.getProperty(
"concurrent_query_thread", Integer.toString(conf.getConcurrentQueryThread()))));
+ conf.setMADQueryThread(
+ Integer.parseInt(
+ properties.getProperty(
+ "mad_query_thread", Integer.toString(conf.getMADQueryThread()))));
+
if (conf.getConcurrentQueryThread() <= 0) {
conf.setConcurrentQueryThread(Runtime.getRuntime().availableProcessors());
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/VirtualStorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/VirtualStorageGroupProcessor.java
old mode 100755
new mode 100644
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java b/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
index b102a1be6cf..41c6575f3bd 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/constant/SQLConstant.java
@@ -133,6 +133,9 @@ public class SQLConstant {
public static final String EXACT_MULTI_QUANTILES_TDIGEST = "exact_multi_quantiles_tdigest";
public static final String EXACT_MULTI_QUANTILES_DDSKETCH_POSITIVE =
"exact_multi_quantiles_ddsketch_positive";
+ public static final String MAD_DD = "mad_dd";
+ public static final String MAD_QS = "mad_qs";
+ public static final String MAD_CORE = "mad_core";
public static final String ALL = "all";
@@ -195,7 +198,10 @@ public class SQLConstant {
EXACT_MULTI_QUANTILES_PR_KLL_FIX_PR,
EXACT_MULTI_QUANTILES_MRL,
EXACT_MULTI_QUANTILES_TDIGEST,
- EXACT_MULTI_QUANTILES_DDSKETCH_POSITIVE));
+ EXACT_MULTI_QUANTILES_DDSKETCH_POSITIVE,
+ MAD_DD,
+ MAD_QS,
+ MAD_CORE));
public static final int TOK_WHERE = 23;
public static final int TOK_INSERT = 24;
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/AggregationQueryOperator.java b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/AggregationQueryOperator.java
index 04b947bf7a8..18f72a5b861 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/AggregationQueryOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/AggregationQueryOperator.java
@@ -178,6 +178,9 @@ public class AggregationQueryOperator extends QueryOperator {
case SQLConstant.EXACT_MULTI_QUANTILES_MRL:
case SQLConstant.EXACT_MULTI_QUANTILES_TDIGEST:
case SQLConstant.EXACT_MULTI_QUANTILES_DDSKETCH_POSITIVE:
+ case SQLConstant.MAD_DD:
+ case SQLConstant.MAD_CORE:
+ case SQLConstant.MAD_QS:
return dataType.isNumeric();
case SQLConstant.COUNT:
case SQLConstant.MIN_TIME:
@@ -250,6 +253,9 @@ public class AggregationQueryOperator extends QueryOperator {
case SQLConstant.EXACT_MULTI_QUANTILES_MRL:
case SQLConstant.EXACT_MULTI_QUANTILES_TDIGEST:
case SQLConstant.EXACT_MULTI_QUANTILES_DDSKETCH_POSITIVE:
+ case SQLConstant.MAD_DD:
+ case SQLConstant.MAD_CORE:
+ case SQLConstant.MAD_QS:
return dataTypes.stream().allMatch(dataTypes.get(0)::equals);
default:
return true;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregationType.java b/server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregationType.java
index 49cb502cde4..c60e5b888ae 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregationType.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/aggregation/AggregationType.java
@@ -82,7 +82,10 @@ public enum AggregationType {
EXACT_MULTI_QUANTILES_PR_KLL_FIX_PR,
EXACT_MULTI_QUANTILES_MRL,
EXACT_MULTI_QUANTILES_TDIGEST,
- EXACT_MULTI_QUANTILES_DDSKETCH_POSITIVE;
+ EXACT_MULTI_QUANTILES_DDSKETCH_POSITIVE,
+ MAD_DD,
+ MAD_QS,
+ MAD_CORE;
/**
* give an integer to return a data type.
@@ -206,6 +209,12 @@ public enum AggregationType {
return EXACT_MULTI_QUANTILES_TDIGEST;
case 56:
return EXACT_MULTI_QUANTILES_DDSKETCH_POSITIVE;
+ case 57:
+ return MAD_DD;
+ case 58:
+ return MAD_CORE;
+ case 59:
+ return MAD_QS;
default:
throw new IllegalArgumentException("Invalid Aggregation Type: " + i);
@@ -386,6 +395,15 @@ public enum AggregationType {
case EXACT_MULTI_QUANTILES_DDSKETCH_POSITIVE:
i = 56;
break;
+ case MAD_DD:
+ i = 57;
+ break;
+ case MAD_CORE:
+ i = 58;
+ break;
+ case MAD_QS:
+ i = 59;
+ break;
default:
throw new IllegalArgumentException("Invalid Aggregation Type: " + this.name());
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MADCoreAggrResult.java b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MADCoreAggrResult.java
new file mode 100644
index 00000000000..09869c8c905
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MADCoreAggrResult.java
@@ -0,0 +1,447 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.query.aggregation.impl;
+
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.aggregation.AggregationType;
+import org.apache.iotdb.db.query.reader.series.IReaderByTimestamp;
+import org.apache.iotdb.db.utils.ValueIterator;
+import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
+import org.apache.iotdb.tsfile.read.common.IBatchDataIterator;
+import org.apache.iotdb.tsfile.utils.CORESketch;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
+import it.unimi.dsi.util.XoRoShiRo128PlusRandom;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+import static org.apache.iotdb.tsfile.file.metadata.enums.TSDataType.DOUBLE;
+
+public class MADCoreAggrResult extends AggregateResult {
+ private String returnType = "value";
+ private TSDataType seriesDataType;
+ private int iteration, BUCKET = 2333, card = 0, pre_card = 0;
+ private double DataMinV, DataMaxV, median, memory = 0, mad, EPSILON;
+ private double[] useful_range;
+ private DoubleArrayList queue;
+ private long n;
+ private CORESketch sketch, finestSketch;
+ DoubleArrayList lastPassData;
+ boolean lastPass = false;
+ long lastN = Long.MAX_VALUE;
+ private int preComputedSketchSize;
+ private boolean hasFinalResult, FindingMedian = true, is_integer = false;
+ public static XoRoShiRo128PlusRandom random = new XoRoShiRo128PlusRandom();
+ long DEBUG = 0;
+
+ public static class SpacePair {
+ public double thread_memory;
+ public CORESketch sketch;
+
+ public SpacePair(double thread_memory, CORESketch sketch) {
+ this.thread_memory = thread_memory;
+ this.sketch = sketch;
+ }
+ }
+
+ private int getBitsOfDataType() {
+ switch (seriesDataType) {
+ case INT32:
+ case FLOAT:
+ return 32;
+ case INT64:
+ case DOUBLE:
+ return 64;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type in aggregation MEDIAN : %s", seriesDataType));
+ }
+ }
+
+ public MADCoreAggrResult(TSDataType seriesDataType) throws UnSupportedDataTypeException {
+ super(DOUBLE, AggregationType.MAD_CORE);
+ this.seriesDataType = seriesDataType;
+ reset();
+ }
+
+ private long dataToLong(Object data) throws UnSupportedDataTypeException {
+ long result;
+ switch (seriesDataType) {
+ case INT32:
+ return (int) data;
+ case FLOAT:
+ result = Float.floatToIntBits((float) data);
+ return (float) data >= 0f ? result : result ^ Long.MAX_VALUE;
+ case INT64:
+ return (long) data;
+ case DOUBLE:
+ result = Double.doubleToLongBits((double) data);
+ return (double) data >= 0d ? result : result ^ Long.MAX_VALUE;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type in aggregation MEDIAN : %s", seriesDataType));
+ }
+ }
+
+ private double longToResult(long result) throws UnSupportedDataTypeException {
+ switch (seriesDataType) {
+ case INT32:
+ return (double) (result);
+ case FLOAT:
+ result = (result >>> 31) == 0 ? result : result ^ Long.MAX_VALUE;
+ return Float.intBitsToFloat((int) (result));
+ case INT64:
+ return (double) (result);
+ case DOUBLE:
+ result = (result >>> 63) == 0 ? result : result ^ Long.MAX_VALUE;
+ return Double.longBitsToDouble(result);
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type in aggregation MEDIAN : %s", seriesDataType));
+ }
+ }
+
+ private void updateStatusFromData(Object data) {
+ double dataD = (double) data;
+ if (iteration == 0) {
+ n++;
+ DataMinV = Math.min(DataMinV, dataD);
+ DataMaxV = Math.max(DataMaxV, dataD);
+ } else if (FindingMedian) {
+ sketch.insert(dataD - DataMinV + 1);
+ } else {
+ sketch.insert_mid(dataD - DataMinV + 1);
+ if (sketch.in_range(dataD - DataMinV + 1, useful_range)) {
+ queue.add(dataD - DataMinV + 1);
+ }
+ }
+ }
+
+ private static int calculate_card(double max_value, double min_value, int space_limit) {
+ int card = 0;
+ while (
+ /*true*/ card <= 30) {
+ if (Math.ceil(Math.pow(2, card) * Math.log(max_value) / Math.log(2))
+ - Math.floor(Math.pow(2, card) * Math.log(min_value) / Math.log(2))
+ > space_limit) {
+ return card - 1;
+ }
+ card += 1;
+ }
+ return card;
+ }
+
+ private static int calculate_card(double[] real_range, int space_limit) {
+ int card = 0;
+ while (
+ /*true*/ card <= 30) {
+ int space_expected = 0;
+ for (int i = 0; i < real_range[6]; ++i) {
+ space_expected +=
+ Math.ceil(Math.pow(2, card) * Math.log(real_range[2 * i + 1]) / Math.log(2))
+ - Math.floor(Math.pow(2, card) * Math.log(real_range[2 * i]) / Math.log(2));
+ }
+ if (space_expected > space_limit) {
+ return card - 1;
+ }
+ card += 1;
+ }
+ return card;
+ }
+
+ @Override
+ public void startIteration() {
+ if (iteration == 0) { // first iteration
+ DataMinV = Double.MAX_VALUE;
+ DataMaxV = -Double.MAX_VALUE;
+ n = 0;
+ } else {
+ if (iteration == 1) {
+ double[] range =
+ new double[] {
+ 1, DataMaxV - DataMinV + 1, 1, DataMaxV - DataMinV + 1, 1, DataMaxV - DataMinV + 1, 1
+ };
+ card = calculate_card(range, BUCKET);
+ sketch = new CORESketch(card, BUCKET, range);
+ pre_card = sketch.card;
+ } else if (FindingMedian) {
+ pre_card = sketch.card;
+ } else {
+ useful_range = sketch.get_range();
+ sketch = new CORESketch(1);
+ queue = new DoubleArrayList();
+ sketch.set_range(useful_range);
+ }
+ }
+ }
+
+ @Override
+ public void setDoubleValue(double doubleValue) {
+ this.hasCandidateResult = true;
+ if (this.returnType.equals("mad")) this.doubleValue = mad;
+ else if (this.returnType.equals("memory")) this.doubleValue = memory;
+ else if (this.returnType.equals("median")) this.doubleValue = median + DataMinV - 1;
+ else if (this.returnType.equals("iteration"))
+ this.doubleValue = iteration - 1; // ignore first pass for MinMaxV
+ }
+
+ @Override
+ public void finishIteration() {
+ iteration++;
+ if (n == 0) {
+ setDoubleValue(0);
+ hasFinalResult = true;
+ return;
+ }
+
+ if (iteration == 1) { // first iteration over
+ System.out.println(
+ "\t[MAD DEBUG CORESketch]\tfirst Pass for MinV,MaxV:\t"
+ + DataMinV
+ + "\t"
+ + DataMaxV
+ + "\tn:\t"
+ + n);
+ return;
+ }
+ if (iteration >= 2 && FindingMedian) { // find median
+ memory = Math.max(memory, sketch.get_bucket_size() * 48.13 / 1024);
+
+ if (sketch.data_read_norm()) {
+ finestSketch = sketch;
+ FindingMedian = false;
+ return;
+ }
+ int m = sketch.mid_half_count_bucket();
+ int[] lr = sketch.edge_half_count_bucket(m);
+ if (is_integer && sketch.bucket_finest(m, lr[0], lr[1])) {
+ finestSketch = sketch;
+ FindingMedian = false;
+ return;
+ }
+ double[] next_range = sketch.generate_useful_range(m, lr[0], lr[1]);
+ memory = Math.max(sketch.get_bucket_size() * 48.13 / 1024, memory);
+ sketch = new CORESketch(0, BUCKET, next_range);
+ double[] real_range = sketch.real_range();
+ card = calculate_card(real_range, BUCKET);
+ if (card == pre_card) {
+ this.hasCandidateResult = true;
+ this.hasFinalResult = true;
+ this.doubleValue = -233;
+ System.out.println(
+ "\t\t[MAD ERROR CORESketch]\tLarger bucket limit needed!!!!!!!!!!!!!!!!!!");
+ return;
+ }
+ sketch.set_card(card);
+ return;
+ }
+ if (!FindingMedian) {
+ memory = Math.max(memory, ((double) queue.size()) * 8 / 1024);
+ // System.out.println("\t\t\t\tafter in range check. |queue|="+queue.size());
+ long[] mid_num = sketch.get_gap();
+ long n_queue = queue.size(), median_rank = (n - 1) / 2 - mid_num[1] - mid_num[0];
+ if (median_rank < 0) median_rank += mid_num[1];
+ assert median_rank >= 0;
+ // System.out.println("\t\t\tmid_rank:"+median_rank);
+ median = getKth(queue, 0, (int) n_queue, (int) median_rank);
+ queue.replaceAll(aDouble -> Math.abs(aDouble - median));
+ double min_mad = Math.max(median - useful_range[1], useful_range[4] - median);
+ long mad_rank = (n - 1) / 2 - mid_num[1] - mid_num[2];
+ if (mad_rank < 0) mad_rank += mid_num[1] + mid_num[2];
+ mad = getKth(queue, 0, (int) n_queue, (int) mad_rank);
+ setDoubleValue(mad);
+ hasFinalResult = true;
+ }
+ }
+
+ public static double getKth(DoubleArrayList data, int L, int R, int K) {
+ if (L >= R) return data.getDouble(L);
+ int pos = L + random.nextInt(R - L);
+ double pivot_v = data.getDouble(pos), swap_v;
+
+ int leP = L, eqR = R;
+ data.set(pos, data.set(--eqR, pivot_v)); // [L,leP): < pivot_v ; [eqR,R): == pivot_v ;
+
+ for (int i = L; i < eqR; i++)
+ if ((swap_v = data.getDouble(i)) < pivot_v) data.set(i, data.set(leP++, swap_v));
+ else if (swap_v == pivot_v) {
+ data.set(i--, data.set(--eqR, swap_v));
+ }
+
+ // if(R-eqR>1)System.out.println("\t\t\t\tk_select. same pivot v. count:"+(R-eqR));
+ if (K < leP - L) return getKth(data, L, leP, K);
+ if (K >= (leP - L) + (R - eqR)) return getKth(data, leP, eqR, K - (leP - L) - (R - eqR));
+ return pivot_v;
+ }
+
+ @Override
+ protected boolean hasCandidateResult() {
+ return hasFinalResult && n > 0;
+ }
+
+ @Override
+ public Double getResult() {
+ return hasCandidateResult() ? getDoubleValue() : null;
+ }
+
+ @Override
+ public void updateResultFromStatistics(Statistics statistics) {
+ switch (statistics.getType()) {
+ case INT32:
+ case INT64:
+ case FLOAT:
+ case DOUBLE:
+ break;
+ case TEXT:
+ case BOOLEAN:
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format(
+ "Unsupported data type in aggregation MEDIAN : %s", statistics.getType()));
+ }
+ n += statistics.getCount();
+ DataMinV = Math.min(DataMinV, (double) statistics.getMinValue());
+ DataMaxV = Math.max(DataMaxV, (double) statistics.getMaxValue());
+ }
+
+ @Override
+ public void updateResultFromPageData(IBatchDataIterator batchIterator) {
+ updateResultFromPageData(batchIterator, Long.MIN_VALUE, Long.MAX_VALUE);
+ }
+
+ @Override
+ public void updateResultFromPageData(
+ IBatchDataIterator batchIterator, long minBound, long maxBound) {
+ while (batchIterator.hasNext()) {
+ if (batchIterator.currentTime() >= maxBound || batchIterator.currentTime() < minBound) {
+ break;
+ }
+ updateStatusFromData(batchIterator.currentValue());
+ batchIterator.next();
+ }
+ }
+
+ @Override
+ public void updateResultUsingTimestamps(
+ long[] timestamps, int length, IReaderByTimestamp dataReader) throws IOException {
+ Object[] values = dataReader.getValuesInTimestamps(timestamps, length);
+ for (int i = 0; i < length; i++) {
+ if (values[i] != null) {
+ updateStatusFromData(values[i]);
+ }
+ }
+ }
+
+ @Override
+ public void updateResultUsingValues(long[] timestamps, int length, ValueIterator valueIterator) {
+ // List<Object> tmp = new ArrayList<>();
+ while (valueIterator.hasNext()) {
+ updateStatusFromData(valueIterator.next());
+ // Object tmpObj = valueIterator.next();
+ // updateStatusFromData(tmpObj, 1);
+ // tmp.add(tmpObj);
+ }
+ //
+ // System.out.println("\t\t[MEDIAN]"+this.hashCode()+"[updateResultUsingValues]"+tmp.toString());
+ }
+
+ @Override
+ public int maxIteration() {
+ return 10;
+ }
+
+ @Override
+ public boolean hasFinalResult() {
+ return hasFinalResult;
+ }
+
+ @Override
+ public void merge(AggregateResult another) {
+ // System.out.println("[DEBUG] [merge] " + this.getResult() + " " + another.getResult());
+ // merge not supported
+ // throw new QueryProcessException("Can't merge MedianAggregateResult");
+ }
+
+ @Override
+ protected void deserializeSpecificFields(ByteBuffer buffer) {
+ this.seriesDataType = TSDataType.deserialize(buffer.get());
+ // TODO
+ }
+
+ @Override
+ protected void serializeSpecificFields(OutputStream outputStream) throws IOException {
+ ReadWriteIOUtils.write(seriesDataType, outputStream);
+ // TODO
+ }
+
+ public long getN() {
+ return n;
+ }
+
+ @Override
+ public void reset() {
+ super.reset();
+ sketch = null;
+ mad = median = memory = 0;
+ n = 0;
+ iteration = 0;
+ hasFinalResult = false;
+ preComputedSketchSize = 0;
+ }
+
+ @Override
+ public boolean canUpdateFromStatistics(Statistics statistics) {
+ return iteration == 0;
+ }
+
+ @Override
+ public boolean groupByLevelBeforeAggregation() {
+ return true;
+ }
+
+ @Override
+ public boolean useStatisticsIfPossible() {
+ return false;
+ }
+
+ @Override
+ public void setAttributes(Map<String, String> attrs) {
+ if (attrs.containsKey("bucket")) {
+ String b = attrs.get("bucket");
+ this.BUCKET = Integer.parseInt(b);
+ }
+ if (attrs.containsKey("is_integer")) {
+ String ii = attrs.get("is_integer");
+ this.is_integer = Boolean.parseBoolean(ii);
+ }
+ if (attrs.containsKey("return_type")) {
+ String q = attrs.get("return_type");
+ this.returnType = q;
+ }
+ System.out.println(
+ " [setAttributes DEBUG]\t\t\tbucket:" + BUCKET + "\t\tis_integer:" + is_integer);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MADDDAggrResult.java b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MADDDAggrResult.java
new file mode 100644
index 00000000000..293bd24e7d5
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MADDDAggrResult.java
@@ -0,0 +1,323 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.query.aggregation.impl;
+
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.aggregation.AggregationType;
+import org.apache.iotdb.db.query.reader.series.IReaderByTimestamp;
+import org.apache.iotdb.db.utils.ValueIterator;
+import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
+import org.apache.iotdb.tsfile.read.common.IBatchDataIterator;
+import org.apache.iotdb.tsfile.utils.DDSketch;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+import static org.apache.iotdb.tsfile.file.metadata.enums.TSDataType.DOUBLE;
+
+public class MADDDAggrResult extends AggregateResult {
+ private String returnType = "value";
+ private TSDataType seriesDataType;
+ private int iteration, BUCKET = 2333;
+ private double DataMinV, DataMaxV, median, memory = 0, mad, EPSILON;
+ private long n;
+ private DDSketch sketch;
+ DoubleArrayList lastPassData;
+ boolean lastPass = false;
+ long lastN = Long.MAX_VALUE;
+ private int preComputedSketchSize;
+ private boolean hasFinalResult;
+ long DEBUG = 0;
+
+ private int getBitsOfDataType() {
+ switch (seriesDataType) {
+ case INT32:
+ case FLOAT:
+ return 32;
+ case INT64:
+ case DOUBLE:
+ return 64;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type in aggregation MEDIAN : %s", seriesDataType));
+ }
+ }
+
+ public MADDDAggrResult(TSDataType seriesDataType) throws UnSupportedDataTypeException {
+ super(DOUBLE, AggregationType.MAD_DD);
+ this.seriesDataType = seriesDataType;
+ reset();
+ }
+
+ private long dataToLong(Object data) throws UnSupportedDataTypeException {
+ long result;
+ switch (seriesDataType) {
+ case INT32:
+ return (int) data;
+ case FLOAT:
+ result = Float.floatToIntBits((float) data);
+ return (float) data >= 0f ? result : result ^ Long.MAX_VALUE;
+ case INT64:
+ return (long) data;
+ case DOUBLE:
+ result = Double.doubleToLongBits((double) data);
+ return (double) data >= 0d ? result : result ^ Long.MAX_VALUE;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type in aggregation MEDIAN : %s", seriesDataType));
+ }
+ }
+
+ private double longToResult(long result) throws UnSupportedDataTypeException {
+ switch (seriesDataType) {
+ case INT32:
+ return (double) (result);
+ case FLOAT:
+ result = (result >>> 31) == 0 ? result : result ^ Long.MAX_VALUE;
+ return Float.intBitsToFloat((int) (result));
+ case INT64:
+ return (double) (result);
+ case DOUBLE:
+ result = (result >>> 63) == 0 ? result : result ^ Long.MAX_VALUE;
+ return Double.longBitsToDouble(result);
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type in aggregation MEDIAN : %s", seriesDataType));
+ }
+ }
+
+ private void updateStatusFromData(Object data) {
+ double dataD = (double) data;
+ if (iteration == 0) {
+ n++;
+ DataMinV = Math.min(DataMinV, dataD);
+ DataMaxV = Math.max(DataMaxV, dataD);
+ } else if (iteration == 1) {
+ sketch.insert(dataD - DataMinV + 1);
+ } else if (iteration == 2) {
+ sketch.insert(Math.abs(dataD - DataMinV + 1 - median) + 1);
+ }
+ }
+
+ @Override
+ public void startIteration() {
+ if (iteration == 0) { // first iteration
+ DataMinV = Double.MAX_VALUE;
+ DataMaxV = -Double.MAX_VALUE;
+ n = 0;
+ } else {
+ sketch = new DDSketch(EPSILON, BUCKET);
+ }
+ }
+
+ @Override
+ public void setDoubleValue(double doubleValue) {
+ this.hasCandidateResult = true;
+ if (this.returnType.equals("mad")) this.doubleValue = mad;
+ else if (this.returnType.equals("memory")) this.doubleValue = memory;
+ else if (this.returnType.equals("median")) this.doubleValue = median + DataMinV - 1;
+ }
+
+ @Override
+ public void finishIteration() {
+ iteration++;
+ if (n == 0) {
+ setDoubleValue(0);
+ hasFinalResult = true;
+ return;
+ }
+
+ if (iteration == 1) { // first iteration over
+ System.out.println(
+ "\t[MAD DEBUG DDSketch]\tfirst Pass for MinV,MaxV:\t"
+ + DataMinV
+ + "\t"
+ + DataMaxV
+ + "\tn:\t"
+ + n);
+ return;
+ }
+ if (iteration == 2) { // find median
+ memory = Math.max(memory, sketch.sketch_size() * 48 / 1024.0);
+ median = sketch.getQuantile(0.5);
+ return;
+ }
+ if (iteration == 3) {
+ memory = Math.max(memory, sketch.sketch_size() * 48 / 1024.0);
+ mad = sketch.getQuantile(0.5) - 1;
+ setDoubleValue(mad);
+ hasFinalResult = true;
+ }
+ }
+
+ @Override
+ protected boolean hasCandidateResult() {
+ return hasFinalResult && n > 0;
+ }
+
+ @Override
+ public Double getResult() {
+ return hasCandidateResult() ? getDoubleValue() : null;
+ }
+
+ @Override
+ public void updateResultFromStatistics(Statistics statistics) {
+ switch (statistics.getType()) {
+ case INT32:
+ case INT64:
+ case FLOAT:
+ case DOUBLE:
+ break;
+ case TEXT:
+ case BOOLEAN:
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format(
+ "Unsupported data type in aggregation MEDIAN : %s", statistics.getType()));
+ }
+ n += statistics.getCount();
+ DataMinV = Math.min(DataMinV, (double) statistics.getMinValue());
+ DataMaxV = Math.max(DataMaxV, (double) statistics.getMaxValue());
+ }
+
+ @Override
+ public void updateResultFromPageData(IBatchDataIterator batchIterator) {
+ updateResultFromPageData(batchIterator, Long.MIN_VALUE, Long.MAX_VALUE);
+ }
+
+ @Override
+ public void updateResultFromPageData(
+ IBatchDataIterator batchIterator, long minBound, long maxBound) {
+ while (batchIterator.hasNext()) {
+ if (batchIterator.currentTime() >= maxBound || batchIterator.currentTime() < minBound) {
+ break;
+ }
+ updateStatusFromData(batchIterator.currentValue());
+ batchIterator.next();
+ }
+ }
+
+ @Override
+ public void updateResultUsingTimestamps(
+ long[] timestamps, int length, IReaderByTimestamp dataReader) throws IOException {
+ Object[] values = dataReader.getValuesInTimestamps(timestamps, length);
+ for (int i = 0; i < length; i++) {
+ if (values[i] != null) {
+ updateStatusFromData(values[i]);
+ }
+ }
+ }
+
+ @Override
+ public void updateResultUsingValues(long[] timestamps, int length, ValueIterator valueIterator) {
+ // List<Object> tmp = new ArrayList<>();
+ while (valueIterator.hasNext()) {
+ updateStatusFromData(valueIterator.next());
+ // Object tmpObj = valueIterator.next();
+ // updateStatusFromData(tmpObj, 1);
+ // tmp.add(tmpObj);
+ }
+ //
+ // System.out.println("\t\t[MEDIAN]"+this.hashCode()+"[updateResultUsingValues]"+tmp.toString());
+ }
+
+ @Override
+ public int maxIteration() {
+ return 10;
+ }
+
+ @Override
+ public boolean hasFinalResult() {
+ return hasFinalResult;
+ }
+
+ @Override
+ public void merge(AggregateResult another) {
+ // System.out.println("[DEBUG] [merge] " + this.getResult() + " " + another.getResult());
+ // merge not supported
+ // throw new QueryProcessException("Can't merge MedianAggregateResult");
+ }
+
+ @Override
+ protected void deserializeSpecificFields(ByteBuffer buffer) {
+ this.seriesDataType = TSDataType.deserialize(buffer.get());
+ // TODO
+ }
+
+ @Override
+ protected void serializeSpecificFields(OutputStream outputStream) throws IOException {
+ ReadWriteIOUtils.write(seriesDataType, outputStream);
+ // TODO
+ }
+
+ public long getN() {
+ return n;
+ }
+
+ @Override
+ public void reset() {
+ super.reset();
+ sketch = null;
+ mad = median = memory = 0;
+ n = 0;
+ iteration = 0;
+ hasFinalResult = false;
+ preComputedSketchSize = 0;
+ }
+
+ @Override
+ public boolean canUpdateFromStatistics(Statistics statistics) {
+ return iteration == 0;
+ }
+
+ @Override
+ public boolean groupByLevelBeforeAggregation() {
+ return true;
+ }
+
+ @Override
+ public boolean useStatisticsIfPossible() {
+ return false;
+ }
+
+ @Override
+ public void setAttributes(Map<String, String> attrs) {
+ if (attrs.containsKey("bucket")) {
+ String b = attrs.get("bucket");
+ this.BUCKET = Integer.parseInt(b);
+ }
+ if (attrs.containsKey("epsilon")) {
+ String b = attrs.get("epsilon");
+ this.EPSILON = Double.parseDouble(b);
+ }
+ if (attrs.containsKey("return_type")) {
+ String q = attrs.get("return_type");
+ this.returnType = q;
+ }
+ System.out.println(" [setAttributes DEBUG]\t\t\tbucket:" + BUCKET + "\t\tepsilon:" + EPSILON);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MADQSAggrResult.java b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MADQSAggrResult.java
new file mode 100644
index 00000000000..42bc48a6512
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/aggregation/impl/MADQSAggrResult.java
@@ -0,0 +1,311 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.query.aggregation.impl;
+
+import org.apache.iotdb.db.query.aggregation.AggregateResult;
+import org.apache.iotdb.db.query.aggregation.AggregationType;
+import org.apache.iotdb.db.query.reader.series.IReaderByTimestamp;
+import org.apache.iotdb.db.utils.ValueIterator;
+import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
+import org.apache.iotdb.tsfile.read.common.IBatchDataIterator;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
+import it.unimi.dsi.util.XoRoShiRo128PlusRandom;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+import static org.apache.iotdb.tsfile.file.metadata.enums.TSDataType.DOUBLE;
+
+public class MADQSAggrResult extends AggregateResult {
+ private String returnType = "value";
+ private TSDataType seriesDataType;
+ private int iteration;
+ private double median, memory = 0, mad;
+ private long n;
+ private DoubleArrayList allData;
+ private boolean hasFinalResult;
+ long DEBUG = 0;
+ public static XoRoShiRo128PlusRandom random = new XoRoShiRo128PlusRandom();
+
+ private int getBitsOfDataType() {
+ switch (seriesDataType) {
+ case INT32:
+ case FLOAT:
+ return 32;
+ case INT64:
+ case DOUBLE:
+ return 64;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type in aggregation MEDIAN : %s", seriesDataType));
+ }
+ }
+
+ public MADQSAggrResult(TSDataType seriesDataType) throws UnSupportedDataTypeException {
+ super(DOUBLE, AggregationType.MAD_QS);
+ this.seriesDataType = seriesDataType;
+ reset();
+ }
+
+ private long dataToLong(Object data) throws UnSupportedDataTypeException {
+ long result;
+ switch (seriesDataType) {
+ case INT32:
+ return (int) data;
+ case FLOAT:
+ result = Float.floatToIntBits((float) data);
+ return (float) data >= 0f ? result : result ^ Long.MAX_VALUE;
+ case INT64:
+ return (long) data;
+ case DOUBLE:
+ result = Double.doubleToLongBits((double) data);
+ return (double) data >= 0d ? result : result ^ Long.MAX_VALUE;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type in aggregation MEDIAN : %s", seriesDataType));
+ }
+ }
+
+ private double longToResult(long result) throws UnSupportedDataTypeException {
+ switch (seriesDataType) {
+ case INT32:
+ return (double) (result);
+ case FLOAT:
+ result = (result >>> 31) == 0 ? result : result ^ Long.MAX_VALUE;
+ return Float.intBitsToFloat((int) (result));
+ case INT64:
+ return (double) (result);
+ case DOUBLE:
+ result = (result >>> 63) == 0 ? result : result ^ Long.MAX_VALUE;
+ return Double.longBitsToDouble(result);
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type in aggregation MEDIAN : %s", seriesDataType));
+ }
+ }
+
+ private void updateStatusFromData(Object data) {
+ double dataD = (double) data;
+ if (iteration == 0) {
+ n++;
+ allData.add(dataD);
+ } else {
+ allData.add(dataD - median);
+ }
+ }
+
+ @Override
+ public void startIteration() {
+ if (iteration == 0) { // first iteration
+ allData = new DoubleArrayList();
+ n = 0;
+ } else {
+ allData.clear();
+ }
+ }
+
+ @Override
+ public void setDoubleValue(double doubleValue) {
+ this.hasCandidateResult = true;
+ if (this.returnType.equals("mad")) this.doubleValue = mad;
+ else if (this.returnType.equals("memory")) this.doubleValue = memory;
+ else if (this.returnType.equals("median")) this.doubleValue = median;
+ }
+
+ @Override
+ public void finishIteration() {
+ iteration++;
+ if (n == 0) {
+ setDoubleValue(0);
+ hasFinalResult = true;
+ return;
+ }
+
+ if (iteration == 1) { // first iteration over
+ median = getKth(allData, 0, (int) n, (int) (n - 1) / 2);
+ return;
+ }
+ if (iteration == 2) { // find median
+ memory = allData.size() / 1024.0 * 8.0;
+ mad = getKth(allData, 0, (int) n, (int) (n - 1) / 2);
+ setDoubleValue(mad);
+ hasFinalResult = true;
+ }
+ }
+
+ public static double getKth(DoubleArrayList data, int L, int R, int K) {
+ int pos = L + random.nextInt(R - L);
+ double pivot_v = data.getDouble(pos), swap_v;
+
+ int leP = L, eqR = R;
+ data.set(pos, data.set(--eqR, pivot_v)); // [L,leP): < pivot_v ; [eqR,R): == pivot_v ;
+
+ for (int i = L; i < eqR; i++)
+ if ((swap_v = data.getDouble(i)) < pivot_v) data.set(i, data.set(leP++, swap_v));
+ else if (swap_v == pivot_v) {
+ data.set(i--, data.set(--eqR, swap_v));
+ }
+
+ // if(R-eqR>1)System.out.println("\t\t\t\tk_select. same pivot v. count:"+(R-eqR));
+ if (K < leP - L) return getKth(data, L, leP, K);
+ if (K >= (leP - L) + (R - eqR)) return getKth(data, leP, eqR, K - (leP - L) - (R - eqR));
+ return pivot_v;
+ }
+
+ @Override
+ protected boolean hasCandidateResult() {
+ return hasFinalResult && n > 0;
+ }
+
+ @Override
+ public Double getResult() {
+ return hasCandidateResult() ? getDoubleValue() : null;
+ }
+
+ @Override
+ public void updateResultFromStatistics(Statistics statistics) {
+ switch (statistics.getType()) {
+ case INT32:
+ case INT64:
+ case FLOAT:
+ case DOUBLE:
+ break;
+ case TEXT:
+ case BOOLEAN:
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format(
+ "Unsupported data type in aggregation MEDIAN : %s", statistics.getType()));
+ }
+ }
+
+ @Override
+ public void updateResultFromPageData(IBatchDataIterator batchIterator) {
+ updateResultFromPageData(batchIterator, Long.MIN_VALUE, Long.MAX_VALUE);
+ }
+
+ @Override
+ public void updateResultFromPageData(
+ IBatchDataIterator batchIterator, long minBound, long maxBound) {
+ while (batchIterator.hasNext()) {
+ if (batchIterator.currentTime() >= maxBound || batchIterator.currentTime() < minBound) {
+ break;
+ }
+ updateStatusFromData(batchIterator.currentValue());
+ batchIterator.next();
+ }
+ }
+
+ @Override
+ public void updateResultUsingTimestamps(
+ long[] timestamps, int length, IReaderByTimestamp dataReader) throws IOException {
+ Object[] values = dataReader.getValuesInTimestamps(timestamps, length);
+ for (int i = 0; i < length; i++) {
+ if (values[i] != null) {
+ updateStatusFromData(values[i]);
+ }
+ }
+ }
+
+ @Override
+ public void updateResultUsingValues(long[] timestamps, int length, ValueIterator valueIterator) {
+ // List<Object> tmp = new ArrayList<>();
+ while (valueIterator.hasNext()) {
+ updateStatusFromData(valueIterator.next());
+ // Object tmpObj = valueIterator.next();
+ // updateStatusFromData(tmpObj, 1);
+ // tmp.add(tmpObj);
+ }
+ //
+ // System.out.println("\t\t[MEDIAN]"+this.hashCode()+"[updateResultUsingValues]"+tmp.toString());
+ }
+
+ @Override
+ public int maxIteration() {
+ return 10;
+ }
+
+ @Override
+ public boolean hasFinalResult() {
+ return hasFinalResult;
+ }
+
+ @Override
+ public void merge(AggregateResult another) {
+ // System.out.println("[DEBUG] [merge] " + this.getResult() + " " + another.getResult());
+ // merge not supported
+ // throw new QueryProcessException("Can't merge MedianAggregateResult");
+ }
+
+ @Override
+ protected void deserializeSpecificFields(ByteBuffer buffer) {
+ this.seriesDataType = TSDataType.deserialize(buffer.get());
+ // TODO
+ }
+
+ @Override
+ protected void serializeSpecificFields(OutputStream outputStream) throws IOException {
+ ReadWriteIOUtils.write(seriesDataType, outputStream);
+ // TODO
+ }
+
+ public long getN() {
+ return n;
+ }
+
+ @Override
+ public void reset() {
+ super.reset();
+ mad = median = memory = 0;
+ n = 0;
+ iteration = 0;
+ hasFinalResult = false;
+ }
+
+ @Override
+ public boolean canUpdateFromStatistics(Statistics statistics) {
+ return false;
+ }
+
+ @Override
+ public boolean groupByLevelBeforeAggregation() {
+ return true;
+ }
+
+ @Override
+ public boolean useStatisticsIfPossible() {
+ return false;
+ }
+
+ @Override
+ public void setAttributes(Map<String, String> attrs) {
+ if (attrs.containsKey("return_type")) {
+ String q = attrs.get("return_type");
+ this.returnType = q;
+ }
+ System.out.println(" [setAttributes DEBUG]\tQuickSelect return_type:\t" + returnType);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/factory/AggregateResultFactory.java b/server/src/main/java/org/apache/iotdb/db/query/factory/AggregateResultFactory.java
index 27022a56137..13f118e885a 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/factory/AggregateResultFactory.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/factory/AggregateResultFactory.java
@@ -161,6 +161,12 @@ public class AggregateResultFactory {
return new ExactMultiQuantilesTDigestAggrResult(dataType);
case SQLConstant.EXACT_MULTI_QUANTILES_DDSKETCH_POSITIVE:
return new ExactMultiQuantilesDDSketchAggrResult(dataType);
+ case SQLConstant.MAD_DD:
+ return new MADDDAggrResult(dataType);
+ case SQLConstant.MAD_CORE:
+ return new MADCoreAggrResult(dataType);
+ case SQLConstant.MAD_QS:
+ return new MADQSAggrResult(dataType);
default:
throw new IllegalArgumentException("Invalid Aggregation function: " + aggrFuncName);
}
@@ -286,6 +292,12 @@ public class AggregateResultFactory {
return new ExactMultiQuantilesTDigestAggrResult(dataType);
case SQLConstant.EXACT_MULTI_QUANTILES_DDSKETCH_POSITIVE:
return new ExactMultiQuantilesDDSketchAggrResult(dataType);
+ case SQLConstant.MAD_DD:
+ return new MADDDAggrResult(dataType);
+ case SQLConstant.MAD_CORE:
+ return new MADCoreAggrResult(dataType);
+ case SQLConstant.MAD_QS:
+ return new MADQSAggrResult(dataType);
default:
throw new IllegalArgumentException("Invalid Aggregation function: " + aggrFuncName);
}
@@ -412,6 +424,12 @@ public class AggregateResultFactory {
return new ExactMultiQuantilesTDigestAggrResult(dataType);
case EXACT_MULTI_QUANTILES_DDSKETCH_POSITIVE:
return new ExactMultiQuantilesDDSketchAggrResult(dataType);
+ case MAD_DD:
+ return new MADDDAggrResult(dataType);
+ case MAD_CORE:
+ return new MADCoreAggrResult(dataType);
+ case MAD_QS:
+ return new MADQSAggrResult(dataType);
default:
throw new IllegalArgumentException("Invalid Aggregation Type: " + aggregationType.name());
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/pool/QueryTaskManager.java b/server/src/main/java/org/apache/iotdb/db/query/pool/QueryTaskManager.java
index f251e28fc98..0e621acdc78 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/pool/QueryTaskManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/pool/QueryTaskManager.java
@@ -42,6 +42,7 @@ public class QueryTaskManager extends AbstractPoolManager {
Math.min(
Runtime.getRuntime().availableProcessors(),
IoTDBDescriptor.getInstance().getConfig().getConcurrentQueryThread());
+ threadCnt = IoTDBDescriptor.getInstance().getConfig().getMADQueryThread();
pool = IoTDBThreadPoolFactory.newFixedThreadPool(threadCnt, ThreadName.QUERY_SERVICE.getName());
}
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
index 2a31bc363d3..97aa2421219 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
@@ -239,6 +239,9 @@ public class SchemaUtils {
case SQLConstant.EXACT_MULTI_QUANTILES_MRL:
case SQLConstant.EXACT_MULTI_QUANTILES_TDIGEST:
case SQLConstant.EXACT_MULTI_QUANTILES_DDSKETCH_POSITIVE:
+ case SQLConstant.MAD_DD:
+ case SQLConstant.MAD_CORE:
+ case SQLConstant.MAD_QS:
return TSDataType.DOUBLE;
case SQLConstant.LAST_VALUE:
case SQLConstant.FIRST_VALUE:
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/TypeInferenceUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/TypeInferenceUtils.java
index ef25c896b19..05778c95f88 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/TypeInferenceUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/TypeInferenceUtils.java
@@ -168,6 +168,9 @@ public class TypeInferenceUtils {
case SQLConstant.EXACT_MULTI_QUANTILES_MRL:
case SQLConstant.EXACT_MULTI_QUANTILES_TDIGEST:
case SQLConstant.EXACT_MULTI_QUANTILES_DDSKETCH_POSITIVE:
+ case SQLConstant.MAD_DD:
+ case SQLConstant.MAD_CORE:
+ case SQLConstant.MAD_QS:
return TSDataType.DOUBLE;
default:
throw new IllegalArgumentException("Invalid Aggregation function: " + aggrFuncName);
diff --git a/session/src/test/java/org/apache/iotdb/session/InsertCsvDataIT.java b/session/src/test/java/org/apache/iotdb/session/InsertCsvDataIT.java
index 68a6ca60608..53a7c9184f7 100644
--- a/session/src/test/java/org/apache/iotdb/session/InsertCsvDataIT.java
+++ b/session/src/test/java/org/apache/iotdb/session/InsertCsvDataIT.java
@@ -197,17 +197,15 @@ public class InsertCsvDataIT {
private static void insertDataFromTXT()
throws IoTDBConnectionException, StatementExecutionException, IOException {
- final int TEST_CASE = 10;
+ final int TEST_CASE = 1;
String[] fileList = new String[10], sgName = new String[10];
- String sketch_size = "256Byte";
- fileList[0] = "1_bitcoin.csv";
+ String sketch_size = "";
+ fileList[0] = "bitcoin-s.csv";
sgName[0] = "root.bitcoin" + sketch_size;
- fileList[1] = "2_SpacecraftThruster.txt";
- sgName[1] = "root.thruster" + sketch_size;
- fileList[2] = "3_taxipredition8M.txt";
- sgName[2] = "root.taxi" + sketch_size;
- fileList[3] = "4_wh.csv";
- sgName[3] = "root.wh" + sketch_size;
+ fileList[1] = "gas-s.csv";
+ sgName[1] = "root.gas" + sketch_size;
+ fileList[2] = "power-s.csv";
+ sgName[2] = "root.power" + sketch_size;
// fileList[1] = "tmp_3_55.txt";
// fileList[2] = "tmp_0_55.txt";
// fileList[3] = "tmp_2_55.txt";
@@ -220,7 +218,7 @@ public class InsertCsvDataIT {
System.out.print("\t\t\t");
String filename = fileList[fileID];
- String folder = "D:\\Study\\Lab\\iotdb\\add_quantile_to_aggregation\\test_project_2";
+ String folder = "E:\\MAD-data";
String filepath = folder + "\\" + filename;
DoubleArrayList vv = new DoubleArrayList();
@@ -340,8 +338,8 @@ public class InsertCsvDataIT {
@Test
public void insertDATA() {
try {
- for (int i = 0; i < 10; i++) prepareTimeSeriesData();
- // insertDataFromTXT();
+ // for (int i = 0; i < 10; i++) prepareTimeSeriesData();
+ insertDataFromTXT();
// append(5);
// insertDataFromTXT();
// insertDataFromTXT(3, 3, 0);
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/Executor.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/Executor.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/QueryProcessor.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/QueryProcessor.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/common/BasicOperator.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/common/BasicOperator.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/common/FilterOperator.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/common/FilterOperator.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/common/Operator.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/common/Operator.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/common/SQLConstant.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/common/SQLConstant.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/common/SingleQuery.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/common/SingleQuery.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/common/TSQueryPlan.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/common/TSQueryPlan.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/exception/BasicOperatorException.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/exception/BasicOperatorException.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/exception/DNFOptimizeException.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/exception/DNFOptimizeException.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/exception/LogicalOptimizeException.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/exception/LogicalOptimizeException.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/exception/MergeFilterException.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/exception/MergeFilterException.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/exception/QueryOperatorException.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/exception/QueryOperatorException.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/exception/QueryProcessorException.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/exception/QueryProcessorException.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/exception/RemoveNotException.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/exception/RemoveNotException.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/optimizer/DNFFilterOptimizer.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/optimizer/DNFFilterOptimizer.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/optimizer/IFilterOptimizer.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/optimizer/IFilterOptimizer.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/optimizer/MergeSingleFilterOptimizer.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/optimizer/MergeSingleFilterOptimizer.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/optimizer/PhysicalOptimizer.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/optimizer/PhysicalOptimizer.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/optimizer/RemoveNotOptimizer.java b/spark-tsfile/src/main/java/org/apache/iotdb/spark/tsfile/qp/optimizer/RemoveNotOptimizer.java
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/scala/org/apache/iotdb/spark/tsfile/DefaultSource.scala b/spark-tsfile/src/main/scala/org/apache/iotdb/spark/tsfile/DefaultSource.scala
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/scala/org/apache/iotdb/spark/tsfile/WideConverter.scala b/spark-tsfile/src/main/scala/org/apache/iotdb/spark/tsfile/WideConverter.scala
old mode 100755
new mode 100644
diff --git a/spark-tsfile/src/main/scala/org/apache/iotdb/spark/tsfile/package.scala b/spark-tsfile/src/main/scala/org/apache/iotdb/spark/tsfile/package.scala
old mode 100755
new mode 100644
diff --git a/test/e2e/cases/cli/cleanup.sh b/test/e2e/cases/cli/cleanup.sh
old mode 100755
new mode 100644
diff --git a/test/e2e/cases/cli/run.sh b/test/e2e/cases/cli/run.sh
old mode 100755
new mode 100644
diff --git a/thrift-sync/src/main/thrift/sync.thrift b/thrift-sync/src/main/thrift/sync.thrift
old mode 100755
new mode 100644
diff --git a/tools/check_sigs.sh b/tools/check_sigs.sh
old mode 100755
new mode 100644
diff --git a/tools/common.sh b/tools/common.sh
old mode 100755
new mode 100644
diff --git a/tools/download_staged_release.sh b/tools/download_staged_release.sh
old mode 100755
new mode 100644
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/basic/BinaryFilter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/basic/BinaryFilter.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/basic/Filter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/basic/Filter.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/basic/UnaryFilter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/basic/UnaryFilter.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/AndFilter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/AndFilter.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/Eq.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/Eq.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/Gt.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/Gt.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/GtEq.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/GtEq.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/Lt.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/Lt.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/LtEq.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/LtEq.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/NotEq.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/NotEq.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/NotFilter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/NotFilter.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/OrFilter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/OrFilter.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/CORESketch.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/CORESketch.java
new file mode 100644
index 00000000000..1dd387f4a9e
--- /dev/null
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/CORESketch.java
@@ -0,0 +1,414 @@
+package org.apache.iotdb.tsfile.utils;
+
+import it.unimi.dsi.fastutil.longs.Long2LongMap;
+import it.unimi.dsi.fastutil.longs.Long2LongOpenHashMap;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.Comparator;
+
+import static org.apache.iotdb.tsfile.utils.FileHelper.GET_SIZE;
+
+public class CORESketch implements Serializable {
+ public int card;
+ public double power_2_card, frac__power_2_card__log_2;
+ public static final double log_2 = Math.log(2);
+ public int max_bucket;
+ public Long2LongOpenHashMap buckets;
+ public transient double[] useful_range;
+ public long data_size;
+ public long pruned_size;
+ private long left_count;
+ private long right_count;
+ private int m, l, r;
+
+ private transient long[] mid_num;
+ private transient Bucket[] bucket_sequence;
+
+ private static class Bucket {
+ long index;
+ long count;
+ double lower_bound;
+ double upper_bound;
+ boolean finest;
+
+ Bucket(long index, long count, double lower_bound, double upper_bound, boolean finest) {
+ this.index = index;
+ this.count = count;
+ this.lower_bound = lower_bound;
+ this.upper_bound = upper_bound;
+ this.finest = finest;
+ }
+ }
+
+ public CORESketch(int bucket) {
+ this.card = 0;
+ power_2_card = Math.pow(2, card);
+ frac__power_2_card__log_2 = power_2_card / log_2;
+ this.max_bucket = bucket;
+ this.useful_range = new double[6];
+ this.buckets = new Long2LongOpenHashMap(this.max_bucket);
+ this.mid_num = new long[4];
+ this.bucket_sequence = new Bucket[(int) max_bucket];
+ this.left_count = 0;
+ this.right_count = 0;
+ this.m = 0;
+ this.l = 0;
+ this.r = 0;
+ }
+
+ public CORESketch(int card, int max_bucket, double[] useful_range) {
+ this.card = card;
+ power_2_card = Math.pow(2, card);
+ frac__power_2_card__log_2 = power_2_card / log_2;
+ this.max_bucket = max_bucket;
+ this.buckets = new Long2LongOpenHashMap(this.max_bucket);
+ this.useful_range = new double[6];
+ System.arraycopy(useful_range, 0, this.useful_range, 0, 6);
+ this.mid_num = new long[4];
+ this.bucket_sequence = new Bucket[(int) max_bucket];
+ this.left_count = 0;
+ this.right_count = 0;
+ this.m = 0;
+ this.l = 0;
+ this.r = 0;
+ }
+
+ public boolean in_range(double v, double[] range) {
+ return (v > range[0] && v < range[1])
+ || (v >= range[2] && v < range[3])
+ || (v > range[4] && v < range[5]);
+ }
+
+ public void insert_mid(double v) {
+ if (v <= useful_range[0]) {
+ mid_num[0] += 1;
+ } else if (v >= useful_range[1] && v < useful_range[2]) {
+ mid_num[1] += 1;
+ } else if (v >= useful_range[3] && v <= useful_range[4]) {
+ mid_num[2] += 1;
+ } else if (v >= useful_range[5]) {
+ mid_num[3] += 1;
+ }
+ }
+
+ public double calculate_maximum_space() {
+ for (long i = 0; i < max_bucket; i++) {
+ buckets.put(i, 1000000 + i);
+ }
+ serialize_bucket();
+ System.out.println(GET_SIZE(buckets));
+ return GET_SIZE(this);
+ }
+
+ public double customize(double v) {
+ if (v < useful_range[0]) {
+ return useful_range[0];
+ } else if (v > useful_range[1] && v < useful_range[2]) {
+ return useful_range[2];
+ } else if (v > useful_range[3] && v < useful_range[4]) {
+ return useful_range[4];
+ } else if (v > useful_range[5]) {
+ return useful_range[5];
+ }
+ return v;
+ }
+
+ public void insert(double v) {
+ data_size += 1;
+ if (!in_range(v, useful_range)) {
+ pruned_size += 1;
+ insert_mid(v);
+ v = customize(v);
+ }
+ long i = (long) Math.ceil(frac__power_2_card__log_2 * Math.log(v));
+ buckets.put(i, buckets.getOrDefault(i, 0L) + 1);
+ }
+
+ public void serialize_bucket() {
+ int i = 0;
+ double zeta = Math.pow(2, Math.pow(2, -card));
+ // zeta=zeta+(zeta-1)/2;
+ bucket_sequence = new Bucket[buckets.size()];
+
+ for (Long2LongMap.Entry e : buckets.long2LongEntrySet()) {
+ double lb = Math.pow(zeta, e.getLongKey() - 1);
+ double ub = Math.pow(zeta, e.getLongKey());
+ boolean finest = Math.ceil(lb) == Math.floor(ub);
+ bucket_sequence[i++] = new Bucket(e.getLongKey(), e.getLongValue(), lb, ub, finest);
+ }
+ Arrays.sort(bucket_sequence, Comparator.comparingLong(o -> o.index));
+ }
+
+ public int mid_half_count_bucket() {
+ serialize_bucket();
+ long count = 0;
+ double rank = 0.5 * (data_size - 1);
+ for (int m = 0; m < bucket_sequence.length; ++m) {
+ count += bucket_sequence[m].count;
+ if (count > rank) {
+ this.m = m;
+ return m;
+ }
+ }
+ System.out.println(
+ "\t\t\t\tall_bucket_tot_size="
+ + count
+ + "\t\t\tdata_size/2:"
+ + rank
+ + "\t\tprunedSize="
+ + pruned_size);
+ this.m = -1;
+ return -1;
+ }
+
+ public int[] edge_half_count_bucket(int m) {
+ long count = bucket_sequence[m].count;
+ int cursor = m;
+ double rank = 0.5 * data_size;
+ int l = m - 1;
+ int r = m + 1;
+ left_count += bucket_sequence[m].count;
+ right_count += bucket_sequence[m].count;
+ while (count <= rank && l >= 0 && r < bucket_sequence.length) {
+ if (bucket_sequence[m].lower_bound - bucket_sequence[l].upper_bound
+ < bucket_sequence[r].lower_bound - bucket_sequence[m].upper_bound) {
+ cursor = l--;
+ left_count += bucket_sequence[cursor].count;
+ } else {
+ cursor = r++;
+ right_count += bucket_sequence[cursor].count;
+ }
+ count += bucket_sequence[cursor].count;
+ }
+
+ while (count <= rank && l >= 0) {
+ cursor = l--;
+ left_count += bucket_sequence[cursor].count;
+ count += bucket_sequence[cursor].count;
+ }
+
+ while (count <= rank && r < bucket_sequence.length) {
+ cursor = r++;
+ right_count += bucket_sequence[cursor].count;
+ count += bucket_sequence[cursor].count;
+ }
+ r -= 1;
+ l += 1;
+ if (r == bucket_sequence.length - 1 && cursor != r) r = -1;
+ if (l == 0 && cursor != l) l = -1;
+ this.l = l;
+ this.r = r;
+ return new int[] {l, r};
+ }
+
+ public double[] real_range() {
+ double[] real_range = new double[7];
+ real_range[6] = 3;
+ real_range[0] = useful_range[0];
+ real_range[1] = useful_range[1];
+ int cursor = 2;
+ if (useful_range[2] < real_range[cursor - 1]) {
+ real_range[cursor - 1] = Math.max(useful_range[3], real_range[cursor - 1]);
+ real_range[6] -= 1;
+ } else {
+ real_range[cursor] = useful_range[2];
+ real_range[cursor + 1] = useful_range[3];
+ cursor += 2;
+ }
+ if (useful_range[4] < real_range[cursor - 1]) {
+ real_range[cursor - 1] = Math.max(useful_range[5], real_range[cursor - 1]);
+ real_range[6] -= 1;
+ } else {
+ real_range[cursor] = useful_range[4];
+ real_range[cursor + 1] = useful_range[5];
+ }
+ return real_range;
+ }
+
+ public void half_count_buckets() {
+ int m = mid_half_count_bucket();
+ int[] lr = edge_half_count_bucket(m);
+ }
+
+ public double[] generate_useful_range(int m, int l, int r) {
+ double[] new_range = new double[6];
+ if (l == -1) {
+ new_range[0] =
+ Math.max(
+ useful_range[0], 2 * bucket_sequence[m].lower_bound - bucket_sequence[r].upper_bound);
+ new_range[1] =
+ Math.min(
+ useful_range[1], 2 * bucket_sequence[m].upper_bound - bucket_sequence[r].lower_bound);
+ new_range[2] = Math.max(useful_range[2], bucket_sequence[m].lower_bound);
+ new_range[3] = Math.min(useful_range[3], bucket_sequence[m].upper_bound);
+ new_range[4] =
+ Math.max(
+ useful_range[4],
+ bucket_sequence[m].lower_bound
+ + bucket_sequence[r].lower_bound
+ - bucket_sequence[m].upper_bound);
+ new_range[5] =
+ Math.min(
+ useful_range[5],
+ bucket_sequence[m].upper_bound
+ + bucket_sequence[r].upper_bound
+ - bucket_sequence[m].lower_bound);
+ }
+ if (r == -1) {
+ new_range[0] =
+ Math.max(
+ useful_range[0],
+ bucket_sequence[m].lower_bound
+ - bucket_sequence[m].upper_bound
+ + bucket_sequence[l].lower_bound);
+ new_range[1] =
+ Math.min(
+ useful_range[1],
+ bucket_sequence[m].upper_bound
+ - bucket_sequence[m].lower_bound
+ + bucket_sequence[l].upper_bound);
+ new_range[2] = Math.max(useful_range[2], bucket_sequence[m].lower_bound);
+ new_range[3] = Math.min(useful_range[3], bucket_sequence[m].upper_bound);
+ new_range[4] =
+ Math.max(
+ useful_range[4], 2 * bucket_sequence[m].lower_bound - bucket_sequence[l].upper_bound);
+ new_range[5] =
+ Math.min(
+ useful_range[5], 2 * bucket_sequence[m].upper_bound - bucket_sequence[l].lower_bound);
+ }
+ new_range[0] =
+ Math.max(
+ useful_range[0],
+ Math.min(
+ bucket_sequence[m].lower_bound
+ - bucket_sequence[m].upper_bound
+ + bucket_sequence[l].lower_bound,
+ 2 * bucket_sequence[m].lower_bound - bucket_sequence[r].upper_bound));
+ new_range[1] =
+ Math.min(
+ useful_range[1],
+ Math.max(
+ bucket_sequence[m].upper_bound
+ - bucket_sequence[m].lower_bound
+ + bucket_sequence[l].upper_bound,
+ 2 * bucket_sequence[m].upper_bound - bucket_sequence[r].lower_bound));
+ new_range[2] = Math.max(useful_range[2], bucket_sequence[m].lower_bound);
+ new_range[3] = Math.min(useful_range[3], bucket_sequence[m].upper_bound);
+ new_range[4] =
+ Math.max(
+ useful_range[4],
+ Math.min(
+ 2 * bucket_sequence[m].lower_bound - bucket_sequence[l].upper_bound,
+ bucket_sequence[m].lower_bound
+ + bucket_sequence[r].lower_bound
+ - bucket_sequence[m].upper_bound));
+ new_range[5] =
+ Math.min(
+ useful_range[5],
+ Math.max(
+ 2 * bucket_sequence[m].upper_bound - bucket_sequence[l].lower_bound,
+ bucket_sequence[m].upper_bound
+ + bucket_sequence[r].upper_bound
+ - bucket_sequence[m].lower_bound));
+ return new_range;
+ }
+
+ public boolean bucket_finest(int m, int l, int r) {
+ return bucket_sequence[m].finest && bucket_sequence[l].finest;
+ }
+
+ public boolean bucket_finest() {
+ return bucket_finest(this.m, this.l, this.r);
+ }
+
+ public double mad(int m, int l, int r) {
+ double choose_right =
+ right_count + left_count - bucket_sequence[m].count - bucket_sequence[l].count;
+ double choose_left =
+ left_count + right_count - bucket_sequence[m].count - bucket_sequence[r].count;
+ if (choose_right < 0.5 * data_size && choose_left >= 0.5 * data_size) {
+ return Math.floor(bucket_sequence[m].upper_bound)
+ - Math.floor(bucket_sequence[l].upper_bound);
+ } else if (choose_right >= 0.5 * data_size && choose_left < 0.5 * data_size) {
+ return Math.floor(bucket_sequence[r].upper_bound)
+ - Math.floor(bucket_sequence[m].upper_bound);
+ } else if (choose_right < 0.5 * data_size && choose_left < 0.5 * data_size) {
+ return Math.max(
+ Math.floor(bucket_sequence[m].upper_bound) - Math.floor(bucket_sequence[l].upper_bound),
+ Math.floor(bucket_sequence[r].upper_bound) - Math.floor(bucket_sequence[m].upper_bound));
+ } else {
+ return Math.min(
+ Math.floor(bucket_sequence[m].upper_bound) - Math.floor(bucket_sequence[l].upper_bound),
+ Math.floor(bucket_sequence[r].upper_bound) - Math.floor(bucket_sequence[m].upper_bound));
+ }
+ }
+
+ public void merge(CORESketch sketch) {
+ int diff = (int) Math.pow(2, Math.abs(card - sketch.card));
+ if (card <= sketch.card) {
+ for (Long2LongMap.Entry e : sketch.buckets.long2LongEntrySet()) {
+ buckets.put(
+ e.getLongKey(), e.getLongValue() + buckets.getOrDefault(e.getLongKey() / diff, 0L));
+ }
+ } else {
+ for (Long2LongMap.Entry e : buckets.long2LongEntrySet()) {
+ sketch.buckets.put(
+ e.getLongKey(),
+ e.getLongValue() + sketch.buckets.getOrDefault(e.getLongKey() / diff, 0L));
+ buckets = sketch.buckets;
+ }
+ card = sketch.card;
+ power_2_card = Math.pow(2, card);
+ frac__power_2_card__log_2 = power_2_card / log_2;
+ }
+ useful_range[0] = Math.min(useful_range[0], sketch.useful_range[0]);
+ useful_range[1] = Math.max(useful_range[1], sketch.useful_range[1]);
+ useful_range[2] = Math.min(useful_range[2], sketch.useful_range[2]);
+ useful_range[3] = Math.max(useful_range[3], sketch.useful_range[3]);
+ useful_range[4] = Math.min(useful_range[4], sketch.useful_range[4]);
+ useful_range[5] = Math.max(useful_range[5], sketch.useful_range[5]);
+ data_size = data_size + sketch.data_size;
+ pruned_size = pruned_size + sketch.pruned_size;
+ }
+
+ public double mad() {
+ return mad(this.m, this.l, this.r);
+ }
+
+ public long useful_count() {
+ return data_size - pruned_size;
+ }
+
+ public int get_bucket_size() {
+ return buckets.size();
+ }
+
+ public boolean data_read(boolean parallel, int threads) {
+ // if (parallel) return useful_count() * 8 * threads <= max_bucket * (32.125 + 16);
+ // else
+ return useful_count() * 8 <= max_bucket * (32.125 + 16);
+ }
+
+ public boolean data_read_norm() {
+ return useful_count() * 8 <= max_bucket * (32.125 + 16);
+ }
+
+ public double[] get_range() {
+ return useful_range;
+ }
+
+ public void set_range(double[] range) {
+ useful_range = range;
+ }
+
+ public long[] get_gap() {
+ return mid_num;
+ }
+
+ public void set_card(int card) {
+ this.card = card;
+ power_2_card = Math.pow(2, card);
+ frac__power_2_card__log_2 = power_2_card / log_2;
+ }
+}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/DDSketch.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/DDSketch.java
new file mode 100644
index 00000000000..53077d59114
--- /dev/null
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/DDSketch.java
@@ -0,0 +1,174 @@
+package org.apache.iotdb.tsfile.utils;
+
+import it.unimi.dsi.fastutil.ints.Int2LongOpenHashMap;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.Map;
+
+import static org.apache.commons.math3.util.Precision.EPSILON;
+import static org.apache.iotdb.tsfile.utils.Params.*;
+
+public class DDSketch implements Serializable {
+ private double alpha;
+ private double gamma;
+ private double multiplier;
+ private int bucket_num_limit;
+ private int threshold_for_compression;
+
+ private Int2LongOpenHashMap positive_buckets;
+ private Int2LongOpenHashMap negative_buckets;
+ private double collapse_bound;
+ private long zero_count;
+
+ private transient double beta;
+ private final transient double[] valid_range;
+
+ private static double MIN_POSITIVE_VALUE = FS_MIN_POSITIVE_VALUE;
+ private static double COEFFICIENT = FS_COMPRESSION_COEFFICIENT;
+
+ public DDSketch() {
+ this(EPSILON, BUCKET_NUM_LIMIT);
+ }
+
+ public DDSketch(double alpha, int bucket_num_limit) {
+ // System.out.println(alpha);
+ this.alpha = alpha;
+ this.bucket_num_limit = Math.max(bucket_num_limit, 2);
+ this.threshold_for_compression = (int) (bucket_num_limit * COEFFICIENT);
+
+ this.gamma = 2 * alpha / (1 - alpha) + 1;
+ this.multiplier = Math.log(Math.E) / (Math.log1p(gamma - 1));
+ this.positive_buckets = new Int2LongOpenHashMap((int) (bucket_num_limit * 0.75));
+ this.negative_buckets = new Int2LongOpenHashMap((int) (bucket_num_limit * 0.25));
+ this.zero_count = 0;
+ this.collapse_bound = -Double.MAX_VALUE;
+ this.valid_range = new double[6];
+ }
+
+ public void insert(double v) {
+ if (v < collapse_bound) {
+ v = collapse_bound;
+ }
+ if (v > MIN_POSITIVE_VALUE) {
+ int i = (int) Math.ceil(Math.log(v) * multiplier);
+ positive_buckets.put(i, positive_buckets.getOrDefault(i, 0L) + 1);
+ } else if (v < -MIN_POSITIVE_VALUE) {
+ int i = (int) Math.ceil(Math.log(-v) * multiplier);
+ negative_buckets.put(i, negative_buckets.getOrDefault(i, 0L) + 1);
+ } else {
+ zero_count++;
+ }
+ collapse(threshold_for_compression);
+ }
+
+ private void collapse(int limit) {
+ if (sketch_size() > limit) {
+ int exceed = sketch_size() - bucket_num_limit;
+ int[] indices = negative_buckets.keySet().toIntArray();
+ Arrays.sort(indices);
+ long count = 0;
+ for (int i = Math.max(0, indices.length - exceed); i < indices.length; ++i) {
+ count += negative_buckets.remove(indices[i]);
+ }
+ if (count > 0) {
+ int i = indices.length - exceed - 1;
+ if (i >= 0) {
+ negative_buckets.put(indices[i], negative_buckets.get(indices[i]) + count);
+ collapse_bound = -Math.pow(gamma, indices[i]);
+ } else {
+ zero_count += count;
+ collapse_bound = 0;
+ }
+ }
+ exceed -= (indices.length - Math.max(0, indices.length - exceed));
+ if (exceed > 0) {
+ count = zero_count;
+ if (zero_count > 0) {
+ exceed--;
+ }
+ indices = positive_buckets.keySet().toIntArray();
+ Arrays.sort(indices);
+ for (int i = exceed - 1; i >= 0; --i) {
+ count += positive_buckets.remove(indices[i]);
+ }
+ positive_buckets.put(indices[exceed], positive_buckets.get(indices[exceed]) + count);
+ collapse_bound = Math.pow(gamma, indices[exceed] - 1);
+ }
+ }
+ }
+
+ private Bucket[] union_buckets() {
+ Bucket[] buckets = new Bucket[sketch_size()];
+ int i = 0;
+ for (Map.Entry<Integer, Long> e : positive_buckets.entrySet()) {
+ buckets[i++] =
+ new Bucket(
+ e.getKey(),
+ Math.pow(gamma, e.getKey() - 1),
+ Math.pow(gamma, e.getKey()),
+ e.getValue());
+ }
+ for (Map.Entry<Integer, Long> e : negative_buckets.entrySet()) {
+ buckets[i++] =
+ new Bucket(
+ e.getKey(),
+ -Math.pow(gamma, e.getKey()),
+ -Math.pow(gamma, e.getKey() - 1),
+ e.getValue());
+ }
+ if (zero_count > 0) {
+ buckets[i] = new Bucket(0, 0, 0, zero_count);
+ }
+ Arrays.sort(buckets, Comparator.comparingDouble(o -> o.lower_bound));
+ return buckets;
+ }
+
+ private long total_count() {
+ return positive_buckets.values().stream().mapToLong(l -> l).sum()
+ + negative_buckets.values().stream().mapToLong(l -> l).sum()
+ + zero_count;
+ }
+
+ public int find_p_index(Bucket[] buckets, long total_count, double q) {
+ long count = 0;
+ double rank = q * (total_count - 1);
+ for (int i = 0; i < buckets.length; ++i) {
+ count += buckets[i].count;
+ if (count > rank) {
+ return i;
+ }
+ }
+ return -1;
+ }
+
+ public double getQuantile(double q) {
+ Bucket[] buckets = union_buckets();
+ long total_count = total_count();
+ Bucket p = buckets[find_p_index(buckets, total_count, q)];
+ if (p.lower_bound < 0) {
+ return 2 * p.lower_bound / (1 + gamma);
+ } else {
+ return 2 * p.upper_bound / (1 + gamma);
+ }
+ }
+
+ public int sketch_size() {
+ return positive_buckets.size() + negative_buckets.size() + (zero_count == 0 ? 0 : 1);
+ }
+
+ private static class Bucket {
+ int index;
+ double lower_bound;
+ double upper_bound;
+ long count;
+
+ Bucket(int index, double lower_bound, double upper_bound, long count) {
+ this.index = index;
+ this.lower_bound = lower_bound;
+ this.upper_bound = upper_bound;
+ this.count = count;
+ }
+ }
+}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/FileHelper.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/FileHelper.java
new file mode 100644
index 00000000000..7c87ba40742
--- /dev/null
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/FileHelper.java
@@ -0,0 +1,111 @@
+package org.apache.iotdb.tsfile.utils;
+
+import it.unimi.dsi.util.XoRoShiRo128PlusRandom;
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.*;
+import java.util.Arrays;
+
+public class FileHelper {
+ public static BufferedReader READ(String filename) throws IOException {
+ File file = new File(filename);
+ BufferedReader reader =
+ new BufferedReader(new InputStreamReader(new FileInputStream(file)), 50 * 1024 * 1024);
+ // reader.mark((int) (file.length() + 1));
+ return reader;
+ }
+
+ public static double[] READ(String filename, int read_count, int total_count) throws IOException {
+
+ XoRoShiRo128PlusRandom random = new XoRoShiRo128PlusRandom(2333);
+ // Random random = new Random();
+ double threshold = read_count * 1.0 / total_count;
+ double[] nums = new double[read_count];
+ int number = 0;
+ while (number < nums.length) {
+ BufferedReader br = READ(filename);
+ br.readLine();
+ String line;
+ if (threshold < 1.0 - 1e-9)
+ for (; number < nums.length && (line = br.readLine()) != null; ++number) {
+ if (random.nextDoubleFast() < threshold) {
+ nums[number] = Double.parseDouble(line);
+ } else {
+ number--;
+ }
+ }
+ else
+ for (; number < nums.length && (line = br.readLine()) != null; ++number) {
+ nums[number] = Double.parseDouble(line);
+ }
+ br.close();
+ }
+ return nums;
+ }
+
+ public static void COMBINE_FILE(String pattern, String distribution) throws IOException {
+ String filename = String.format(pattern, distribution, 0);
+ BufferedReader br = FileHelper.READ(filename);
+ String[] headers = StringUtils.split(br.readLine(), ',');
+ int cols = headers.length, rows = 0;
+ while (br.readLine() != null) {
+ rows++;
+ }
+ br.close();
+ double[][] cache = new double[rows][cols];
+
+ for (int i = 0; i <= 9; ++i) {
+ filename = String.format(pattern, distribution, i);
+ br = FileHelper.READ(filename);
+ br.readLine();
+
+ for (int r = 0; r < rows; ++r) {
+ Double[] nums =
+ Arrays.stream(br.readLine().split(",", -1))
+ .map(
+ s -> {
+ if (s.isEmpty()) {
+ return 0.0;
+ } else {
+ return Double.parseDouble(s);
+ }
+ })
+ .toArray(Double[]::new);
+ for (int c = 0; c < cols; ++c) {
+ cache[r][c] += nums[c];
+ }
+ }
+ }
+
+ filename = String.format(pattern, distribution, 20).replace("20", "");
+ FileWriter fw = new FileWriter(filename);
+ fw.write(StringUtils.join(headers, ',') + "\n");
+ for (int r = 0; r < rows; ++r) {
+ for (int c = 0; c < cols; ++c) {
+ fw.write((cache[r][c] > 0 ? (cache[r][c] / 10) : "") + ((c == cols - 1) ? "\n" : ","));
+ }
+ }
+ fw.close();
+ }
+
+ public static String ADD_SUFFIX(String s, char sep, String suffix) {
+ String[] tmp = StringUtils.split(s, sep);
+ tmp[0] += suffix;
+ return StringUtils.join(tmp, sep);
+ }
+
+ public static int GET_SIZE(Object v) {
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ ObjectOutputStream oos;
+ try {
+ oos = new ObjectOutputStream(bos);
+ oos.writeObject(v);
+ oos.flush();
+ oos.close();
+ bos.close();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ return bos.toByteArray().length;
+ }
+}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Mad.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Mad.java
new file mode 100644
index 00000000000..89f4cc5e219
--- /dev/null
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Mad.java
@@ -0,0 +1,33 @@
+package org.apache.iotdb.tsfile.utils;
+
+import java.io.Serializable;
+
+public class Mad implements Serializable {
+ public double result;
+ public double error_bound;
+
+ public Mad(double result, double error_bound) {
+ this.result = result;
+ this.error_bound = error_bound;
+ }
+
+ public String toString() {
+ return result + " - " + error_bound;
+ }
+
+ public double getResult() {
+ return result;
+ }
+
+ public void setResult(double result) {
+ this.result = result;
+ }
+
+ public double getError_bound() {
+ return error_bound;
+ }
+
+ public void setError_bound(double error_bound) {
+ this.error_bound = error_bound;
+ }
+}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Params.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Params.java
new file mode 100644
index 00000000000..c1e3265cc40
--- /dev/null
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Params.java
@@ -0,0 +1,9 @@
+package org.apache.iotdb.tsfile.utils;
+
+public class Params {
+ public static int SPACE_LIMIT = 1024;
+ public static int BUCKET_NUM_LIMIT = 2048;
+ public static int ITERATION_COUNT = 3;
+ public static double FS_MIN_POSITIVE_VALUE = 1e-6;
+ public static double FS_COMPRESSION_COEFFICIENT = 1.5;
+}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/TPSketch.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/TPSketch.java
new file mode 100644
index 00000000000..d6caee509a4
--- /dev/null
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/TPSketch.java
@@ -0,0 +1,447 @@
+package org.apache.iotdb.tsfile.utils;
+
+import it.unimi.dsi.fastutil.ints.Int2LongMap;
+import it.unimi.dsi.fastutil.ints.Int2LongOpenHashMap;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.NoSuchElementException;
+
+import static org.apache.commons.math3.util.Precision.EPSILON;
+import static org.apache.iotdb.tsfile.utils.Params.*;
+
+public class TPSketch implements Serializable {
+ private double alpha;
+ private double gamma;
+ private double multiplier;
+ private int bucket_num_limit;
+ private int threshold_for_compression;
+
+ private Int2LongOpenHashMap positive_buckets;
+ private Int2LongOpenHashMap negative_buckets;
+ private double collapse_bound;
+ private long zero_count;
+
+ private transient double beta;
+ private final transient double[] valid_range;
+
+ private static double MIN_POSITIVE_VALUE = FS_MIN_POSITIVE_VALUE;
+ private static double COEFFICIENT = FS_COMPRESSION_COEFFICIENT;
+
+ public TPSketch() {
+ this(EPSILON, BUCKET_NUM_LIMIT);
+ }
+
+ public TPSketch(double alpha, int bucket_num_limit) {
+ // System.out.println(alpha);
+ this.alpha = alpha;
+ this.bucket_num_limit = Math.max(bucket_num_limit, 2);
+ this.threshold_for_compression = (int) (bucket_num_limit * COEFFICIENT);
+
+ this.gamma = 2 * alpha / (1 - alpha) + 1;
+ this.multiplier = Math.log(Math.E) / (Math.log1p(gamma - 1));
+ this.beta = 1;
+ this.positive_buckets = new Int2LongOpenHashMap((int) (bucket_num_limit * 0.75));
+ this.negative_buckets = new Int2LongOpenHashMap((int) (bucket_num_limit * 0.25));
+ this.zero_count = 0;
+ this.collapse_bound = -Double.MAX_VALUE;
+ this.valid_range = new double[6];
+ }
+
+ public void insert(double v) {
+ if (v < collapse_bound) {
+ v = collapse_bound;
+ }
+ if (v > MIN_POSITIVE_VALUE) {
+ int i = (int) Math.ceil(Math.log(v) * multiplier);
+ positive_buckets.put(i, positive_buckets.getOrDefault(i, 0L) + 1);
+ } else if (v < -MIN_POSITIVE_VALUE) {
+ int i = (int) Math.ceil(Math.log(-v) * multiplier);
+ negative_buckets.put(i, negative_buckets.getOrDefault(i, 0L) + 1);
+ } else {
+ zero_count++;
+ }
+ collapse(threshold_for_compression);
+ }
+
+ public void insert(double v, double[] bounds) {
+ if (v < bounds[0]) {
+ v = bounds[0];
+ } else if (v > bounds[1] && v < bounds[2]) {
+ v = bounds[2];
+ } else if (v > bounds[3] && v < bounds[4]) {
+ v = bounds[4];
+ } else if (v > bounds[5]) {
+ v = bounds[5];
+ }
+ insert(v);
+ }
+
+ public void merge(TPSketch sketch) {
+ zero_count += sketch.zero_count;
+ for (Int2LongMap.Entry entry : sketch.positive_buckets.int2LongEntrySet()) {
+ positive_buckets.put(
+ entry.getIntKey(),
+ positive_buckets.getOrDefault(entry.getIntKey(), 0L) + entry.getLongValue());
+ }
+
+ for (Int2LongMap.Entry entry : sketch.negative_buckets.int2LongEntrySet()) {
+ negative_buckets.put(
+ entry.getIntKey(),
+ negative_buckets.getOrDefault(entry.getIntKey(), 0L) + entry.getLongValue());
+ }
+
+ this.collapse(bucket_num_limit);
+ }
+
+ private void collapse(int limit) {
+ if (sketch_size() > limit) {
+ int exceed = sketch_size() - bucket_num_limit;
+ int[] indices = negative_buckets.keySet().toIntArray();
+ Arrays.sort(indices);
+ long count = 0;
+ for (int i = Math.max(0, indices.length - exceed); i < indices.length; ++i) {
+ count += negative_buckets.remove(indices[i]);
+ }
+ if (count > 0) {
+ int i = indices.length - exceed - 1;
+ if (i >= 0) {
+ negative_buckets.put(indices[i], negative_buckets.get(indices[i]) + count);
+ collapse_bound = -Math.pow(gamma, indices[i]);
+ } else {
+ zero_count += count;
+ collapse_bound = 0;
+ }
+ }
+ exceed -= (indices.length - Math.max(0, indices.length - exceed));
+ if (exceed > 0) {
+ count = zero_count;
+ if (zero_count > 0) {
+ exceed--;
+ }
+ indices = positive_buckets.keySet().toIntArray();
+ Arrays.sort(indices);
+ for (int i = exceed - 1; i >= 0; --i) {
+ count += positive_buckets.remove(indices[i]);
+ }
+ positive_buckets.put(indices[exceed], positive_buckets.get(indices[exceed]) + count);
+ collapse_bound = Math.pow(gamma, indices[exceed] - 1);
+ }
+ }
+ }
+
+ private Bucket[] union_buckets() {
+ Bucket[] buckets = new Bucket[sketch_size()];
+ int i = 0;
+ for (Int2LongMap.Entry e : positive_buckets.int2LongEntrySet()) {
+ buckets[i++] =
+ new Bucket(
+ (int) e.getIntKey(),
+ Math.pow(gamma, e.getIntKey() - 1),
+ Math.pow(gamma, e.getKey()),
+ e.getLongValue());
+ }
+ for (Int2LongMap.Entry e : negative_buckets.int2LongEntrySet()) {
+ buckets[i++] =
+ new Bucket(
+ (int) e.getIntKey(),
+ -Math.pow(gamma, e.getIntKey()),
+ -Math.pow(gamma, e.getKey() - 1),
+ e.getLongValue());
+ }
+ if (zero_count > 0) {
+ buckets[i] = new Bucket(0, 0, 0, zero_count);
+ }
+ Arrays.sort(buckets, Comparator.comparingDouble(o -> o.lower_bound));
+ return buckets;
+ }
+
+ private long total_count() {
+ return positive_buckets.values().stream().mapToLong(l -> l).sum()
+ + negative_buckets.values().stream().mapToLong(l -> l).sum()
+ + zero_count;
+ }
+
+ private int find_p_index(Bucket[] buckets, long total_count) {
+ long count = 0;
+ double rank = 0.5 * (total_count - 1);
+ for (int i = 0; i < buckets.length; ++i) {
+ count += buckets[i].count;
+ if (count > rank) {
+ return i;
+ }
+ }
+ return -1;
+ }
+
+ private int find_q_index(int p, Bucket[] buckets, long total_count) {
+ int q = p;
+ long count = buckets[p].count;
+ double rank = 0.5 * (total_count - 1);
+ int l = p - 1;
+ int r = p + 1;
+ while (count <= rank && l >= 0 && r < buckets.length) {
+ if (buckets[p].lower_bound - buckets[l].upper_bound
+ < buckets[r].lower_bound - buckets[p].upper_bound) {
+ q = l--;
+ } else {
+ q = r++;
+ }
+ count += buckets[q].count;
+ }
+
+ while (count <= rank && l >= 0) {
+ q = l--;
+ count += buckets[q].count;
+ }
+
+ while (count <= rank && r < buckets.length) {
+ q = r++;
+ count += buckets[q].count;
+ }
+
+ double m_lower_bound = buckets[p].lower_bound + buckets[p].upper_bound - buckets[q].upper_bound;
+ double m_upper_bound = buckets[p].lower_bound + buckets[p].upper_bound - buckets[q].lower_bound;
+ if (p > q) {
+ r--;
+ if (buckets[r].lower_bound <= m_lower_bound && buckets[r].upper_bound >= m_upper_bound) {
+ q = r;
+ }
+ } else if (p < q) {
+ l++;
+ if (buckets[l].lower_bound <= m_lower_bound && buckets[l].upper_bound >= m_upper_bound) {
+ q = l;
+ }
+ }
+ return q;
+ }
+
+ private void setValid_range(Bucket p, Bucket q) {
+ valid_range[0] = p.lower_bound;
+ valid_range[1] = p.upper_bound;
+ valid_range[2] = p.lower_bound + q.lower_bound - p.upper_bound;
+ valid_range[3] = p.upper_bound + q.upper_bound - p.lower_bound;
+ valid_range[4] = 2 * p.lower_bound - q.upper_bound;
+ valid_range[5] = 2 * p.upper_bound - q.lower_bound;
+ Arrays.sort(valid_range);
+ // Arrays.stream(valid_range).forEach(System.out::println);
+ }
+
+ private double minDelta(double delta1, double delta2) {
+ double delta;
+ if (delta1 < 0 && delta2 < 0) {
+ delta = 0;
+ } else if (delta1 < 0) {
+ delta = delta2;
+ } else if (delta2 < 0) {
+ delta = delta1;
+ } else {
+ delta = Math.min(delta1, delta2);
+ }
+ return delta;
+ }
+
+ public double getMedian() {
+ Bucket[] buckets = union_buckets();
+ long total_count = total_count();
+ Bucket p = buckets[find_p_index(buckets, total_count)];
+ if (p.lower_bound < 0) {
+ return 2 * p.lower_bound / (1 + gamma);
+ } else {
+ return 2 * p.upper_bound / (1 + gamma);
+ }
+ }
+
+ public Mad getMad() {
+ beta = 1;
+ Bucket[] buckets = union_buckets();
+ // show(buckets);
+ long total_count = total_count();
+ int p_index = find_p_index(buckets, total_count);
+
+ if (p_index == -1) {
+ throw new NoSuchElementException("No values in the sketch");
+ }
+
+ int q_index = find_q_index(p_index, buckets, total_count);
+ Bucket p = buckets[p_index];
+ Bucket q = buckets[q_index];
+ if (p.upper_bound == collapse_bound
+ || q.upper_bound == collapse_bound
+ || p.lower_bound == collapse_bound
+ || q.lower_bound == collapse_bound) {
+ // System.out.println("\t\tTP BOOM\t"+"The sketch has been compressed too much");
+ return new Mad(0, Double.MAX_VALUE);
+ // throw new IllegalArgumentException("The sketch has been compressed too much");
+ }
+
+ // System.out.println("P and Q: " + p.index + ", " + q.index);
+ if (p.lower_bound * q.lower_bound > 0) {
+ if (p.lower_bound == q.lower_bound) {
+ return new Mad(0, Double.MAX_VALUE);
+ } else {
+ double mad =
+ 2
+ * (p.upper_bound - q.lower_bound)
+ * (p.lower_bound - q.upper_bound)
+ / ((gamma + 1) * Math.abs(p.lower_bound - q.lower_bound));
+ double gamma_p_q = Math.max(p.upper_bound / q.upper_bound, q.upper_bound / p.upper_bound);
+ double delta;
+ if (Math.abs(p.lower_bound) < Math.abs(q.lower_bound)) {
+ delta =
+ minDelta(
+ gamma_p_q / Math.pow(gamma, 2) - 1 / gamma + Math.pow(gamma, -3),
+ 1 / (Math.pow(gamma, 3) - gamma_p_q * gamma + Math.pow(gamma, 2)));
+ } else {
+ delta =
+ minDelta(
+ Math.pow(gamma, -2) + Math.pow(gamma, -3) - 1 / (gamma_p_q * gamma),
+ 1 / (Math.pow(gamma, 2) / gamma_p_q + Math.pow(gamma, 3) - gamma));
+ }
+ beta = 1 - 2 / (1 + delta);
+ if (needTwoPass()) {
+ setValid_range(p, q);
+ return new Mad(mad, (1 + 2 / (gamma_p_q - 1)) * alpha);
+ } else {
+ return new Mad(0, Double.MAX_VALUE);
+ }
+ }
+ } else {
+ double mad =
+ 2
+ * Math.max(
+ Math.abs(p.upper_bound - q.lower_bound), Math.abs(q.upper_bound - p.lower_bound))
+ / (gamma + 1);
+ return new Mad(mad, alpha);
+ }
+ }
+
+ public int sketch_size() {
+ return positive_buckets.size() + negative_buckets.size() + (zero_count == 0 ? 0 : 1);
+ }
+
+ public boolean needTwoPass() {
+ return beta > 0 && beta < 1;
+ }
+
+ public double getBeta() {
+ return beta;
+ }
+
+ public double[] getValid_range() {
+ return valid_range;
+ }
+
+ public void show(Bucket[] buckets) {
+ for (Bucket bucket : buckets) {
+ System.out.println(bucket.index + ": " + bucket.count);
+ }
+ }
+
+ private static class Bucket {
+ int index;
+ double lower_bound;
+ double upper_bound;
+ long count;
+
+ Bucket(int index, double lower_bound, double upper_bound, long count) {
+ this.index = index;
+ this.lower_bound = lower_bound;
+ this.upper_bound = upper_bound;
+ this.count = count;
+ }
+ }
+
+ public double getAlpha() {
+ return alpha;
+ }
+
+ public void setAlpha(double alpha) {
+ this.alpha = alpha;
+ }
+
+ public double getGamma() {
+ return gamma;
+ }
+
+ public void setGamma(double gamma) {
+ this.gamma = gamma;
+ }
+
+ public double getMultiplier() {
+ return multiplier;
+ }
+
+ public void setMultiplier(double multiplier) {
+ this.multiplier = multiplier;
+ }
+
+ public int getBucket_num_limit() {
+ return bucket_num_limit;
+ }
+
+ public void setBucket_num_limit(int bucket_num_limit) {
+ this.bucket_num_limit = bucket_num_limit;
+ }
+
+ public int getThreshold_for_compression() {
+ return threshold_for_compression;
+ }
+
+ public void setThreshold_for_compression(int threshold_for_compression) {
+ this.threshold_for_compression = threshold_for_compression;
+ }
+
+ public Int2LongOpenHashMap getPositive_buckets() {
+ return positive_buckets;
+ }
+
+ public void setPositive_buckets(Int2LongOpenHashMap positive_buckets) {
+ this.positive_buckets = positive_buckets;
+ }
+
+ public Int2LongOpenHashMap getNegative_buckets() {
+ return negative_buckets;
+ }
+
+ public void setNegative_buckets(Int2LongOpenHashMap negative_buckets) {
+ this.negative_buckets = negative_buckets;
+ }
+
+ public double getCollapse_bound() {
+ return collapse_bound;
+ }
+
+ public void setCollapse_bound(double collapse_bound) {
+ this.collapse_bound = collapse_bound;
+ }
+
+ public long getZero_count() {
+ return zero_count;
+ }
+
+ public void setZero_count(long zero_count) {
+ this.zero_count = zero_count;
+ }
+
+ public void setBeta(double beta) {
+ this.beta = beta;
+ }
+
+ public static double getMinPositiveValue() {
+ return MIN_POSITIVE_VALUE;
+ }
+
+ public static void setMinPositiveValue(double minPositiveValue) {
+ MIN_POSITIVE_VALUE = minPositiveValue;
+ }
+
+ public static double getCOEFFICIENT() {
+ return COEFFICIENT;
+ }
+
+ public static void setCOEFFICIENT(double COEFFICIENT) {
+ TPSketch.COEFFICIENT = COEFFICIENT;
+ }
+}
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/read/ReadTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/ReadTest.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/read/TimePlainEncodeReadTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/TimePlainEncodeReadTest.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/read/query/timegenerator/TsFileGeneratorForSeriesReaderByTimestamp.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/query/timegenerator/TsFileGeneratorForSeriesReaderByTimestamp.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/FileGenerator.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/FileGenerator.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/FileUtilsTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/FileUtilsTest.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/RecordUtilsTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/RecordUtilsTest.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/StringContainerTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/StringContainerTest.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/TsFileGeneratorForTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/TsFileGeneratorForTest.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/write/PerfTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/write/PerfTest.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/write/WriteTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/write/WriteTest.java
old mode 100755
new mode 100644
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/write/writer/PageWriterTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/write/writer/PageWriterTest.java
old mode 100755
new mode 100644