This is an automated email from the ASF dual-hosted git repository.
leirui pushed a commit to branch research/M4-visualization
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/research/M4-visualization by
this push:
new 13051b6208 add before
13051b6208 is described below
commit 13051b620893c9e43911fab8a05d2c28a8bc51a7
Author: Lei Rui <[email protected]>
AuthorDate: Tue Oct 11 16:49:32 2022 +0800
add before
---
.../java/org/apache/iotdb/db/service/IoTDB.java | 17 +++
.../org/apache/iotdb/db/service/TSServiceImpl.java | 114 ++++++++++--------
.../iotdb/session/MyBasicOperationTest5.java | 131 +++++++++++++++++++++
.../session/MyRealDataTest1_WriteAndQuery.java | 102 ++++++++--------
.../session/MySmallRealDataWriteQueryTest.java | 4 +-
.../tsfile/common/constant/TsFileConstant.java | 77 ++++++++++++
.../encoding/decoder/DeltaBinaryDecoder.java | 15 +--
.../encoding/encoder/DeltaBinaryEncoder.java | 11 +-
8 files changed, 343 insertions(+), 128 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
index a35f70bd0a..76c16abc54 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
@@ -64,6 +64,23 @@ public class IoTDB implements IoTDBMBean {
} catch (IOException e) {
logger.error("meet error when doing start checking", e);
}
+ // IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ // config.setEnableCPV(true);
+ // config.setTimestampPrecision("ms");
+ // config.setAvgSeriesPointNumberThreshold(100);
+ // config.setUnSeqTsFileSize(1073741824);
+ // config.setSeqTsFileSize(1073741824);
+ // config.setCompactionStrategy(CompactionStrategy.NO_COMPACTION);
+ // config.setEnableUnseqCompaction(false);
+ // config.setEnablePerformanceStat(false);
+ // TSFileDescriptor.getInstance()
+ // .getConfig()
+ // .setEnableRegularityTimeDecode(true);
+ //
TSFileDescriptor.getInstance().getConfig().setRegularTimeInterval(1000);
+ //
TSFileDescriptor.getInstance().getConfig().setPageSizeInByte(1073741824);
+ // config.setSchemaDir("D:\\github\\m4-lsm\\data_kob_2_true\\system");
+ // config.setDataDirs(new
String[]{"D:\\github\\m4-lsm\\data_kob_2_true\\data"});
+
IoTDB daemon = IoTDB.getInstance();
daemon.active();
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
index 7d5ba86751..577af67c61 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
@@ -18,6 +18,22 @@
*/
package org.apache.iotdb.db.service;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import org.antlr.v4.runtime.misc.ParseCancellationException;
import org.apache.iotdb.db.IOMonitor;
import org.apache.iotdb.db.auth.AuthException;
import org.apache.iotdb.db.auth.AuthorityChecker;
@@ -131,29 +147,13 @@ import
org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
-
-import org.antlr.v4.runtime.misc.ParseCancellationException;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.sql.SQLException;
-import java.time.ZoneId;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.stream.Collectors;
-
-/** Thrift RPC implementation at server side. */
+/**
+ * Thrift RPC implementation at server side.
+ */
public class TSServiceImpl implements TSIService.Iface {
private static final Logger LOGGER =
LoggerFactory.getLogger(TSServiceImpl.class);
@@ -327,7 +327,9 @@ public class TSServiceImpl implements TSIService.Iface {
}
}
- /** release single operation resource */
+ /**
+ * release single operation resource
+ */
public void releaseQueryResource(long queryId) throws StorageEngineException
{
sessionManager.releaseQueryResource(queryId);
}
@@ -608,13 +610,13 @@ public class TSServiceImpl implements TSIService.Iface {
return physicalPlan.isQuery()
? internalExecuteQueryStatement(
- statement,
- req.statementId,
- physicalPlan,
- req.fetchSize,
- req.timeout,
- sessionManager.getUsername(req.getSessionId()),
- req.isEnableRedirectQuery())
+ statement,
+ req.statementId,
+ physicalPlan,
+ req.fetchSize,
+ req.timeout,
+ sessionManager.getUsername(req.getSessionId()),
+ req.isEnableRedirectQuery())
: executeUpdateStatement(physicalPlan, req.getSessionId());
} catch (InterruptedException e) {
LOGGER.error(INFO_INTERRUPT_ERROR, req, e);
@@ -641,13 +643,13 @@ public class TSServiceImpl implements TSIService.Iface {
// System.out.println("====DEBUG====: fetchSize=" + req.fetchSize);
return physicalPlan.isQuery()
? internalExecuteQueryStatement(
- statement,
- req.statementId,
- physicalPlan,
- req.fetchSize,
- req.timeout,
- sessionManager.getUsername(req.getSessionId()),
- req.isEnableRedirectQuery())
+ statement,
+ req.statementId,
+ physicalPlan,
+ req.fetchSize,
+ req.timeout,
+ sessionManager.getUsername(req.getSessionId()),
+ req.isEnableRedirectQuery())
: RpcUtils.getTSExecuteStatementResp(
TSStatusCode.EXECUTE_STATEMENT_ERROR, "Statement is not a query
statement.");
} catch (InterruptedException e) {
@@ -667,16 +669,18 @@ public class TSServiceImpl implements TSIService.Iface {
ret.setStatus(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS));
ret.setExecutionInfo(
IOMonitor.print()
- + ". timeColumnTS2DIFFLoadBatchCost="
+ + ". timeColumnTS2DIFFLoadBatchCost= "
+ TsFileConstant.timeColumnTS2DIFFLoadBatchCost.getSum()
- + "us"
+ + " us"
+ + ". countLoadIntBatch= "
+ + TsFileConstant.countLoadIntBatch
// + ". countForRegularEqual="
// + TsFileConstant.countForRegularEqual
// + ". countForRegularNOTEqual="
// + TsFileConstant.countForRegularNOTEqual
// + ". countForRegularZero="
// + TsFileConstant.countForRegularZero
- );
+ );
IOMonitor.finish();
return ret;
}
@@ -692,13 +696,13 @@ public class TSServiceImpl implements TSIService.Iface {
processor.rawDataQueryReqToPhysicalPlan(req,
sessionManager.getZoneId(req.sessionId));
return physicalPlan.isQuery()
? internalExecuteQueryStatement(
- "",
- req.statementId,
- physicalPlan,
- req.fetchSize,
- config.getQueryTimeoutThreshold(),
- sessionManager.getUsername(req.sessionId),
- req.isEnableRedirectQuery())
+ "",
+ req.statementId,
+ physicalPlan,
+ req.fetchSize,
+ config.getQueryTimeoutThreshold(),
+ sessionManager.getUsername(req.sessionId),
+ req.isEnableRedirectQuery())
: RpcUtils.getTSExecuteStatementResp(
TSStatusCode.EXECUTE_STATEMENT_ERROR, "Statement is not a query
statement.");
} catch (InterruptedException e) {
@@ -714,7 +718,7 @@ public class TSServiceImpl implements TSIService.Iface {
/**
* @param plan must be a plan for Query: FillQueryPlan, AggregationPlan,
GroupByTimePlan, UDFPlan,
- * some AuthorPlan
+ * some AuthorPlan
*/
@SuppressWarnings({"squid:S3776", "squid:S1141"}) // Suppress high Cognitive
Complexity warning
private TSExecuteStatementResp internalExecuteQueryStatement(
@@ -726,8 +730,8 @@ public class TSServiceImpl implements TSIService.Iface {
String username,
boolean enableRedirect)
throws QueryProcessException, SQLException, StorageEngineException,
- QueryFilterOptimizationException, MetadataException, IOException,
InterruptedException,
- TException, AuthException {
+ QueryFilterOptimizationException, MetadataException, IOException,
InterruptedException,
+ TException, AuthException {
// start record execution time
IOMonitor.setSQL(statement);
@@ -865,7 +869,9 @@ public class TSServiceImpl implements TSIService.Iface {
dataSet.getDataTypes().stream().map(Enum::toString).collect(Collectors.toList()));
}
- /** get ResultSet schema */
+ /**
+ * get ResultSet schema
+ */
private TSExecuteStatementResp getQueryColumnHeaders(PhysicalPlan
physicalPlan, String username)
throws AuthException, TException, QueryProcessException,
MetadataException {
@@ -1134,10 +1140,12 @@ public class TSServiceImpl implements TSIService.Iface {
return encoder;
}
- /** create QueryDataSet and buffer it for fetchResults */
+ /**
+ * create QueryDataSet and buffer it for fetchResults
+ */
private QueryDataSet createQueryDataSet(long queryId, PhysicalPlan
physicalPlan, int fetchSize)
throws QueryProcessException, QueryFilterOptimizationException,
StorageEngineException,
- IOException, MetadataException, SQLException, TException,
InterruptedException {
+ IOException, MetadataException, SQLException, TException,
InterruptedException {
QueryContext context = genQueryContext(queryId, physicalPlan.isDebug());
if (physicalPlan instanceof QueryPlan) {
@@ -1196,7 +1204,7 @@ public class TSServiceImpl implements TSIService.Iface {
statement, sessionManager.getZoneId(sessionId),
DEFAULT_FETCH_SIZE);
return physicalPlan.isQuery()
? RpcUtils.getTSExecuteStatementResp(
- TSStatusCode.EXECUTE_STATEMENT_ERROR, "Statement is a query
statement.")
+ TSStatusCode.EXECUTE_STATEMENT_ERROR, "Statement is a query
statement.")
: executeUpdateStatement(physicalPlan, sessionId);
}
@@ -1626,7 +1634,9 @@ public class TSServiceImpl implements TSIService.Iface {
return insertTabletPlan;
}
- /** construct one InsertMultiTabletPlan and process it */
+ /**
+ * construct one InsertMultiTabletPlan and process it
+ */
public TSStatus insertTabletsInternal(TSInsertTabletsReq req) throws
IllegalPathException {
List<InsertTabletPlan> insertTabletPlanList = new ArrayList<>();
InsertMultiTabletPlan insertMultiTabletPlan = new InsertMultiTabletPlan();
diff --git
a/session/src/test/java/org/apache/iotdb/session/MyBasicOperationTest5.java
b/session/src/test/java/org/apache/iotdb/session/MyBasicOperationTest5.java
new file mode 100644
index 0000000000..e6f11c7659
--- /dev/null
+++ b/session/src/test/java/org/apache/iotdb/session/MyBasicOperationTest5.java
@@ -0,0 +1,131 @@
+package org.apache.iotdb.session;
+
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics;
+import org.junit.Assert;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.text.DecimalFormat;
+
+public class MyBasicOperationTest5 {
+
+ public static void main(String[] args) throws IOException {
+ // op1: direct construct regularBytes when decoding
+ // op2: readRegularBytes
+
+ int repeat = 10000000;
+ DescriptiveStatistics op1 = new DescriptiveStatistics();
+ DescriptiveStatistics op2 = new DescriptiveStatistics();
+ int writeWidth = 3;
+ int minDeltaBase = 3;
+
+ for (int k = 0; k < repeat; k++) {
+ int regularTimeInterval =
+ (int) ((Math.random() * (Math.pow(2, writeWidth) - 0)) + 0) +
minDeltaBase;
+
+ // test op1
+ long start = System.nanoTime();
+ byte[][] regularBytes =
+ TsFileConstant.generateRegularByteArray(writeWidth,
regularTimeInterval, minDeltaBase);
+ long elapsedTime = System.nanoTime() - start;
+ op1.addValue(elapsedTime / 1000.0);
+
+ long sum1 = 0;
+ for (int i = 0; i < 8; i++) {
+ byte[] byteArray = regularBytes[i];
+ int num = byteArray.length;
+ for (int j = 0; j < num; j++) {
+ sum1 += byteArray[j];
+ }
+ }
+
+ // prepare data for op2
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ for (int i = 0; i < 8; i++) {
+ byte[] byteArray = regularBytes[i];
+ int num = byteArray.length;
+ ReadWriteIOUtils.write(num, out);
+ for (int j = 0; j < num; j++) {
+ ReadWriteIOUtils.write(byteArray[j], out);
+ }
+ }
+ ByteBuffer byteBuffer = ByteBuffer.wrap(out.toByteArray());
+
+ // test op2
+ start = System.nanoTime();
+ byte[][] res1 = TsFileConstant.readRegularBytes(byteBuffer);
+ elapsedTime = System.nanoTime() - start;
+ op2.addValue(elapsedTime / 1000.0);
+
+ long sum2 = 0;
+ for (int i = 0; i < 8; i++) {
+ byte[] byteArray = res1[i];
+ int num = byteArray.length;
+ for (int j = 0; j < num; j++) {
+ sum2 += byteArray[j];
+ }
+ }
+ Assert.assertEquals(sum1, sum2);
+ }
+
+ printStat(op1, "op1-direct construct regularBytes");
+ printStat(op2, "op2-readRegularBytes");
+ System.out.println("op1/op2=" + op1.getMean() / op2.getMean());
+ System.out.println("op2/op1=" + op2.getMean() / op1.getMean());
+ }
+
+ private static String printStat(DescriptiveStatistics statistics, String
name) {
+ DecimalFormat df = new DecimalFormat("#,###.00");
+ double max = statistics.getMax();
+ double min = statistics.getMin();
+ double mean = statistics.getMean();
+ double std = statistics.getStandardDeviation();
+ double p25 = statistics.getPercentile(25);
+ double p50 = statistics.getPercentile(50);
+ double p75 = statistics.getPercentile(75);
+ double p90 = statistics.getPercentile(90);
+ double p95 = statistics.getPercentile(95);
+ String res =
+ name
+ + "_stats"
+ + ": "
+ + "num="
+ + statistics.getN()
+ + ", "
+ + "sum="
+ + df.format(statistics.getSum())
+ + "us,"
+ + "mean="
+ + df.format(mean)
+ + ", "
+ + "min="
+ + df.format(min)
+ + ", "
+ + "max="
+ + df.format(max)
+ + ", "
+ + "std="
+ + df.format(std)
+ + ", "
+ + "p25="
+ + df.format(p25)
+ + ", "
+ + "p50="
+ + df.format(p50)
+ + ", "
+ + "p75="
+ + df.format(p75)
+ + ", "
+ + "p90="
+ + df.format(p90)
+ + ", "
+ + "p95="
+ + df.format(p95);
+ System.out.println(res);
+ return res;
+ }
+}
diff --git
a/session/src/test/java/org/apache/iotdb/session/MyRealDataTest1_WriteAndQuery.java
b/session/src/test/java/org/apache/iotdb/session/MyRealDataTest1_WriteAndQuery.java
index 5e35237174..aefbdef833 100644
---
a/session/src/test/java/org/apache/iotdb/session/MyRealDataTest1_WriteAndQuery.java
+++
b/session/src/test/java/org/apache/iotdb/session/MyRealDataTest1_WriteAndQuery.java
@@ -2,13 +2,9 @@ package org.apache.iotdb.session;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.engine.compaction.CompactionStrategy;
-import org.apache.iotdb.db.utils.EnvironmentUtils;
-import org.apache.iotdb.jdbc.Config;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.SessionDataSet.DataIterator;
-import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.junit.After;
@@ -208,47 +204,47 @@ public class MyRealDataTest1_WriteAndQuery {
@Before
public void setUp() throws Exception {
- config.setEnableCPV(true);
- config.setTimestampPrecision(timestamp_precision);
- config.setAvgSeriesPointNumberThreshold(iotdb_chunk_point_size);
- config.setUnSeqTsFileSize(1073741824);
- config.setSeqTsFileSize(1073741824);
- config.setCompactionStrategy(CompactionStrategy.NO_COMPACTION);
- config.setEnableUnseqCompaction(false);
- config.setEnablePerformanceStat(false);
-
- TSFileDescriptor.getInstance()
- .getConfig()
- .setEnableRegularityTimeDecode(enableRegularityTimeDecode);
-
TSFileDescriptor.getInstance().getConfig().setRegularTimeInterval(regularTimeInterval);
- TSFileDescriptor.getInstance().getConfig().setPageSizeInByte(1073741824);
-
- EnvironmentUtils.envSetUp(); // start after configuration settings
- Class.forName(Config.JDBC_DRIVER_NAME);
-
- System.out.println("[WriteData] device=" + device);
- System.out.println("[WriteData] measurement=" + measurement);
- System.out.println("[WriteData] dataType=" + tsDataType);
- System.out.println("[WriteData] timestamp_precision=" +
timestamp_precision);
- System.out.println("[WriteData] dataMinTime=" + dataMinTime);
- System.out.println("[WriteData] dataMaxTime=" + dataMaxTime);
- System.out.println("[WriteData] total_time_length=" + total_time_length);
- System.out.println("[WriteData] total_point_number=" + total_point_number);
- System.out.println("[WriteData] iotdb_chunk_point_size=" +
iotdb_chunk_point_size);
- System.out.println("[WriteData] derived estimated chunkAvgTimeLen =" +
chunkAvgTimeLen);
- System.out.println("[WriteData] filePath=" + filePath);
- System.out.println("[WriteData] deletePercentage=" + deletePercentage);
- System.out.println("[WriteData] deleteLenPercentage=" +
deleteLenPercentage);
- System.out.println("[WriteData] timeIdx=" + timeIdx);
- System.out.println("[WriteData] valueIdx=" + valueIdx);
- System.out.println(
- "[WriteData] enableRegularityTimeDecode="
- +
TSFileDescriptor.getInstance().getConfig().isEnableRegularityTimeDecode());
+ // config.setEnableCPV(true);
+ // config.setTimestampPrecision(timestamp_precision);
+ // config.setAvgSeriesPointNumberThreshold(iotdb_chunk_point_size);
+ // config.setUnSeqTsFileSize(1073741824);
+ // config.setSeqTsFileSize(1073741824);
+ // config.setCompactionStrategy(CompactionStrategy.NO_COMPACTION);
+ // config.setEnableUnseqCompaction(false);
+ // config.setEnablePerformanceStat(false);
+ //
+ // TSFileDescriptor.getInstance()
+ // .getConfig()
+ // .setEnableRegularityTimeDecode(enableRegularityTimeDecode);
+ //
TSFileDescriptor.getInstance().getConfig().setRegularTimeInterval(regularTimeInterval);
+ //
TSFileDescriptor.getInstance().getConfig().setPageSizeInByte(1073741824);
+ //
+ // EnvironmentUtils.envSetUp(); // start after configuration settings
+ // Class.forName(Config.JDBC_DRIVER_NAME);
+ //
+ // System.out.println("[WriteData] device=" + device);
+ // System.out.println("[WriteData] measurement=" + measurement);
+ // System.out.println("[WriteData] dataType=" + tsDataType);
+ // System.out.println("[WriteData] timestamp_precision=" +
timestamp_precision);
+ // System.out.println("[WriteData] dataMinTime=" + dataMinTime);
+ // System.out.println("[WriteData] dataMaxTime=" + dataMaxTime);
+ // System.out.println("[WriteData] total_time_length=" +
total_time_length);
+ // System.out.println("[WriteData] total_point_number=" +
total_point_number);
+ // System.out.println("[WriteData] iotdb_chunk_point_size=" +
iotdb_chunk_point_size);
+ // System.out.println("[WriteData] derived estimated chunkAvgTimeLen ="
+ chunkAvgTimeLen);
+ // System.out.println("[WriteData] filePath=" + filePath);
+ // System.out.println("[WriteData] deletePercentage=" +
deletePercentage);
+ // System.out.println("[WriteData] deleteLenPercentage=" +
deleteLenPercentage);
+ // System.out.println("[WriteData] timeIdx=" + timeIdx);
+ // System.out.println("[WriteData] valueIdx=" + valueIdx);
+ // System.out.println(
+ // "[WriteData] enableRegularityTimeDecode="
+ // +
TSFileDescriptor.getInstance().getConfig().isEnableRegularityTimeDecode());
}
@After
public void tearDown() throws Exception {
- EnvironmentUtils.cleanEnv();
+ // EnvironmentUtils.cleanEnv();
}
/** Before writing data, make sure check the server parameter
configurations. */
@@ -257,8 +253,8 @@ public class MyRealDataTest1_WriteAndQuery {
// timeIdx valueIdx
@Test
public void test1() throws Exception {
- System.out.println("writing data...");
- writeData();
+ // System.out.println("writing data...");
+ // writeData();
System.out.println("querying data...");
System.out.println("[QueryData] query range=" + range);
@@ -268,15 +264,17 @@ public class MyRealDataTest1_WriteAndQuery {
throw new IOException("Approach wrong. Only accepts mac/moc/cpv");
}
System.out.println("[QueryData] approach=" + approach);
- if (approach.equals("moc")) {
- System.out.println(
- "MAKE SURE you have set the enable_CPV as false in
`iotdb-engine.properties` for MOC!");
- Assert.assertFalse(config.isEnableCPV());
- } else if (approach.equals("cpv")) {
- System.out.println(
- "MAKE SURE you have set the enable_CPV as true in
`iotdb-engine.properties` for CPV!");
- Assert.assertTrue(config.isEnableCPV());
- }
+ // if (approach.equals("moc")) {
+ // System.out.println(
+ // "MAKE SURE you have set the enable_CPV as false in
`iotdb-engine.properties` for
+ // MOC!");
+ // Assert.assertFalse(config.isEnableCPV());
+ // } else if (approach.equals("cpv")) {
+ // System.out.println(
+ // "MAKE SURE you have set the enable_CPV as true in
`iotdb-engine.properties` for
+ // CPV!");
+ // Assert.assertTrue(config.isEnableCPV());
+ // }
long minTime;
long maxTime;
diff --git
a/session/src/test/java/org/apache/iotdb/session/MySmallRealDataWriteQueryTest.java
b/session/src/test/java/org/apache/iotdb/session/MySmallRealDataWriteQueryTest.java
index ea35d5e91f..6c177f146e 100644
---
a/session/src/test/java/org/apache/iotdb/session/MySmallRealDataWriteQueryTest.java
+++
b/session/src/test/java/org/apache/iotdb/session/MySmallRealDataWriteQueryTest.java
@@ -9,6 +9,7 @@ import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.SessionDataSet.DataIterator;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.junit.After;
@@ -62,7 +63,7 @@ public class MySmallRealDataWriteQueryTest {
private static int valueIdx = 1; // 值idx，从0开始
private static int w = 3;
private static long range = total_time_length;
- private static boolean enableRegularityTimeDecode = false;
+ private static boolean enableRegularityTimeDecode = true;
private static long regularTimeInterval = 511996L;
// private static long regularTimeInterval = 511997L;
private static String approach = "cpv"; // 选择查询执行算法: 1: MAC, 2: MOC, 3: CPV
@@ -222,6 +223,7 @@ public class MySmallRealDataWriteQueryTest {
System.out.println("[QueryData] query result line number=" + c);
dataSet.closeOperationHandle();
session.close();
+ System.out.println("countLoadIntBatch=" +
TsFileConstant.countLoadIntBatch);
}
public void writeData()
diff --git
a/tsfile/src/main/java/org/apache/iotdb/tsfile/common/constant/TsFileConstant.java
b/tsfile/src/main/java/org/apache/iotdb/tsfile/common/constant/TsFileConstant.java
index 606fab6ad8..4f694432dd 100644
---
a/tsfile/src/main/java/org/apache/iotdb/tsfile/common/constant/TsFileConstant.java
+++
b/tsfile/src/main/java/org/apache/iotdb/tsfile/common/constant/TsFileConstant.java
@@ -18,9 +18,13 @@
*/
package org.apache.iotdb.tsfile.common.constant;
+import org.apache.iotdb.tsfile.utils.BytesUtils;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics;
import java.io.IOException;
+import java.nio.ByteBuffer;
public class TsFileConstant {
@@ -55,6 +59,8 @@ public class TsFileConstant {
public static DescriptiveStatistics prepareAllRegulars = new
DescriptiveStatistics();
+ public static long countLoadIntBatch = 0;
+
public static long bytesToLong_byteNum1_wholeByte = 0;
public static long bytesToLong_byteNum1_smallByte = 0;
public static long byteToLong_byteNums_firstByte_wholeByte = 0;
@@ -91,6 +97,77 @@ public class TsFileConstant {
private TsFileConstant() {}
+ /**
+ * Pad the bytes with bit-packed newRegularDelta at 8 relative positions.
This function is used
+ * for regularity-aware decoding to compare regularBytes with newDeltaBytes.
Example:
+ * writeWidth=3, bit-packed newRegularDelta = 0b101, return regularBytes as:
0->0b10110110,
+ * 1->0b11011011, 2->0b01101101, 3->0b10110110, 4->0b11011011,
5->0b01101101, 6->0b10110110 &
+ * 0b11011011, 7->0b11011011 & 0b01101101.
+ */
+ public static byte[][] generateRegularByteArray(
+ int writeWidth, long regularTimeInterval, long minDeltaBase) throws
IOException {
+ long newRegularDelta = regularTimeInterval - minDeltaBase;
+ if (writeWidth == 0 || newRegularDelta < 0 || newRegularDelta >=
Math.pow(2, writeWidth)) {
+ throw new IOException(
+ "writeWidth == 0 || newRegularDelta < 0 || newRegularDelta >=
Math.pow(2, writeWidth)");
+ }
+
+ byte[][] regularBytes = new byte[8][]; // 8 relative positions.
relativePos->bytes
+ for (int i = 0; i < 8; i++) {
+ // i is the starting position in the byte from high to low bits
+
+ int endPos = i + writeWidth - 1; // starting from 0
+ int byteNum = endPos / 8 + 1;
+ byte[] byteArray = new byte[byteNum];
+ if (newRegularDelta != 0) {
+ // put bit-packed newRegularDelta starting at position i,
+ // and pad the front and back with newRegularDeltas.
+ // Otherwise if newRegularDelta=0, just leave byteArray as initial
zeros
+
+ // 1. deal with padding the first byte
+ for (int x = i - 1; x >= 0; x--) {
+ // y is the position in the bit-packed newRegularDelta,
0->packWidth-1 from low to
+ // high bits
+ int y = (i - x - 1) % writeWidth;
+ // get the bit indicated by y pos
+ int value = BytesUtils.getLongN(newRegularDelta, y);
+ // put the bit indicated by y pos into regularBytes
+ // setByte pos is from high to low starting from 0, corresponding to
x
+ byteArray[0] = BytesUtils.setByteN(byteArray[0], x, value);
+ }
+
+ // 2. deal with putting newRegularDeltas
+ BytesUtils.longToBytes(newRegularDelta, byteArray, i, writeWidth);
+
+ // 3. deal with padding the last byte
+ for (int x = endPos + 1; x < byteNum * 8; x++) {
+ // y is the position in the bit-packed newRegularDelta,
0->packWidth-1 from low to
+ // high bits
+ int y = writeWidth - 1 - (x - endPos - 1) % writeWidth;
+ // get the bit indicated by y pos
+ int value = BytesUtils.getLongN(newRegularDelta, y);
+ // put the bit indicated by y pos into regularBytes
+ // setByte pos is from high to low starting from 0, corresponding to
x
+ byteArray[byteNum - 1] = BytesUtils.setByteN(byteArray[byteNum - 1],
x, value);
+ }
+ }
+ regularBytes[i] = byteArray;
+ }
+ return regularBytes;
+ }
+
+ public static byte[][] readRegularBytes(ByteBuffer buffer) {
+ byte[][] regularBytes = new byte[8][];
+ for (int i = 0; i < 8; i++) {
+ int byteArrayLength = ReadWriteIOUtils.readInt(buffer);
+ regularBytes[i] = new byte[byteArrayLength];
+ for (int j = 0; j < byteArrayLength; j++) {
+ regularBytes[i][j] = ReadWriteIOUtils.readByte(buffer);
+ }
+ }
+ return regularBytes;
+ }
+
public static void printByteToLongStatistics() {
System.out.println("bytesToLong_byteNum1_wholeByte=" +
bytesToLong_byteNum1_wholeByte);
System.out.println("bytesToLong_byteNum1_smallByte=" +
bytesToLong_byteNum1_smallByte);
diff --git
a/tsfile/src/main/java/org/apache/iotdb/tsfile/encoding/decoder/DeltaBinaryDecoder.java
b/tsfile/src/main/java/org/apache/iotdb/tsfile/encoding/decoder/DeltaBinaryDecoder.java
index 2e740e09d6..ff8a15ec8c 100644
---
a/tsfile/src/main/java/org/apache/iotdb/tsfile/encoding/decoder/DeltaBinaryDecoder.java
+++
b/tsfile/src/main/java/org/apache/iotdb/tsfile/encoding/decoder/DeltaBinaryDecoder.java
@@ -209,6 +209,7 @@ public abstract class DeltaBinaryDecoder extends Decoder {
* @return long value
*/
protected long loadIntBatch(ByteBuffer buffer) {
+ TsFileConstant.countLoadIntBatch++;
long start = System.nanoTime();
packNum = ReadWriteIOUtils.readInt(buffer);
@@ -246,7 +247,7 @@ public abstract class DeltaBinaryDecoder extends Decoder {
} else {
// [CASE 3]
// read regularBytes and deltaBuf
- byte[][] regularBytes = readRegularBytes(buffer);
+ byte[][] regularBytes = TsFileConstant.readRegularBytes(buffer);
encodingLength = ceil(packNum * packWidth);
deltaBuf = new byte[encodingLength];
buffer.get(deltaBuf);
@@ -335,18 +336,6 @@ public abstract class DeltaBinaryDecoder extends Decoder {
firstValue = ReadWriteIOUtils.readLong(buffer);
}
- private byte[][] readRegularBytes(ByteBuffer buffer) {
- byte[][] regularBytes = new byte[8][];
- for (int i = 0; i < 8; i++) {
- int byteArrayLength = ReadWriteIOUtils.readInt(buffer);
- regularBytes[i] = new byte[byteArrayLength];
- for (int j = 0; j < byteArrayLength; j++) {
- regularBytes[i][j] = ReadWriteIOUtils.readByte(buffer);
- }
- }
- return regularBytes;
- }
-
@Override
protected void allocateDataArray() {
data = new long[packNum];
diff --git
a/tsfile/src/main/java/org/apache/iotdb/tsfile/encoding/encoder/DeltaBinaryEncoder.java
b/tsfile/src/main/java/org/apache/iotdb/tsfile/encoding/encoder/DeltaBinaryEncoder.java
index 4efe7244e2..f8fecf4454 100644
---
a/tsfile/src/main/java/org/apache/iotdb/tsfile/encoding/encoder/DeltaBinaryEncoder.java
+++
b/tsfile/src/main/java/org/apache/iotdb/tsfile/encoding/encoder/DeltaBinaryEncoder.java
@@ -318,22 +318,13 @@ public abstract class DeltaBinaryEncoder extends Encoder {
byteArray[byteNum - 1] = BytesUtils.setByteN(byteArray[byteNum
- 1], x, value);
}
}
- // regularBytes[i] = byteArray;
+ // flush byteArray
int num = byteArray.length;
ReadWriteIOUtils.write(num, out);
for (int j = 0; j < num; j++) {
ReadWriteIOUtils.write(byteArray[j], out);
}
}
- // // TODO write out
- // for (int i = 0; i < 8; i++) {
- // byte[] byteArray = regularBytes[i];
- // int num = byteArray.length;
- // ReadWriteIOUtils.write(num, out);
- // for (int j = 0; j < num; j++) {
- // ReadWriteIOUtils.write(byteArray[j], out);
- // }
- // }
}
}
}
