This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch research/auto-aligned
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/research/auto-aligned by this
push:
new e955d08 add autoaligned time series (#5012)
e955d08 is described below
commit e955d086e4f74c95cbca19e5330005609521e06d
Author: Chenguang Fang <[email protected]>
AuthorDate: Mon Feb 7 18:53:21 2022 +0800
add autoaligned time series (#5012)
---
.../org/apache/iotdb/db/qp/sql/IoTDBSqlLexer.g4 | 8 +
.../org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 | 8 +-
.../iotdb/db/engine/flush/FlushGroupingEngine.java | 362 ++++++++++++++++++++
.../iotdb/db/engine/flush/MemTableFlushTask.java | 14 +-
.../iotdb/db/engine/memtable/AbstractMemTable.java | 71 ++++
.../memtable/AutoAlignedWritableMemChunk.java | 367 +++++++++++++++++++++
.../memtable/AutoAlignedWritableMemChunkGroup.java | 116 +++++++
.../apache/iotdb/db/engine/memtable/IMemTable.java | 8 +
.../db/engine/storagegroup/TsFileProcessor.java | 67 +++-
.../org/apache/iotdb/db/metadata/MManager.java | 94 ++++--
.../apache/iotdb/db/metadata/idtable/IDTable.java | 9 +
.../db/metadata/idtable/IDTableHashmapImpl.java | 56 ++++
.../db/metadata/idtable/entry/DeviceEntry.java | 9 +
.../db/metadata/idtable/entry/DiskSchemaEntry.java | 22 ++
.../db/metadata/idtable/entry/SchemaEntry.java | 31 ++
.../iotdb/db/metadata/logfile/MLogWriter.java | 27 +-
.../iotdb/db/metadata/mnode/EntityMNode.java | 12 +
.../iotdb/db/metadata/mnode/IEntityMNode.java | 4 +
.../org/apache/iotdb/db/metadata/mtree/MTree.java | 63 ++++
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 13 +
.../org/apache/iotdb/db/qp/logical/Operator.java | 3 +-
.../iotdb/db/qp/logical/crud/InsertOperator.java | 8 +
.../sys/CreateAutoAlignedTimeSeriesOperator.java | 131 ++++++++
.../apache/iotdb/db/qp/physical/PhysicalPlan.java | 49 +--
.../iotdb/db/qp/physical/crud/InsertPlan.java | 10 +
.../iotdb/db/qp/physical/crud/InsertRowPlan.java | 23 ++
.../sys/CreateAutoAlignedTimeSeriesPlan.java | 267 +++++++++++++++
.../apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java | 59 ++++
.../db/utils/datastructure/AlignedTVList.java | 4 +
.../chunk/AutoAlignedChunkGroupWriterImpl.java | 301 +++++++++++++++++
.../write/chunk/AutoAlignedChunkWriterImpl.java | 239 ++++++++++++++
31 files changed, 2360 insertions(+), 95 deletions(-)
diff --git a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlLexer.g4
b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlLexer.g4
index 024cf78..e703b12 100644
--- a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlLexer.g4
+++ b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlLexer.g4
@@ -85,6 +85,14 @@ ATTRIBUTES
: A T T R I B U T E S
;
+AUTOALIGN
+ : A U T O A L I G N
+ ;
+
+AUTOALIGNED
+ : A U T O A L I G N E D
+ ;
+
AUTOREGISTER
: A U T O R E G I S T E R
;
diff --git a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
index 7a2e82f..201f271 100644
--- a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
+++ b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
@@ -81,6 +81,7 @@ createStorageGroup
// Create Timeseries
createTimeseries
: CREATE ALIGNED TIMESERIES fullPath alignedMeasurements?
#createAlignedTimeseries
+ | CREATE AUTOALIGNED TIMESERIES fullPath autoAlignedMeasurements?
#createAutoAlignedTimeseries
| CREATE TIMESERIES fullPath attributeClauses #createNonAlignedTimeseries
;
@@ -89,6 +90,11 @@ alignedMeasurements
(COMMA nodeNameWithoutWildcard attributeClauses)* RR_BRACKET
;
+autoAlignedMeasurements
+ : LR_BRACKET nodeNameWithoutWildcard attributeClauses
+ (COMMA nodeNameWithoutWildcard attributeClauses)* RR_BRACKET
+ ;
+
// Create Schema Template
createSchemaTemplate
: CREATE SCHEMA? TEMPLATE templateName=identifier
@@ -431,7 +437,7 @@ timeInterval
// Insert Statement
insertStatement
- : INSERT INTO prefixPath insertColumnsSpec ALIGNED? VALUES insertValuesSpec
+ : INSERT INTO prefixPath insertColumnsSpec ALIGNED? AUTOALIGNED? VALUES
insertValuesSpec
;
insertColumnsSpec
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/flush/FlushGroupingEngine.java
b/server/src/main/java/org/apache/iotdb/db/engine/flush/FlushGroupingEngine.java
new file mode 100644
index 0000000..822f3fd
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/flush/FlushGroupingEngine.java
@@ -0,0 +1,362 @@
+/*
+ * 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.engine.flush;
+
+import org.apache.iotdb.db.utils.datastructure.AlignedTVList;
+import org.apache.iotdb.tsfile.utils.BitMap;
+import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.iotdb.db.rescon.PrimitiveArrayManager.ARRAY_SIZE;
+
+public class FlushGroupingEngine {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(MemTableFlushTask.class);
+
+ public static class ColumnGroup {
+ public ArrayList<Integer> columns;
+ public int maxGain = -1;
+ public int maxGainPosition = -1;
+ public long lastTimestampIdx;
+ public long startTimestampIdx;
+ public int timeSeriesLength; // number of exist timestamps, i.e., zeros in
bitmap
+ public long interval;
+ public ArrayList<BitMap> bitmap;
+ public int size; // total length of the bitmap
+ public ArrayList<Integer> gains;
+
+ public ColumnGroup(
+ ArrayList<Integer> columns, int maxGain, ArrayList<BitMap> bitmap, int
size, int GroupNum) {
+ this.columns = columns;
+ this.maxGain = maxGain;
+ this.bitmap = bitmap;
+ this.size = size;
+ this.gains = new ArrayList<>(Collections.nCopies(GroupNum, -1));
+ this.updateTimestampInfo();
+ }
+
+ public void updateTimestampInfo() {
+ int onesCount = 0;
+ this.startTimestampIdx = -1;
+ this.lastTimestampIdx = -1;
+ int bitmapSize = ARRAY_SIZE;
+ if (this.bitmap == null) {}
+
+ for (int i = 0; i < this.bitmap.size(); i++) {
+ if (this.bitmap.get(i) == null) {
+ if (this.startTimestampIdx == -1) {
+ this.startTimestampIdx = i * bitmapSize;
+ }
+ if (i * bitmapSize + bitmapSize - 1 < size) {
+ this.lastTimestampIdx = i * bitmapSize + bitmapSize - 1;
+ } else {
+ if (i * bitmapSize - 1 < size) {
+ this.lastTimestampIdx = size - 1;
+ }
+ }
+ } else {
+ BitMap bm = this.bitmap.get(i);
+ for (int j = 0; j < bm.getSize() / Byte.SIZE; j++) {
+ int onesNumber = countOnesForByte(bm.getByteArray()[j]);
+ if (this.startTimestampIdx == -1 && onesNumber != 8) {
+ this.startTimestampIdx = i * bitmapSize + j * 8 +
firstZero(bm.getByteArray()[j]);
+ }
+ if (onesNumber != 8) {
+ if (i * bitmapSize + j * 8 + lastZero(bm.getByteArray()[j]) <
size) {
+ this.lastTimestampIdx = i * bitmapSize + j * 8 +
lastZero(bm.getByteArray()[j]);
+ }
+ }
+ onesCount += countOnesForByte(bm.getByteArray()[j]);
+ }
+ }
+ }
+ this.timeSeriesLength = this.size - onesCount;
+ }
+
+ public static ColumnGroup mergeColumnGroup(ColumnGroup g1, ColumnGroup g2)
{
+ // merge bitmap
+ ArrayList<BitMap> newBitmap = new ArrayList<>();
+ for (int i = 0; i < g1.bitmap.size(); i++) {
+ int bitmapLength = ARRAY_SIZE;
+ byte[] bitmap = new byte[bitmapLength / Byte.SIZE + 1];
+ for (int j = 0; j < bitmap.length; j++) {
+ byte g1byte = 0;
+ if (g1.bitmap.get(i) != null) {
+ g1byte = g1.bitmap.get(i).getByteArray()[j];
+ }
+ byte g2byte = 0;
+ if (g2.bitmap.get(i) != null) {
+ g2byte = g2.bitmap.get(i).getByteArray()[j];
+ }
+ bitmap[j] = (byte) (g1byte & g2byte);
+ }
+ newBitmap.add(new BitMap(ARRAY_SIZE, bitmap));
+ }
+ // merge columns
+ ArrayList<Integer> newColumns = (ArrayList<Integer>) g1.columns.clone();
+ newColumns.addAll(g2.columns);
+
+ // update posIndex
+ return new ColumnGroup(newColumns, -1, newBitmap, g1.size,
g1.gains.size() - 1);
+ }
+ }
+
+ /** autoaligned grouping method * */
+ public static void grouping(AlignedTVList dataList, List<IMeasurementSchema>
schemaList) {
+ try {
+ if (dataList.getBitMaps() == null) {
+ return;
+ }
+
+ int timeSeriesNumber = dataList.getValues().size();
+ int size = dataList.rowCount();
+ ArrayList<ColumnGroup> columnGroups = new ArrayList<>();
+
+ if (timeSeriesNumber == 0) {
+ return;
+ }
+ if (timeSeriesNumber == 1) {
+ return;
+ }
+
+ // init posMap, maxGain, groupingResult
+ for (int i = 0; i < timeSeriesNumber; i++) {
+ ArrayList<Integer> newGroup = new ArrayList<>();
+ newGroup.add(i);
+ ArrayList<BitMap> bitmap = (ArrayList<BitMap>)
dataList.getBitMaps().get(i);
+ if (bitmap == null) {
+ bitmap = new ArrayList<>();
+ for (int j = 0; j < dataList.getBitMaps().size(); j++) {
+ if (dataList.getBitMaps().get(j) != null) {
+ int sizeBitmap = dataList.getBitMaps().get(j).size();
+ for (int k = 0; k < sizeBitmap; k++) {
+ bitmap.add(null);
+ }
+ }
+ }
+ }
+ columnGroups.add(new ColumnGroup(newGroup, -1, bitmap, size,
timeSeriesNumber));
+ }
+
+ // init gain matrix
+ for (int i = 0; i < timeSeriesNumber; i++) {
+ for (int j = i + 1; j < timeSeriesNumber; j++) {
+ int gain = computeGain(columnGroups.get(i), columnGroups.get(j),
size);
+ columnGroups.get(i).gains.set(j, gain);
+ columnGroups.get(j).gains.set(i, gain);
+ if (columnGroups.get(i).maxGain < gain) {
+ columnGroups.get(i).maxGain = gain;
+ columnGroups.get(i).maxGainPosition = j;
+ }
+ if (columnGroups.get(j).maxGain < gain) {
+ columnGroups.get(j).maxGain = gain;
+ columnGroups.get(j).maxGainPosition = i;
+ }
+ }
+ }
+
+ while (true) {
+ /** merge * */
+
+ // find the main gain
+ int maxGainOfAll = -1;
+ int maxGainOfAllPos = -1;
+ for (int i = 0; i < columnGroups.size(); i++) {
+ if (columnGroups.get(i).maxGain > maxGainOfAll) {
+ maxGainOfAll = columnGroups.get(i).maxGain;
+ maxGainOfAllPos = i;
+ }
+ }
+ if (maxGainOfAll <= 0) {
+ break;
+ }
+
+ // merge the group, create a new group
+ int source = maxGainOfAllPos;
+ int target = columnGroups.get(source).maxGainPosition;
+
+ ColumnGroup newGroup =
+ ColumnGroup.mergeColumnGroup(columnGroups.get(source),
columnGroups.get(target));
+
+ // remove the old groups
+ columnGroups.remove(target);
+ columnGroups.remove(source);
+
+ // load target into source
+ columnGroups.add(source, newGroup);
+
+ /** update * */
+
+ // update gains
+ // remove the target, and update the source
+ for (int i = 0; i < columnGroups.size(); i++) {
+ if (i != source) {
+ ColumnGroup g = columnGroups.get(i);
+ ColumnGroup gSource = columnGroups.get(source);
+ g.gains.remove(target);
+ int gain = computeGain(g, gSource, size);
+ g.gains.set(source, gain);
+ // update the maxgain in i
+ if ((g.maxGainPosition != source) && (g.maxGainPosition !=
target)) {
+ if (gain > g.maxGain) {
+ g.maxGain = gain;
+ g.maxGainPosition = source;
+ } else {
+ if (target < g.maxGainPosition) {
+ g.maxGainPosition -= 1;
+ }
+ }
+ } else {
+ g.maxGain = gain;
+ g.maxGainPosition = source;
+ for (int j = 0; j < g.gains.size(); j++) {
+ if (g.gains.get(j) > g.maxGain) {
+ g.maxGain = g.gains.get(j);
+ g.maxGainPosition = j;
+ }
+ }
+ }
+ // update the maxgain and data in source
+ gSource.gains.set(i, gain);
+ if (gain > gSource.maxGain) {
+ gSource.maxGain = gain;
+ gSource.maxGainPosition = i;
+ }
+ }
+ }
+ }
+ } catch (Exception e) {
+ LOGGER.error(e.getMessage());
+ }
+ }
+
+ public static int computeOverlap(ArrayList<BitMap> col1, ArrayList<BitMap>
col2, int size) {
+ int bitmapSize = ARRAY_SIZE;
+ int overlaps = 0;
+ for (int i = 0; i < col1.size(); i++) {
+ byte[] byteCol1 = null;
+ byte[] byteCol2 = null;
+ if (col1.get(i) != null) {
+ byteCol1 = col1.get(i).getByteArray();
+ }
+ if (col2.get(i) != null) {
+ byteCol2 = col2.get(i).getByteArray();
+ }
+ overlaps += computeOverlapForByte(byteCol1, byteCol2, bitmapSize);
+ }
+ int repeatOverlap = bitmapSize - (size % bitmapSize);
+ if (size % bitmapSize == 0) {
+ repeatOverlap = 0;
+ }
+ return overlaps - repeatOverlap;
+ }
+
+ public static int computeOverlapForByte(byte[] col1, byte[] col2, int size) {
+ if (col1 == null && col2 == null) {
+ return size;
+ }
+ int onesCount = 0;
+ if (col1 == null) {
+ for (int i = 0; i < size / Byte.SIZE; i++) {
+ onesCount += countOnesForByte(col2[i]);
+ }
+ return size - onesCount;
+ }
+ if (col2 == null) {
+ for (int i = 0; i < size / Byte.SIZE; i++) {
+ onesCount += countOnesForByte(col1[i]);
+ }
+ return size - onesCount;
+ }
+ for (int i = 0; i < size / Byte.SIZE; i++) {
+ onesCount += countOnesForByte((byte) (col1[i] | col2[i]));
+ }
+ return size - onesCount;
+ }
+
+ public static int computeGain(ColumnGroup g1, ColumnGroup g2, int size) {
+ int overlap = computeOverlap(g1.bitmap, g2.bitmap, size);
+ if (g1.columns.size() == 1 && g2.columns.size() == 1) {
+ // col - col
+ int gain = overlap * Long.SIZE - 2 * (g1.timeSeriesLength +
g2.timeSeriesLength - overlap);
+ return gain;
+ }
+ if (g1.columns.size() == 1) {
+ // col - group
+ int m_s_a = g1.timeSeriesLength;
+ int n_g_a = g2.columns.size();
+ int m_g_a = g2.timeSeriesLength;
+ int gain = overlap * Long.SIZE + n_g_a * m_g_a - (n_g_a + 1) * (m_s_a +
m_g_a - overlap);
+ return gain;
+ }
+ if (g2.columns.size() == 1) {
+ // group - col
+ int m_s_a = g2.timeSeriesLength;
+ int n_g_a = g1.columns.size();
+ int m_g_a = g1.timeSeriesLength;
+ int gain = overlap * Long.SIZE + n_g_a * m_g_a - (n_g_a + 1) * (m_s_a +
m_g_a - overlap);
+ return gain;
+ }
+
+ // group - group
+ int n_g_a = g1.columns.size();
+ int m_g_a = g1.timeSeriesLength;
+ int n_g_b = g2.columns.size();
+ int m_g_b = g2.timeSeriesLength;
+
+ int gain = overlap * Long.SIZE + (n_g_a + n_g_b) * overlap - n_g_a * m_g_b
- n_g_b * m_g_a;
+ return gain;
+ }
+
+ /** belows are some util functions* */
+ public static int countOnesForByte(byte x) {
+ return ((x >> 7) & 1)
+ + ((x >> 6) & 1)
+ + ((x >> 5) & 1)
+ + ((x >> 4) & 1)
+ + ((x >> 3) & 1)
+ + ((x >> 2) & 1)
+ + ((x >> 1) & 1)
+ + (x & 1);
+ }
+
+ public static int firstZero(byte x) {
+ for (int i = 0; i <= 7; i++) {
+ if (((x >> i) & 1) == 0) {
+ return i;
+ }
+ }
+ return -1;
+ }
+
+ public static int lastZero(byte x) {
+ for (int i = 7; i >= 0; i--) {
+ if (((x >> i) & 1) == 0) {
+ return i;
+ }
+ }
+ return -1;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
b/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
index 3501db0..5eaaae8 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
@@ -21,23 +21,24 @@ package org.apache.iotdb.db.engine.flush;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.flush.pool.FlushSubTaskPoolManager;
-import org.apache.iotdb.db.engine.memtable.IMemTable;
-import org.apache.iotdb.db.engine.memtable.IWritableMemChunk;
-import org.apache.iotdb.db.engine.memtable.IWritableMemChunkGroup;
+import org.apache.iotdb.db.engine.memtable.*;
import org.apache.iotdb.db.exception.runtime.FlushRunTimeException;
import org.apache.iotdb.db.metadata.idtable.entry.IDeviceID;
import org.apache.iotdb.db.rescon.SystemInfo;
import org.apache.iotdb.db.service.metrics.Metric;
import org.apache.iotdb.db.service.metrics.MetricsService;
import org.apache.iotdb.db.service.metrics.Tag;
+import org.apache.iotdb.db.utils.datastructure.AlignedTVList;
import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
import org.apache.iotdb.tsfile.write.chunk.IChunkWriter;
+import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@@ -122,6 +123,13 @@ public class MemTableFlushTask {
series.sortTvListForFlush();
sortTime += System.currentTimeMillis() - startTime;
encodingTaskQueue.put(series);
+ if (series instanceof AutoAlignedWritableMemChunk) {
+ /** perform our grouping strategy* */
+ List<IMeasurementSchema> schemaList =
+ ((AutoAlignedWritableMemChunk) series).getSchemaList();
+ AlignedTVList dataList = ((AutoAlignedWritableMemChunk)
series).getList();
+ FlushGroupingEngine.grouping(dataList, schemaList);
+ }
}
encodingTaskQueue.put(new EndChunkGroupIoTask());
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
b/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
index 57116e9..d067312 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
@@ -136,6 +136,25 @@ public abstract class AbstractMemTable implements
IMemTable {
return memChunkGroup;
}
+ private IWritableMemChunkGroup
createAutoAlignedMemChunkGroupIfNotExistAndGet(
+ IDeviceID deviceId, List<IMeasurementSchema> schemaList) {
+ IWritableMemChunkGroup memChunkGroup =
+ memTableMap.computeIfAbsent(
+ deviceId,
+ k -> {
+ seriesNumber += schemaList.size();
+ totalPointsNumThreshold += ((long) avgSeriesPointNumThreshold) *
schemaList.size();
+ return new AutoAlignedWritableMemChunkGroup(schemaList);
+ });
+ for (IMeasurementSchema schema : schemaList) {
+ if (!memChunkGroup.contains(schema.getMeasurementId())) {
+ seriesNumber++;
+ totalPointsNumThreshold += avgSeriesPointNumThreshold;
+ }
+ }
+ return memChunkGroup;
+ }
+
@Override
public void insert(InsertRowPlan insertRowPlan) {
// if this insert plan isn't from storage engine (mainly from test), we
should set a temp device
@@ -216,6 +235,47 @@ public abstract class AbstractMemTable implements
IMemTable {
}
@Override
+ public void insertAutoAlignedRow(InsertRowPlan insertRowPlan) {
+ // if this insert plan isn't from storage engine, we should set a temp
device id for it
+ if (insertRowPlan.getDeviceID() == null) {
+ insertRowPlan.setDeviceID(
+
DeviceIDFactory.getInstance().getDeviceID(insertRowPlan.getDevicePath()));
+ }
+
+ updatePlanIndexes(insertRowPlan.getIndex());
+ String[] measurements = insertRowPlan.getMeasurements();
+ List<IMeasurementSchema> schemaList = new ArrayList<>();
+ List<TSDataType> dataTypes = new ArrayList<>();
+ for (int i = 0; i < insertRowPlan.getMeasurements().length; i++) {
+ if (measurements[i] == null) {
+ continue;
+ }
+ IMeasurementSchema schema =
insertRowPlan.getMeasurementMNodes()[i].getSchema();
+ schemaList.add(schema);
+ dataTypes.add(schema.getType());
+ }
+ if (schemaList.isEmpty()) {
+ return;
+ }
+ memSize +=
+ MemUtils.getAlignedRecordsSize(dataTypes, insertRowPlan.getValues(),
disableMemControl);
+ writeAutoAlignedRow(
+ insertRowPlan.getDeviceID(),
+ schemaList,
+ insertRowPlan.getTime(),
+ insertRowPlan.getValues());
+ int pointsInserted =
+ insertRowPlan.getMeasurements().length -
insertRowPlan.getFailedMeasurementNumber();
+ totalPointsNum += pointsInserted;
+
+ if
(MetricConfigDescriptor.getInstance().getMetricConfig().getEnableMetric()) {
+ MetricsService.getInstance()
+ .getMetricManager()
+ .count(pointsInserted, Metric.QUANTITY.toString(),
Tag.NAME.toString(), METRIC_POINT_IN);
+ }
+ }
+
+ @Override
public void insertTablet(InsertTabletPlan insertTabletPlan, int start, int
end)
throws WriteProcessException {
updatePlanIndexes(insertTabletPlan.getIndex());
@@ -281,6 +341,17 @@ public abstract class AbstractMemTable implements
IMemTable {
memChunkGroup.write(insertTime, objectValue, schemaList);
}
+ @Override
+ public void writeAutoAlignedRow(
+ IDeviceID deviceId,
+ List<IMeasurementSchema> schemaList,
+ long insertTime,
+ Object[] objectValue) {
+ IWritableMemChunkGroup memChunkGroup =
+ createAutoAlignedMemChunkGroupIfNotExistAndGet(deviceId, schemaList);
+ memChunkGroup.write(insertTime, objectValue, schemaList);
+ }
+
@SuppressWarnings("squid:S3776") // high Cognitive Complexity
@Override
public void write(InsertTabletPlan insertTabletPlan, int start, int end) {
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/memtable/AutoAlignedWritableMemChunk.java
b/server/src/main/java/org/apache/iotdb/db/engine/memtable/AutoAlignedWritableMemChunk.java
new file mode 100755
index 0000000..4070c84
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/memtable/AutoAlignedWritableMemChunk.java
@@ -0,0 +1,367 @@
+/*
+ * 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.engine.memtable;
+
+import org.apache.iotdb.db.utils.datastructure.AlignedTVList;
+import org.apache.iotdb.db.utils.datastructure.TVList;
+import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.utils.BitMap;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.write.chunk.AutoAlignedChunkWriterImpl;
+import org.apache.iotdb.tsfile.write.chunk.IChunkWriter;
+import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public class AutoAlignedWritableMemChunk implements IWritableMemChunk {
+
+ private final Map<String, Integer> measurementIndexMap;
+ private final List<IMeasurementSchema> schemaList;
+ private AlignedTVList list;
+ private static final String UNSUPPORTED_TYPE = "Unsupported data type:";
+ private static final Logger LOGGER =
LoggerFactory.getLogger(AutoAlignedWritableMemChunk.class);
+
+ public AutoAlignedWritableMemChunk(List<IMeasurementSchema> schemaList) {
+ this.measurementIndexMap = new LinkedHashMap<>();
+ List<TSDataType> dataTypeList = new ArrayList<>();
+ this.schemaList = schemaList;
+ for (int i = 0; i < schemaList.size(); i++) {
+ measurementIndexMap.put(schemaList.get(i).getMeasurementId(), i);
+ dataTypeList.add(schemaList.get(i).getType());
+ }
+ this.list = AlignedTVList.newAlignedList(dataTypeList);
+ }
+
+ public Set<String> getAllMeasurements() {
+ return measurementIndexMap.keySet();
+ }
+
+ public boolean containsMeasurement(String measurementId) {
+ return measurementIndexMap.containsKey(measurementId);
+ }
+
+ @Override
+ public void putLong(long t, long v) {
+ throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
TSDataType.VECTOR);
+ }
+
+ @Override
+ public void putInt(long t, int v) {
+ throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
TSDataType.VECTOR);
+ }
+
+ @Override
+ public void putFloat(long t, float v) {
+ throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
TSDataType.VECTOR);
+ }
+
+ @Override
+ public void putDouble(long t, double v) {
+ throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
TSDataType.VECTOR);
+ }
+
+ @Override
+ public void putBinary(long t, Binary v) {
+ throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
TSDataType.VECTOR);
+ }
+
+ @Override
+ public void putBoolean(long t, boolean v) {
+ throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
TSDataType.VECTOR);
+ }
+
+ @Override
+ public void putAlignedValue(long t, Object[] v, int[] columnIndexArray) {
+ list.putAlignedValue(t, v, columnIndexArray);
+ }
+
+ @Override
+ public void putLongs(long[] t, long[] v, BitMap bitMap, int start, int end) {
+ throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
TSDataType.VECTOR);
+ }
+
+ @Override
+ public void putInts(long[] t, int[] v, BitMap bitMap, int start, int end) {
+ throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
TSDataType.VECTOR);
+ }
+
+ @Override
+ public void putFloats(long[] t, float[] v, BitMap bitMap, int start, int
end) {
+ throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
TSDataType.VECTOR);
+ }
+
+ @Override
+ public void putDoubles(long[] t, double[] v, BitMap bitMap, int start, int
end) {
+ throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
TSDataType.VECTOR);
+ }
+
+ @Override
+ public void putBinaries(long[] t, Binary[] v, BitMap bitMap, int start, int
end) {
+ throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
TSDataType.VECTOR);
+ }
+
+ @Override
+ public void putBooleans(long[] t, boolean[] v, BitMap bitMap, int start, int
end) {
+ throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
TSDataType.VECTOR);
+ }
+
+ @Override
+ public void putAlignedValues(
+ long[] t, Object[] v, BitMap[] bitMaps, int[] columnIndexArray, int
start, int end) {
+ list.putAlignedValues(t, v, bitMaps, columnIndexArray, start, end);
+ }
+
+ @Override
+ public void write(long insertTime, Object objectValue) {
+ throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
TSDataType.VECTOR);
+ }
+
+ @Override
+ public void writeAlignedValue(
+ long insertTime, Object[] objectValue, List<IMeasurementSchema>
schemaList) {
+ int[] columnIndexArray = checkColumnsInInsertPlan(schemaList);
+ putAlignedValue(insertTime, objectValue, columnIndexArray);
+ }
+
+ @Override
+ public void write(
+ long[] times, Object valueList, BitMap bitMap, TSDataType dataType, int
start, int end) {
+ throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
TSDataType.VECTOR);
+ }
+
+ @Override
+ public void writeAlignedValues(
+ long[] times,
+ Object[] valueList,
+ BitMap[] bitMaps,
+ List<IMeasurementSchema> schemaList,
+ int start,
+ int end) {
+ int[] columnIndexArray = checkColumnsInInsertPlan(schemaList);
+ putAlignedValues(times, valueList, bitMaps, columnIndexArray, start, end);
+ }
+
+ private int[] checkColumnsInInsertPlan(List<IMeasurementSchema>
schemaListInInsertPlan) {
+ Map<String, Integer> measurementIdsInInsertPlan = new HashMap<>();
+ for (int i = 0; i < schemaListInInsertPlan.size(); i++) {
+
measurementIdsInInsertPlan.put(schemaListInInsertPlan.get(i).getMeasurementId(),
i);
+ if
(!containsMeasurement(schemaListInInsertPlan.get(i).getMeasurementId())) {
+ this.measurementIndexMap.put(
+ schemaListInInsertPlan.get(i).getMeasurementId(),
measurementIndexMap.size());
+ this.schemaList.add(schemaListInInsertPlan.get(i));
+ this.list.extendColumn(schemaListInInsertPlan.get(i).getType());
+ }
+ }
+ int[] columnIndexArray = new int[measurementIndexMap.size()];
+ measurementIndexMap.forEach(
+ (measurementId, i) -> {
+ columnIndexArray[i] =
measurementIdsInInsertPlan.getOrDefault(measurementId, -1);
+ });
+ return columnIndexArray;
+ }
+
+ @Override
+ public TVList getTVList() {
+ return list;
+ }
+
+ @Override
+ public long count() {
+ return (long) list.rowCount() * measurementIndexMap.size();
+ }
+
+ public long alignedListSize() {
+ return list.rowCount();
+ }
+
+ @Override
+ public IMeasurementSchema getSchema() {
+ return null;
+ }
+
+ public List<IMeasurementSchema> getSchemaList() {
+ return schemaList;
+ }
+
+ public AlignedTVList getList() {
+ return list;
+ }
+
+ @Override
+ public TVList getSortedTvListForQuery() {
+ sortTVList();
+ // increase reference count
+ list.increaseReferenceCount();
+ return list;
+ }
+
+ @Override
+ public TVList getSortedTvListForQuery(List<IMeasurementSchema> schemaList) {
+ sortTVList();
+ // increase reference count
+ list.increaseReferenceCount();
+ List<Integer> columnIndexList = new ArrayList<>();
+ for (IMeasurementSchema measurementSchema : schemaList) {
+ columnIndexList.add(
+
measurementIndexMap.getOrDefault(measurementSchema.getMeasurementId(), -1));
+ }
+ return list.getTvListByColumnIndex(columnIndexList);
+ }
+
+ private void sortTVList() {
+ // check reference count
+ if ((list.getReferenceCount() > 0 && !list.isSorted())) {
+ list = list.clone();
+ }
+
+ if (!list.isSorted()) {
+ list.sort();
+ }
+ }
+
+ @Override
+ public void sortTvListForFlush() {
+ sortTVList();
+ }
+
+ @Override
+ public int delete(long lowerBound, long upperBound) {
+ return list.delete(lowerBound, upperBound);
+ }
+
+ public Pair<Integer, Boolean> deleteDataFromAColumn(
+ long lowerBound, long upperBound, String measurementId) {
+ return list.delete(lowerBound, upperBound,
measurementIndexMap.get(measurementId));
+ }
+
+ public void removeColumn(String measurementId) {
+ list.deleteColumn(measurementIndexMap.get(measurementId));
+ IMeasurementSchema schemaToBeRemoved =
schemaList.get(measurementIndexMap.get(measurementId));
+ schemaList.remove(schemaToBeRemoved);
+ measurementIndexMap.clear();
+ for (int i = 0; i < schemaList.size(); i++) {
+ measurementIndexMap.put(schemaList.get(i).getMeasurementId(), i);
+ }
+ }
+
+ @Override
+ public IChunkWriter createIChunkWriter() {
+ return new AutoAlignedChunkWriterImpl(schemaList);
+ }
+
+ @Override
+ public void encode(IChunkWriter chunkWriter) {
+ AutoAlignedChunkWriterImpl autoAlignedChunkWriter =
(AutoAlignedChunkWriterImpl) chunkWriter;
+ List<Integer> timeDuplicateAlignedRowIndexList = null;
+ for (int sortedRowIndex = 0; sortedRowIndex < list.rowCount();
sortedRowIndex++) {
+ long time = list.getTime(sortedRowIndex);
+
+ // skip duplicated data
+ if ((sortedRowIndex + 1 < list.rowCount() && (time ==
list.getTime(sortedRowIndex + 1)))) {
+ // record the time duplicated row index list for vector type
+ if (timeDuplicateAlignedRowIndexList == null) {
+ timeDuplicateAlignedRowIndexList = new ArrayList<>();
+
timeDuplicateAlignedRowIndexList.add(list.getValueIndex(sortedRowIndex));
+ }
+ timeDuplicateAlignedRowIndexList.add(list.getValueIndex(sortedRowIndex
+ 1));
+ continue;
+ }
+ List<TSDataType> dataTypes = list.getTsDataTypes();
+ int originRowIndex = list.getValueIndex(sortedRowIndex);
+ for (int columnIndex = 0; columnIndex < dataTypes.size(); columnIndex++)
{
+ // write the time duplicated rows
+ if (timeDuplicateAlignedRowIndexList != null
+ && !timeDuplicateAlignedRowIndexList.isEmpty()) {
+ originRowIndex =
+ list.getValidRowIndexForTimeDuplicatedRows(
+ timeDuplicateAlignedRowIndexList, columnIndex);
+ }
+ boolean isNull = list.isValueMarked(originRowIndex, columnIndex);
+ switch (dataTypes.get(columnIndex)) {
+ case BOOLEAN:
+ autoAlignedChunkWriter.write(
+ time, list.getBooleanByValueIndex(originRowIndex,
columnIndex), isNull);
+ break;
+ case INT32:
+ autoAlignedChunkWriter.write(
+ time, list.getIntByValueIndex(originRowIndex, columnIndex),
isNull);
+ break;
+ case INT64:
+ autoAlignedChunkWriter.write(
+ time, list.getLongByValueIndex(originRowIndex, columnIndex),
isNull);
+ break;
+ case FLOAT:
+ autoAlignedChunkWriter.write(
+ time, list.getFloatByValueIndex(originRowIndex, columnIndex),
isNull);
+ break;
+ case DOUBLE:
+ autoAlignedChunkWriter.write(
+ time, list.getDoubleByValueIndex(originRowIndex, columnIndex),
isNull);
+ break;
+ case TEXT:
+ autoAlignedChunkWriter.write(
+ time, list.getBinaryByValueIndex(originRowIndex, columnIndex),
isNull);
+ break;
+ default:
+ LOGGER.error(
+ "AlignedWritableMemChunk does not support data type: {}",
+ dataTypes.get(columnIndex));
+ break;
+ }
+ }
+ autoAlignedChunkWriter.write(time);
+ timeDuplicateAlignedRowIndexList = null;
+ }
+ }
+
+ @Override
+ public void release() {
+ if (list.getReferenceCount() == 0) {
+ list.clear();
+ }
+ }
+
+ @Override
+ public long getFirstPoint() {
+ if (list.rowCount() == 0) {
+ return Long.MAX_VALUE;
+ }
+ return getSortedTvListForQuery().getTimeValuePair(0).getTimestamp();
+ }
+
+ @Override
+ public long getLastPoint() {
+ if (list.rowCount() == 0) {
+ return Long.MIN_VALUE;
+ }
+ return getSortedTvListForQuery()
+ .getTimeValuePair(getSortedTvListForQuery().rowCount() - 1)
+ .getTimestamp();
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/memtable/AutoAlignedWritableMemChunkGroup.java
b/server/src/main/java/org/apache/iotdb/db/engine/memtable/AutoAlignedWritableMemChunkGroup.java
new file mode 100644
index 0000000..8801f82
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/memtable/AutoAlignedWritableMemChunkGroup.java
@@ -0,0 +1,116 @@
+/*
+ * 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.engine.memtable;
+
+import org.apache.iotdb.db.metadata.path.AlignedPath;
+import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.tsfile.utils.BitMap;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
+
+import java.util.*;
+
+public class AutoAlignedWritableMemChunkGroup implements
IWritableMemChunkGroup {
+
+ private AutoAlignedWritableMemChunk memChunk;
+
+ public AutoAlignedWritableMemChunkGroup(List<IMeasurementSchema> schemaList)
{
+ memChunk = new AutoAlignedWritableMemChunk(schemaList);
+ }
+
+ @Override
+ public void writeValues(
+ long[] times,
+ Object[] columns,
+ BitMap[] bitMaps,
+ List<IMeasurementSchema> schemaList,
+ int start,
+ int end) {
+ memChunk.writeAlignedValues(times, columns, bitMaps, schemaList, start,
end);
+ }
+
+ @Override
+ public void release() {
+ memChunk.release();
+ }
+
+ @Override
+ public long count() {
+ return memChunk.count();
+ }
+
+ /**
+ * Check whether this MemChunkGroup contains a measurement. If a
VECTOR_PLACEHOLDER passed from
+ * outer, always return true because AlignedMemChunkGroup existing.
+ */
+ @Override
+ public boolean contains(String measurement) {
+ // used for calculate memtable size
+ if (AlignedPath.VECTOR_PLACEHOLDER.equals(measurement)) {
+ return true;
+ }
+ return memChunk.containsMeasurement(measurement);
+ }
+
+ @Override
+ public void write(long insertTime, Object[] objectValue,
List<IMeasurementSchema> schemaList) {
+ memChunk.writeAlignedValue(insertTime, objectValue, schemaList);
+ }
+
+ @Override
+ public Map<String, IWritableMemChunk> getMemChunkMap() {
+ if (memChunk.count() == 0) {
+ return Collections.emptyMap();
+ }
+ return Collections.singletonMap("", memChunk);
+ }
+
+ @Override
+ public int delete(
+ PartialPath originalPath, PartialPath devicePath, long startTimestamp,
long endTimestamp) {
+ int deletedPointsNumber = 0;
+ Set<String> measurements = memChunk.getAllMeasurements();
+ List<String> columnsToBeRemoved = new ArrayList<>();
+ for (String measurement : measurements) {
+ PartialPath fullPath = devicePath.concatNode(measurement);
+ if (originalPath.matchFullPath(fullPath)) {
+ Pair<Integer, Boolean> deleteInfo =
+ memChunk.deleteDataFromAColumn(startTimestamp, endTimestamp,
measurement);
+ deletedPointsNumber += deleteInfo.left;
+ if (Boolean.TRUE.equals(deleteInfo.right)) {
+ columnsToBeRemoved.add(measurement);
+ }
+ }
+ }
+ for (String columnToBeRemoved : columnsToBeRemoved) {
+ memChunk.removeColumn(columnToBeRemoved);
+ }
+ return deletedPointsNumber;
+ }
+
+ @Override
+ public long getCurrentTVListSize(String measurement) {
+ return memChunk.getTVList().rowCount();
+ }
+
+ public AutoAlignedWritableMemChunk getAlignedMemChunk() {
+ return memChunk;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/memtable/IMemTable.java
b/server/src/main/java/org/apache/iotdb/db/engine/memtable/IMemTable.java
index 69aeb47..0867197 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/memtable/IMemTable.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/memtable/IMemTable.java
@@ -57,6 +57,12 @@ public interface IMemTable {
List<IMeasurementSchema> schemaList,
long insertTime,
Object[] objectValue);
+
+ void writeAutoAlignedRow(
+ IDeviceID deviceId,
+ List<IMeasurementSchema> schemaList,
+ long insertTime,
+ Object[] objectValue);
/**
* write data in the range [start, end). Null value in each column values
will be replaced by the
* subsequent non-null value, e.g., {1, null, 3, null, 5} will be {1, 3, 5,
null, 5}
@@ -100,6 +106,8 @@ public interface IMemTable {
void insertAlignedRow(InsertRowPlan insertRowPlan);
+ void insertAutoAlignedRow(InsertRowPlan insertRowPlan);
+
/**
* insert tablet into this memtable. The rows to be inserted are in the
range [start, end). Null
* value in each column values will be replaced by the subsequent non-null
value, e.g., {1, null,
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index 71eaef1..fda3d04 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -27,10 +27,7 @@ import org.apache.iotdb.db.engine.flush.FlushListener;
import org.apache.iotdb.db.engine.flush.FlushManager;
import org.apache.iotdb.db.engine.flush.MemTableFlushTask;
import org.apache.iotdb.db.engine.flush.NotifyFlushMemTable;
-import org.apache.iotdb.db.engine.memtable.AlignedWritableMemChunk;
-import org.apache.iotdb.db.engine.memtable.AlignedWritableMemChunkGroup;
-import org.apache.iotdb.db.engine.memtable.IMemTable;
-import org.apache.iotdb.db.engine.memtable.PrimitiveMemTable;
+import org.apache.iotdb.db.engine.memtable.*;
import org.apache.iotdb.db.engine.modification.Deletion;
import org.apache.iotdb.db.engine.modification.Modification;
import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
@@ -217,7 +214,9 @@ public class TsFileProcessor {
long[] memIncrements = null;
if (enableMemControl) {
- if (insertRowPlan.isAligned()) {
+ if (insertRowPlan.isAutoAligned()) {
+ memIncrements = checkAutoAlignedMemCostAndAddToTspInfo(insertRowPlan);
+ } else if (insertRowPlan.isAligned()) {
memIncrements = checkAlignedMemCostAndAddToTspInfo(insertRowPlan);
} else {
memIncrements = checkMemCostAndAddToTspInfo(insertRowPlan);
@@ -239,7 +238,9 @@ public class TsFileProcessor {
}
}
- if (insertRowPlan.isAligned()) {
+ if (insertRowPlan.isAutoAligned()) {
+ workMemTable.insertAutoAlignedRow(insertRowPlan);
+ } else if (insertRowPlan.isAligned()) {
workMemTable.insertAlignedRow(insertRowPlan);
} else {
workMemTable.insert(insertRowPlan);
@@ -384,6 +385,60 @@ public class TsFileProcessor {
return new long[] {memTableIncrement, textDataIncrement,
chunkMetadataIncrement};
}
+ private long[] checkAutoAlignedMemCostAndAddToTspInfo(InsertRowPlan
insertRowPlan)
+ throws WriteProcessException {
+ // memory of increased PrimitiveArray and TEXT values, e.g., add a
long[128], add 128*8
+ long memTableIncrement = 0L;
+ long textDataIncrement = 0L;
+ long chunkMetadataIncrement = 0L;
+ AutoAlignedWritableMemChunk autoAlignedMemChunk = null;
+ // get device id
+ IDeviceID deviceID = null;
+ try {
+ deviceID = getDeviceID(insertRowPlan.getDevicePath().getFullPath());
+ } catch (IllegalPathException e) {
+ throw new WriteProcessException(e);
+ }
+
+ if (workMemTable.checkIfChunkDoesNotExist(deviceID,
AlignedPath.VECTOR_PLACEHOLDER)) {
+ // ChunkMetadataIncrement
+ chunkMetadataIncrement +=
+ ChunkMetadata.calculateRamSize(AlignedPath.VECTOR_PLACEHOLDER,
TSDataType.VECTOR)
+ * insertRowPlan.getDataTypes().length;
+ memTableIncrement +=
AlignedTVList.alignedTvListArrayMemCost(insertRowPlan.getDataTypes());
+ } else {
+ // here currentChunkPointNum >= 1
+ long currentChunkPointNum =
+ workMemTable.getCurrentTVListSize(deviceID,
AlignedPath.VECTOR_PLACEHOLDER);
+ memTableIncrement +=
+ (currentChunkPointNum % PrimitiveArrayManager.ARRAY_SIZE) == 0
+ ?
AlignedTVList.alignedTvListArrayMemCost(insertRowPlan.getDataTypes())
+ : 0;
+ autoAlignedMemChunk =
+ ((AutoAlignedWritableMemChunkGroup)
workMemTable.getMemTableMap().get(deviceID))
+ .getAlignedMemChunk();
+ }
+ for (int i = 0; i < insertRowPlan.getDataTypes().length; i++) {
+ // skip failed Measurements
+ if (insertRowPlan.getDataTypes()[i] == null ||
insertRowPlan.getMeasurements()[i] == null) {
+ continue;
+ }
+ // extending the column of aligned mem chunk
+ if (autoAlignedMemChunk != null
+ &&
!autoAlignedMemChunk.containsMeasurement(insertRowPlan.getMeasurements()[i])) {
+ memTableIncrement +=
+ (autoAlignedMemChunk.alignedListSize() /
PrimitiveArrayManager.ARRAY_SIZE + 1)
+ * insertRowPlan.getDataTypes()[i].getDataTypeSize();
+ }
+ // TEXT data mem size
+ if (insertRowPlan.getDataTypes()[i] == TSDataType.TEXT) {
+ textDataIncrement += MemUtils.getBinarySize((Binary)
insertRowPlan.getValues()[i]);
+ }
+ }
+ updateMemoryInfo(memTableIncrement, chunkMetadataIncrement,
textDataIncrement);
+ return new long[] {memTableIncrement, textDataIncrement,
chunkMetadataIncrement};
+ }
+
@SuppressWarnings("squid:S3776") // high Cognitive Complexity
private long[] checkAlignedMemCostAndAddToTspInfo(InsertRowPlan
insertRowPlan)
throws WriteProcessException {
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
index a56f57e..2bbf1ed 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
@@ -60,26 +60,7 @@ import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
-import org.apache.iotdb.db.qp.physical.sys.ActivateTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.AppendTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.AutoCreateDeviceMNodePlan;
-import org.apache.iotdb.db.qp.physical.sys.ChangeAliasPlan;
-import org.apache.iotdb.db.qp.physical.sys.ChangeTagOffsetPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateContinuousQueryPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan;
-import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.DropContinuousQueryPlan;
-import org.apache.iotdb.db.qp.physical.sys.DropTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.PruneTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
-import org.apache.iotdb.db.qp.physical.sys.SetTTLPlan;
-import org.apache.iotdb.db.qp.physical.sys.SetTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.ShowDevicesPlan;
-import org.apache.iotdb.db.qp.physical.sys.ShowTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.UnsetTemplatePlan;
+import org.apache.iotdb.db.qp.physical.sys.*;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.dataset.ShowDevicesResult;
import org.apache.iotdb.db.query.dataset.ShowTimeSeriesResult;
@@ -454,6 +435,11 @@ public class MManager {
(CreateAlignedTimeSeriesPlan) plan;
createAlignedTimeSeries(createAlignedTimeSeriesPlan);
break;
+ case CREATE_AUTOALIGNED_TIMESERIES:
+ CreateAutoAlignedTimeSeriesPlan createAutoAlignedTimeSeriesPlan =
+ (CreateAutoAlignedTimeSeriesPlan) plan;
+ createAutoAlignedTimeSeries(createAutoAlignedTimeSeriesPlan);
+ break;
case DELETE_TIMESERIES:
DeleteTimeSeriesPlan deleteTimeSeriesPlan = (DeleteTimeSeriesPlan)
plan;
// cause we only has one path for one DeleteTimeSeriesPlan
@@ -720,6 +706,74 @@ public class MManager {
}
}
+ public void createAutoAlignedTimeSeries(
+ PartialPath prefixPath,
+ List<String> measurements,
+ List<TSDataType> dataTypes,
+ List<TSEncoding> encodings,
+ List<CompressionType> compressors)
+ throws MetadataException {
+ createAutoAlignedTimeSeries(
+ new CreateAutoAlignedTimeSeriesPlan(
+ prefixPath, measurements, dataTypes, encodings, compressors,
null));
+ }
+
+ /**
+ * create aligned timeseries
+ *
+ * @param plan CreateAlignedTimeSeriesPlan
+ */
+ public void createAutoAlignedTimeSeries(CreateAutoAlignedTimeSeriesPlan plan)
+ throws MetadataException {
+ if (!allowToCreateNewSeries) {
+ throw new MetadataException(
+ "IoTDB system load is too large to create timeseries, "
+ + "please increase MAX_HEAP_SIZE in iotdb-env.sh/bat and
restart");
+ }
+ try {
+ PartialPath prefixPath = plan.getPrefixPath();
+ List<String> measurements = plan.getMeasurements();
+ List<TSDataType> dataTypes = plan.getDataTypes();
+ List<TSEncoding> encodings = plan.getEncodings();
+
+ for (int i = 0; i < measurements.size(); i++) {
+ SchemaUtils.checkDataTypeWithEncoding(dataTypes.get(i),
encodings.get(i));
+ }
+
+ ensureStorageGroup(prefixPath);
+
+ // create time series in MTree
+ mtree.createAutoAlignedTimeseries(
+ prefixPath,
+ measurements,
+ plan.getDataTypes(),
+ plan.getEncodings(),
+ plan.getCompressors());
+
+ // the cached mNode may be replaced by new entityMNode in mtree
+ mNodeCache.invalidate(prefixPath);
+
+ // update statistics and schemaDataTypeNumMap
+ totalSeriesNumber.addAndGet(measurements.size());
+ if (totalSeriesNumber.get() * ESTIMATED_SERIES_SIZE >=
MTREE_SIZE_THRESHOLD) {
+ logger.warn("Current series number {} is too large...",
totalSeriesNumber);
+ allowToCreateNewSeries = false;
+ }
+ // write log
+ if (!isRecovering) {
+ logWriter.createAutoAlignedTimeseries(plan);
+ }
+ } catch (IOException e) {
+ throw new MetadataException(e);
+ }
+
+ // update id table
+ if (config.isEnableIDTable()) {
+ IDTable idTable =
IDTableManager.getInstance().getIDTable(plan.getPrefixPath());
+ idTable.createAutoAlignedTimeseries(plan);
+ }
+ }
+
private void ensureStorageGroup(PartialPath path) throws MetadataException {
try {
mtree.getBelongedStorageGroup(path);
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/idtable/IDTable.java
b/server/src/main/java/org/apache/iotdb/db/metadata/idtable/IDTable.java
index 5ab0d47..93a377d 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/idtable/IDTable.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/idtable/IDTable.java
@@ -33,6 +33,7 @@ import org.apache.iotdb.db.metadata.path.MeasurementPath;
import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateAutoAlignedTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
import org.apache.iotdb.db.utils.TestOnly;
import org.apache.iotdb.tsfile.read.TimeValuePair;
@@ -52,6 +53,14 @@ public interface IDTable {
IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
/**
+ * create autoaligned timeseries
+ *
+ * @param plan create aligned timeseries plan
+ * @throws MetadataException if the device is not aligned, throw it
+ */
+ void createAutoAlignedTimeseries(CreateAutoAlignedTimeSeriesPlan plan)
throws MetadataException;
+
+ /**
* create aligned timeseries
*
* @param plan create aligned timeseries plan
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/idtable/IDTableHashmapImpl.java
b/server/src/main/java/org/apache/iotdb/db/metadata/idtable/IDTableHashmapImpl.java
index fb21776..b07dcab 100644
---
a/server/src/main/java/org/apache/iotdb/db/metadata/idtable/IDTableHashmapImpl.java
+++
b/server/src/main/java/org/apache/iotdb/db/metadata/idtable/IDTableHashmapImpl.java
@@ -35,6 +35,7 @@ import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateAutoAlignedTimeSeriesPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.utils.TestOnly;
@@ -110,6 +111,34 @@ public class IDTableHashmapImpl implements IDTable {
}
/**
+ * create autoaligned timeseries
+ *
+ * @param plan create aligned timeseries plan
+ * @throws MetadataException if the device is not aligned, throw it
+ */
+ public synchronized void
createAutoAlignedTimeseries(CreateAutoAlignedTimeSeriesPlan plan)
+ throws MetadataException {
+ DeviceEntry deviceEntry =
+ getDeviceEntryWithAutoAlignedCheck(plan.getPrefixPath().toString(),
true);
+
+ for (int i = 0; i < plan.getMeasurements().size(); i++) {
+ PartialPath fullPath =
+ new PartialPath(plan.getPrefixPath().toString(),
plan.getMeasurements().get(i));
+ SchemaEntry schemaEntry =
+ new SchemaEntry(
+ plan.getDataTypes().get(i),
+ plan.getEncodings().get(i),
+ plan.getCompressors().get(i),
+ deviceEntry.getDeviceID(),
+ fullPath,
+ true,
+ true,
+ IDiskSchemaManager);
+ deviceEntry.putSchemaEntry(plan.getMeasurements().get(i), schemaEntry);
+ }
+ }
+
+ /**
* create timeseries
*
* @param plan create timeseries plan
@@ -388,6 +417,33 @@ public class IDTableHashmapImpl implements IDTable {
return deviceEntry;
}
+ private DeviceEntry getDeviceEntryWithAutoAlignedCheck(String deviceName,
boolean isAutoAligned)
+ throws MetadataException {
+ IDeviceID deviceID = DeviceIDFactory.getInstance().getDeviceID(deviceName);
+ int slot = calculateSlot(deviceID);
+
+ DeviceEntry deviceEntry = idTables[slot].get(deviceID);
+ // new device
+ if (deviceEntry == null) {
+ deviceEntry = new DeviceEntry(deviceID);
+ deviceEntry.setAutoAligned(isAutoAligned);
+ idTables[slot].put(deviceID, deviceEntry);
+
+ return deviceEntry;
+ }
+
+ // check aligned
+ if (deviceEntry.isAligned() != isAutoAligned) {
+ throw new MetadataException(
+ String.format(
+ "Timeseries under path [%s]'s align value is [%b], which is not
consistent with insert plan",
+ deviceName, deviceEntry.isAligned()));
+ }
+
+ // reuse device entry in map
+ return deviceEntry;
+ }
+
/**
* calculate slot that this deviceID should in
*
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/idtable/entry/DeviceEntry.java
b/server/src/main/java/org/apache/iotdb/db/metadata/idtable/entry/DeviceEntry.java
index e5cbd90..7fbb35c 100644
---
a/server/src/main/java/org/apache/iotdb/db/metadata/idtable/entry/DeviceEntry.java
+++
b/server/src/main/java/org/apache/iotdb/db/metadata/idtable/entry/DeviceEntry.java
@@ -34,6 +34,7 @@ public class DeviceEntry {
Map<String, SchemaEntry> measurementMap;
boolean isAligned;
+ boolean isAutoAligned;
// for managing last time
// time partition -> last time
@@ -86,10 +87,18 @@ public class DeviceEntry {
return isAligned;
}
+ public boolean isAutoAligned() {
+ return isAutoAligned;
+ }
+
public void setAligned(boolean aligned) {
isAligned = aligned;
}
+ public void setAutoAligned(boolean autoAligned) {
+ isAutoAligned = autoAligned;
+ }
+
public IDeviceID getDeviceID() {
return deviceID;
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/idtable/entry/DiskSchemaEntry.java
b/server/src/main/java/org/apache/iotdb/db/metadata/idtable/entry/DiskSchemaEntry.java
index 556759b..bfb984e 100644
---
a/server/src/main/java/org/apache/iotdb/db/metadata/idtable/entry/DiskSchemaEntry.java
+++
b/server/src/main/java/org/apache/iotdb/db/metadata/idtable/entry/DiskSchemaEntry.java
@@ -50,6 +50,9 @@ public class DiskSchemaEntry {
// whether this device is aligned
public boolean isAligned;
+ // whether this device is autoaligned
+ public boolean isAutoAligned;
+
// this entry's serialized size
public transient long entrySize;
@@ -72,6 +75,25 @@ public class DiskSchemaEntry {
this.isAligned = isAligned;
}
+ public DiskSchemaEntry(
+ String deviceID,
+ String seriesKey,
+ String measurementName,
+ byte type,
+ byte encoding,
+ byte compressor,
+ boolean isAligned,
+ boolean isAutoAligned) {
+ this.deviceID = deviceID;
+ this.seriesKey = seriesKey;
+ this.measurementName = measurementName;
+ this.type = type;
+ this.encoding = encoding;
+ this.compressor = compressor;
+ this.isAligned = isAligned;
+ this.isAutoAligned = isAutoAligned;
+ }
+
public int serialize(OutputStream outputStream) throws IOException {
int byteLen = 0;
byteLen += ReadWriteIOUtils.write(deviceID, outputStream);
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/idtable/entry/SchemaEntry.java
b/server/src/main/java/org/apache/iotdb/db/metadata/idtable/entry/SchemaEntry.java
index 77a07e1..f660979 100644
---
a/server/src/main/java/org/apache/iotdb/db/metadata/idtable/entry/SchemaEntry.java
+++
b/server/src/main/java/org/apache/iotdb/db/metadata/idtable/entry/SchemaEntry.java
@@ -109,6 +109,37 @@ public class SchemaEntry implements ILastCacheContainer {
}
}
+ public SchemaEntry(
+ TSDataType dataType,
+ TSEncoding encoding,
+ CompressionType compressionType,
+ IDeviceID deviceID,
+ PartialPath fullPath,
+ boolean isAligned,
+ boolean isAutoAligned,
+ IDiskSchemaManager IDiskSchemaManager) {
+ schema |= dataType.serialize();
+ schema |= (((long) encoding.serialize()) << 8);
+ schema |= (((long) compressionType.serialize()) << 16);
+
+ lastTime = Long.MIN_VALUE;
+
+ // write log file
+ if (config.isEnableIDTableLogFile()) {
+ DiskSchemaEntry diskSchemaEntry =
+ new DiskSchemaEntry(
+ deviceID.toStringID(),
+ fullPath.getFullPath(),
+ fullPath.getMeasurement(),
+ dataType.serialize(),
+ encoding.serialize(),
+ compressionType.serialize(),
+ isAligned,
+ isAutoAligned);
+ schema |= (IDiskSchemaManager.serialize(diskSchemaEntry) << 25);
+ }
+ }
+
/**
* get ts data type from long value of schema
*
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/logfile/MLogWriter.java
b/server/src/main/java/org/apache/iotdb/db/metadata/logfile/MLogWriter.java
index 303322d..fc347c1 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/logfile/MLogWriter.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/logfile/MLogWriter.java
@@ -25,27 +25,7 @@ import org.apache.iotdb.db.metadata.mnode.IMeasurementMNode;
import org.apache.iotdb.db.metadata.mnode.IStorageGroupMNode;
import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
-import org.apache.iotdb.db.qp.physical.sys.ActivateTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.AppendTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.AutoCreateDeviceMNodePlan;
-import org.apache.iotdb.db.qp.physical.sys.ChangeAliasPlan;
-import org.apache.iotdb.db.qp.physical.sys.ChangeTagOffsetPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateContinuousQueryPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan;
-import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.DropContinuousQueryPlan;
-import org.apache.iotdb.db.qp.physical.sys.DropTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.MNodePlan;
-import org.apache.iotdb.db.qp.physical.sys.MeasurementMNodePlan;
-import org.apache.iotdb.db.qp.physical.sys.PruneTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
-import org.apache.iotdb.db.qp.physical.sys.SetTTLPlan;
-import org.apache.iotdb.db.qp.physical.sys.SetTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.StorageGroupMNodePlan;
-import org.apache.iotdb.db.qp.physical.sys.UnsetTemplatePlan;
+import org.apache.iotdb.db.qp.physical.sys.*;
import org.apache.iotdb.db.writelog.io.LogWriter;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -128,6 +108,11 @@ public class MLogWriter implements AutoCloseable {
putLog(createAlignedTimeSeriesPlan);
}
+ public void createAutoAlignedTimeseries(
+ CreateAutoAlignedTimeSeriesPlan createAutoAlignedTimeSeriesPlan) throws
IOException {
+ putLog(createAutoAlignedTimeSeriesPlan);
+ }
+
public void deleteTimeseries(DeleteTimeSeriesPlan deleteTimeSeriesPlan)
throws IOException {
putLog(deleteTimeSeriesPlan);
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/mnode/EntityMNode.java
b/server/src/main/java/org/apache/iotdb/db/metadata/mnode/EntityMNode.java
index 7bbb69d..62af863 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/mnode/EntityMNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/mnode/EntityMNode.java
@@ -37,6 +37,8 @@ public class EntityMNode extends InternalMNode implements
IEntityMNode {
private volatile boolean isAligned = false;
+ private volatile boolean isAutoAligned = false;
+
private volatile Map<String, ILastCacheContainer> lastCacheMap = null;
/**
@@ -111,10 +113,20 @@ public class EntityMNode extends InternalMNode implements
IEntityMNode {
}
@Override
+ public boolean isAutoAligned() {
+ return isAutoAligned;
+ }
+
+ @Override
public void setAligned(boolean isAligned) {
this.isAligned = isAligned;
}
+ @Override
+ public void setAutoAligned(boolean isAutoAligned) {
+ this.isAutoAligned = isAutoAligned;
+ }
+
public ILastCacheContainer getLastCacheContainer(String measurementId) {
checkLastCacheMap();
return lastCacheMap.computeIfAbsent(measurementId, k -> new
LastCacheContainer());
diff --git
a/server/src/main/java/org/apache/iotdb/db/metadata/mnode/IEntityMNode.java
b/server/src/main/java/org/apache/iotdb/db/metadata/mnode/IEntityMNode.java
index a6780cb..0f3885d 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/mnode/IEntityMNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/mnode/IEntityMNode.java
@@ -39,8 +39,12 @@ public interface IEntityMNode extends IMNode {
boolean isAligned();
+ boolean isAutoAligned();
+
void setAligned(boolean isAligned);
+ void setAutoAligned(boolean isAutoAligned);
+
ILastCacheContainer getLastCacheContainer(String measurementId);
Map<String, ILastCacheContainer> getTemplateLastCaches();
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTree.java
b/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTree.java
index bddc4cf..fc56048 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTree.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTree.java
@@ -486,6 +486,69 @@ public class MTree implements Serializable {
}
}
+ /**
+ * Create aligned timeseries with full paths from root to one leaf node.
Before creating
+ * timeseries, the * storage group should be set first, throw exception
otherwise
+ *
+ * @param devicePath device path
+ * @param measurements measurements list
+ * @param dataTypes data types list
+ * @param encodings encodings list
+ * @param compressors compressor
+ */
+ public void createAutoAlignedTimeseries(
+ PartialPath devicePath,
+ List<String> measurements,
+ List<TSDataType> dataTypes,
+ List<TSEncoding> encodings,
+ List<CompressionType> compressors)
+ throws MetadataException {
+ MetaFormatUtils.checkSchemaMeasurementNames(measurements);
+ Pair<IMNode, Template> pair = checkAndAutoCreateInternalPath(devicePath);
+ IMNode cur = pair.left;
+ Template upperTemplate = pair.right;
+
+ // synchronize check and add, we need addChild and add Alias become atomic
operation
+ // only write on mtree will be synchronized
+ synchronized (this) {
+ for (String measurement : measurements) {
+ if (cur.hasChild(measurement)) {
+ throw new PathAlreadyExistException(devicePath.getFullPath() + "." +
measurement);
+ }
+ }
+
+ if (upperTemplate != null) {
+ for (String measurement : measurements) {
+ if (upperTemplate.getDirectNode(measurement) != null) {
+ throw new TemplateImcompatibeException(
+ devicePath.concatNode(measurement).getFullPath(),
upperTemplate.getName());
+ }
+ }
+ }
+
+ if (cur.isEntity() && !cur.getAsEntityMNode().isAligned()) {
+ throw new AlignedTimeseriesException(
+ "Timeseries under this entity is not aligned, please use
createTimeseries or change entity.",
+ devicePath.getFullPath());
+ }
+
+ IEntityMNode entityMNode = MNodeUtils.setToEntity(cur);
+ entityMNode.setAligned(true);
+ entityMNode.setAutoAligned(true);
+
+ for (int i = 0; i < measurements.size(); i++) {
+ IMeasurementMNode measurementMNode =
+ MeasurementMNode.getMeasurementMNode(
+ entityMNode,
+ measurements.get(i),
+ new MeasurementSchema(
+ measurements.get(i), dataTypes.get(i), encodings.get(i),
compressors.get(i)),
+ null);
+ entityMNode.addChild(measurements.get(i), measurementMNode);
+ }
+ }
+ }
+
private Pair<IMNode, Template> checkAndAutoCreateInternalPath(PartialPath
devicePath)
throws MetadataException {
String[] nodeNames = devicePath.getNodes();
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
index 9540444..198631c 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
@@ -273,6 +273,8 @@ public class PlanExecutor implements IPlanExecutor {
return createTimeSeries((CreateTimeSeriesPlan) plan);
case CREATE_ALIGNED_TIMESERIES:
return createAlignedTimeSeries((CreateAlignedTimeSeriesPlan) plan);
+ case CREATE_AUTOALIGNED_TIMESERIES:
+ return createAutoAlignedTimeSeries((CreateAutoAlignedTimeSeriesPlan)
plan);
case CREATE_MULTI_TIMESERIES:
return createMultiTimeSeries((CreateMultiTimeSeriesPlan) plan);
case ALTER_TIMESERIES:
@@ -1826,6 +1828,17 @@ public class PlanExecutor implements IPlanExecutor {
return true;
}
+ private boolean createAutoAlignedTimeSeries(
+ CreateAutoAlignedTimeSeriesPlan createAutoAlignedTimeSeriesPlan)
+ throws QueryProcessException {
+ try {
+
IoTDB.metaManager.createAutoAlignedTimeSeries(createAutoAlignedTimeSeriesPlan);
+ } catch (MetadataException e) {
+ throw new QueryProcessException(e);
+ }
+ return true;
+ }
+
@SuppressWarnings("squid:S3776") // high Cognitive Complexity
private boolean createMultiTimeSeries(CreateMultiTimeSeriesPlan multiPlan)
throws BatchProcessException {
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
b/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
index de439e9..4e8a7a8 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/logical/Operator.java
@@ -194,6 +194,7 @@ public abstract class Operator {
APPEND_TEMPLATE,
DROP_TEMPLATE,
- SHOW_QUERY_RESOURCE
+ SHOW_QUERY_RESOURCE,
+ CREATE_AUTOALIGNED_TIMESERIES
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/InsertOperator.java
b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/InsertOperator.java
index 99176c4..dc247d3 100644
---
a/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/InsertOperator.java
+++
b/server/src/main/java/org/apache/iotdb/db/qp/logical/crud/InsertOperator.java
@@ -39,6 +39,7 @@ public class InsertOperator extends Operator {
private List<String[]> valueLists;
private boolean isAligned;
+ private boolean isAutoAligned;
public InsertOperator(int tokenIntType) {
super(tokenIntType);
@@ -85,6 +86,11 @@ public class InsertOperator extends Operator {
isAligned = aligned;
}
+ public void setAutoAligned(boolean autoAligned) {
+ isAligned = autoAligned;
+ isAutoAligned = autoAligned;
+ }
+
@Override
public PhysicalPlan generatePhysicalPlan(PhysicalGenerator generator)
throws QueryProcessException {
@@ -99,6 +105,7 @@ public class InsertOperator extends Operator {
InsertRowPlan insertRowPlan =
new InsertRowPlan(device, times[0], measurementList,
valueLists.get(0));
insertRowPlan.setAligned(isAligned);
+ insertRowPlan.setAutoAligned(isAutoAligned);
return insertRowPlan;
}
InsertRowsPlan insertRowsPlan = new InsertRowsPlan();
@@ -112,6 +119,7 @@ public class InsertOperator extends Operator {
InsertRowPlan insertRowPlan =
new InsertRowPlan(device, times[i], measurementList.clone(),
valueLists.get(i));
insertRowPlan.setAligned(isAligned);
+ insertRowPlan.setAutoAligned(isAutoAligned);
insertRowsPlan.addOneInsertRowPlan(insertRowPlan, i);
}
return insertRowsPlan;
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/CreateAutoAlignedTimeSeriesOperator.java
b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/CreateAutoAlignedTimeSeriesOperator.java
new file mode 100755
index 0000000..1f8072d
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/CreateAutoAlignedTimeSeriesOperator.java
@@ -0,0 +1,131 @@
+/*
+ * 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.qp.logical.sys;
+
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.exception.runtime.SQLParserException;
+import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.qp.logical.Operator;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateAutoAlignedTimeSeriesPlan;
+import org.apache.iotdb.db.qp.strategy.PhysicalGenerator;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public class CreateAutoAlignedTimeSeriesOperator extends Operator {
+
+ private PartialPath prefixPath;
+ private List<String> measurements = new ArrayList<>();
+ private List<TSDataType> dataTypes = new ArrayList<>();
+ private List<TSEncoding> encodings = new ArrayList<>();
+ private List<CompressionType> compressors = new ArrayList<>();
+ private List<String> aliasList = null;
+
+ public CreateAutoAlignedTimeSeriesOperator(int tokenIntType) {
+ super(tokenIntType);
+ operatorType = OperatorType.CREATE_ALIGNED_TIMESERIES;
+ }
+
+ public PartialPath getPrefixPath() {
+ return prefixPath;
+ }
+
+ public void setPrefixPath(PartialPath prefixPath) {
+ this.prefixPath = prefixPath;
+ }
+
+ public List<String> getMeasurements() {
+ return measurements;
+ }
+
+ public void setMeasurements(List<String> measurements) {
+ this.measurements = measurements;
+ }
+
+ public void addMeasurement(String measurement) {
+ this.measurements.add(measurement);
+ }
+
+ public List<TSDataType> getDataTypes() {
+ return dataTypes;
+ }
+
+ public void setDataTypes(List<TSDataType> dataTypes) {
+ this.dataTypes = dataTypes;
+ }
+
+ public void addDataType(TSDataType dataType) {
+ this.dataTypes.add(dataType);
+ }
+
+ public List<TSEncoding> getEncodings() {
+ return encodings;
+ }
+
+ public void setEncodings(List<TSEncoding> encodings) {
+ this.encodings = encodings;
+ }
+
+ public void addEncoding(TSEncoding encoding) {
+ this.encodings.add(encoding);
+ }
+
+ public List<CompressionType> getCompressors() {
+ return compressors;
+ }
+
+ public void setCompressors(List<CompressionType> compressors) {
+ this.compressors = compressors;
+ }
+
+ public void addCompressor(CompressionType compression) {
+ this.compressors.add(compression);
+ }
+
+ public List<String> getAliasList() {
+ return aliasList;
+ }
+
+ public void setAliasList(List<String> aliasList) {
+ this.aliasList = aliasList;
+ }
+
+ public void addAliasList(String alias) {
+ this.aliasList.add(alias);
+ }
+
+ @Override
+ public PhysicalPlan generatePhysicalPlan(PhysicalGenerator generator)
+ throws QueryProcessException {
+ Set<String> measurementSet = new HashSet<>(measurements);
+ if (measurementSet.size() < measurements.size()) {
+ throw new SQLParserException(
+ "the measurement under an aligned device is not allowed to have the
same measurement name");
+ }
+
+ return new CreateAutoAlignedTimeSeriesPlan(
+ prefixPath, measurements, dataTypes, encodings, compressors,
aliasList);
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
index 0267b03..c0df4de 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
@@ -30,48 +30,7 @@ import
org.apache.iotdb.db.qp.physical.crud.InsertRowsOfOneDevicePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowsPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.qp.physical.crud.SelectIntoPlan;
-import org.apache.iotdb.db.qp.physical.sys.ActivateTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.AlterTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.AppendTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
-import org.apache.iotdb.db.qp.physical.sys.AutoCreateDeviceMNodePlan;
-import org.apache.iotdb.db.qp.physical.sys.ChangeAliasPlan;
-import org.apache.iotdb.db.qp.physical.sys.ChangeTagOffsetPlan;
-import org.apache.iotdb.db.qp.physical.sys.ClearCachePlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateContinuousQueryPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateFunctionPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateIndexPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateMultiTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateSnapshotPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.CreateTriggerPlan;
-import org.apache.iotdb.db.qp.physical.sys.DataAuthPlan;
-import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan;
-import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.DropContinuousQueryPlan;
-import org.apache.iotdb.db.qp.physical.sys.DropFunctionPlan;
-import org.apache.iotdb.db.qp.physical.sys.DropIndexPlan;
-import org.apache.iotdb.db.qp.physical.sys.DropTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.DropTriggerPlan;
-import org.apache.iotdb.db.qp.physical.sys.FlushPlan;
-import org.apache.iotdb.db.qp.physical.sys.LoadConfigurationPlan;
-import org.apache.iotdb.db.qp.physical.sys.LogPlan;
-import org.apache.iotdb.db.qp.physical.sys.MNodePlan;
-import org.apache.iotdb.db.qp.physical.sys.MeasurementMNodePlan;
-import org.apache.iotdb.db.qp.physical.sys.MergePlan;
-import org.apache.iotdb.db.qp.physical.sys.PruneTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
-import org.apache.iotdb.db.qp.physical.sys.SetSystemModePlan;
-import org.apache.iotdb.db.qp.physical.sys.SetTTLPlan;
-import org.apache.iotdb.db.qp.physical.sys.SetTemplatePlan;
-import org.apache.iotdb.db.qp.physical.sys.ShowDevicesPlan;
-import org.apache.iotdb.db.qp.physical.sys.ShowTimeSeriesPlan;
-import org.apache.iotdb.db.qp.physical.sys.StartTriggerPlan;
-import org.apache.iotdb.db.qp.physical.sys.StopTriggerPlan;
-import org.apache.iotdb.db.qp.physical.sys.StorageGroupMNodePlan;
-import org.apache.iotdb.db.qp.physical.sys.UnsetTemplatePlan;
+import org.apache.iotdb.db.qp.physical.sys.*;
import org.apache.iotdb.db.qp.utils.EmptyOutputStream;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
@@ -328,6 +287,9 @@ public abstract class PhysicalPlan {
case CREATE_ALIGNED_TIMESERIES:
plan = new CreateAlignedTimeSeriesPlan();
break;
+ case CREATE_AUTOALIGNED_TIMESERIES:
+ plan = new CreateAutoAlignedTimeSeriesPlan();
+ break;
case DELETE_TIMESERIES:
plan = new DeleteTimeSeriesPlan();
break;
@@ -557,7 +519,8 @@ public abstract class PhysicalPlan {
UNSET_TEMPLATE,
APPEND_TEMPLATE,
PRUNE_TEMPLATE,
- DROP_TEMPLATE
+ DROP_TEMPLATE,
+ CREATE_AUTOALIGNED_TIMESERIES
}
public long getIndex() {
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
index df71f87..809aa01 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java
@@ -40,6 +40,8 @@ public abstract class InsertPlan extends PhysicalPlan {
protected PartialPath devicePath;
protected boolean isAligned;
+ protected boolean isAutoAligned;
+
protected String[] measurements;
// get from client
protected TSDataType[] dataTypes;
@@ -115,10 +117,18 @@ public abstract class InsertPlan extends PhysicalPlan {
return isAligned;
}
+ public boolean isAutoAligned() {
+ return isAutoAligned;
+ }
+
public void setAligned(boolean aligned) {
isAligned = aligned;
}
+ public void setAutoAligned(boolean autoAligned) {
+ isAutoAligned = autoAligned;
+ }
+
public abstract long getMinTime();
/**
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertRowPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertRowPlan.java
index ee263cc..848428f 100644
---
a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertRowPlan.java
+++
b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertRowPlan.java
@@ -114,6 +114,26 @@ public class InsertRowPlan extends InsertPlan {
this.isAligned = isAligned;
}
+ public InsertRowPlan(
+ PartialPath prefixPath,
+ long insertTime,
+ String[] measurementList,
+ ByteBuffer values,
+ boolean isAligned,
+ boolean isAutoAligned)
+ throws QueryProcessException {
+ super(Operator.OperatorType.INSERT);
+ this.time = insertTime;
+ this.devicePath = prefixPath;
+ this.measurements = measurementList;
+ this.dataTypes = new TSDataType[measurementList.length];
+ this.values = new Object[measurementList.length];
+ this.fillValues(values);
+ isNeedInferType = false;
+ this.isAligned = isAligned;
+ this.isAutoAligned = isAutoAligned;
+ }
+
@TestOnly
public InsertRowPlan(
PartialPath prefixPath,
@@ -363,6 +383,7 @@ public class InsertRowPlan extends InsertPlan {
stream.writeLong(index);
stream.write((byte) (isAligned ? 1 : 0));
+ stream.write((byte) (isAutoAligned ? 1 : 0));
}
private void putValues(DataOutputStream outputStream) throws
QueryProcessException, IOException {
@@ -513,6 +534,7 @@ public class InsertRowPlan extends InsertPlan {
buffer.putLong(index);
buffer.put((byte) (isAligned ? 1 : 0));
+ buffer.put((byte) (isAutoAligned ? 1 : 0));
}
@Override
@@ -542,6 +564,7 @@ public class InsertRowPlan extends InsertPlan {
isNeedInferType = buffer.get() == 1;
this.index = buffer.getLong();
isAligned = buffer.get() == 1;
+ isAutoAligned = buffer.get() == 1;
}
@Override
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateAutoAlignedTimeSeriesPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateAutoAlignedTimeSeriesPlan.java
new file mode 100755
index 0000000..2174900
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateAutoAlignedTimeSeriesPlan.java
@@ -0,0 +1,267 @@
+/*
+ * 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.qp.physical.sys;
+
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.qp.logical.Operator;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+public class CreateAutoAlignedTimeSeriesPlan extends PhysicalPlan {
+
+ private static final Logger logger =
+ LoggerFactory.getLogger(CreateAutoAlignedTimeSeriesPlan.class);
+
+ private PartialPath prefixPath;
+ private List<String> measurements;
+ private List<TSDataType> dataTypes;
+ private List<TSEncoding> encodings;
+ private List<CompressionType> compressors;
+ private List<String> aliasList;
+
+ public CreateAutoAlignedTimeSeriesPlan() {
+ super(Operator.OperatorType.CREATE_AUTOALIGNED_TIMESERIES);
+ canBeSplit = false;
+ }
+
+ public CreateAutoAlignedTimeSeriesPlan(
+ PartialPath prefixPath,
+ List<String> measurements,
+ List<TSDataType> dataTypes,
+ List<TSEncoding> encodings,
+ List<CompressionType> compressors,
+ List<String> aliasList) {
+ super(Operator.OperatorType.CREATE_AUTOALIGNED_TIMESERIES);
+ this.prefixPath = prefixPath;
+ this.measurements = measurements;
+ this.dataTypes = dataTypes;
+ this.encodings = encodings;
+ this.compressors = compressors;
+ this.aliasList = aliasList;
+ this.canBeSplit = false;
+ }
+
+ public PartialPath getPrefixPath() {
+ return prefixPath;
+ }
+
+ public void setPrefixPath(PartialPath prefixPath) {
+ this.prefixPath = prefixPath;
+ }
+
+ public List<String> getMeasurements() {
+ return measurements;
+ }
+
+ public void setMeasurements(List<String> measurements) {
+ this.measurements = measurements;
+ }
+
+ public List<TSDataType> getDataTypes() {
+ return dataTypes;
+ }
+
+ public void setDataTypes(List<TSDataType> dataTypes) {
+ this.dataTypes = dataTypes;
+ }
+
+ public List<TSEncoding> getEncodings() {
+ return encodings;
+ }
+
+ public void setEncodings(List<TSEncoding> encodings) {
+ this.encodings = encodings;
+ }
+
+ public List<CompressionType> getCompressors() {
+ return compressors;
+ }
+
+ public void setCompressors(List<CompressionType> compressor) {
+ this.compressors = compressors;
+ }
+
+ public List<String> getAliasList() {
+ return aliasList;
+ }
+
+ public void setAliasList(List<String> aliasList) {
+ this.aliasList = aliasList;
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "devicePath: %s, measurements: %s, dataTypes: %s, encodings: %s,
compressions: %s",
+ prefixPath, measurements, dataTypes, encodings, compressors);
+ }
+
+ @Override
+ public List<PartialPath> getPaths() {
+ List<PartialPath> paths = new ArrayList<>();
+ for (String measurement : measurements) {
+ try {
+ paths.add(new PartialPath(prefixPath.getFullPath(), measurement));
+ } catch (IllegalPathException e) {
+ logger.error("Failed to get paths of CreateAutoAlignedTimeSeriesPlan.
", e);
+ }
+ }
+ return paths;
+ }
+
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ stream.writeByte((byte)
PhysicalPlanType.CREATE_AUTOALIGNED_TIMESERIES.ordinal());
+ byte[] bytes = prefixPath.getFullPath().getBytes();
+ stream.writeInt(bytes.length);
+ stream.write(bytes);
+
+ ReadWriteIOUtils.write(measurements.size(), stream);
+ for (String measurement : measurements) {
+ ReadWriteIOUtils.write(measurement, stream);
+ }
+ for (TSDataType dataType : dataTypes) {
+ stream.write(dataType.ordinal());
+ }
+ for (TSEncoding encoding : encodings) {
+ stream.write(encoding.ordinal());
+ }
+ for (CompressionType compressor : compressors) {
+ stream.write(compressor.ordinal());
+ }
+
+ // alias
+ if (aliasList != null) {
+ stream.write(1);
+ for (String alias : aliasList) {
+ ReadWriteIOUtils.write(alias, stream);
+ }
+ } else {
+ stream.write(0);
+ }
+ stream.writeLong(index);
+ }
+
+ @Override
+ public void serializeImpl(ByteBuffer buffer) {
+ buffer.put((byte)
PhysicalPlanType.CREATE_AUTOALIGNED_TIMESERIES.ordinal());
+ byte[] bytes = prefixPath.getFullPath().getBytes();
+ buffer.putInt(bytes.length);
+ buffer.put(bytes);
+
+ ReadWriteIOUtils.write(measurements.size(), buffer);
+ for (String measurement : measurements) {
+ ReadWriteIOUtils.write(measurement, buffer);
+ }
+ for (TSDataType dataType : dataTypes) {
+ buffer.put((byte) dataType.ordinal());
+ }
+ for (TSEncoding encoding : encodings) {
+ buffer.put((byte) encoding.ordinal());
+ }
+ for (CompressionType compressor : compressors) {
+ buffer.put((byte) compressor.ordinal());
+ }
+
+ // alias
+ if (aliasList != null) {
+ buffer.put((byte) 1);
+ for (String alias : aliasList) {
+ ReadWriteIOUtils.write(alias, buffer);
+ }
+ } else {
+ buffer.put((byte) 0);
+ }
+
+ buffer.putLong(index);
+ }
+
+ @Override
+ public void deserialize(ByteBuffer buffer) throws IllegalPathException {
+ int length = buffer.getInt();
+ byte[] bytes = new byte[length];
+ buffer.get(bytes);
+
+ prefixPath = new PartialPath(new String(bytes));
+ int size = ReadWriteIOUtils.readInt(buffer);
+ measurements = new ArrayList<>();
+ for (int i = 0; i < size; i++) {
+ measurements.add(ReadWriteIOUtils.readString(buffer));
+ }
+ dataTypes = new ArrayList<>();
+ for (int i = 0; i < size; i++) {
+ dataTypes.add(TSDataType.values()[buffer.get()]);
+ }
+ encodings = new ArrayList<>();
+ for (int i = 0; i < size; i++) {
+ encodings.add(TSEncoding.values()[buffer.get()]);
+ }
+ compressors = new ArrayList<>();
+ for (int i = 0; i < size; i++) {
+ compressors.add(CompressionType.values()[buffer.get()]);
+ }
+
+ // alias
+ if (buffer.get() == 1) {
+ aliasList = new ArrayList<>();
+ for (int i = 0; i < size; i++) {
+ aliasList.add(ReadWriteIOUtils.readString(buffer));
+ }
+ }
+
+ this.index = buffer.getLong();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ CreateAutoAlignedTimeSeriesPlan that = (CreateAutoAlignedTimeSeriesPlan) o;
+
+ return Objects.equals(prefixPath, that.prefixPath)
+ && Objects.equals(measurements, that.measurements)
+ && Objects.equals(dataTypes, that.dataTypes)
+ && Objects.equals(encodings, that.encodings)
+ && Objects.equals(compressors, that.compressors);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(prefixPath, measurements, dataTypes, encodings,
compressors);
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
b/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
index 97a0158..faeb0b2 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
@@ -223,6 +223,26 @@ public class IoTDBSqlVisitor extends
IoTDBSqlParserBaseVisitor<Operator> {
}
}
+ @Override
+ public Operator visitCreateAutoAlignedTimeseries(
+ IoTDBSqlParser.CreateAutoAlignedTimeseriesContext ctx) {
+ CreateAutoAlignedTimeSeriesOperator createAutoAlignedTimeSeriesOperator =
+ new
CreateAutoAlignedTimeSeriesOperator(SQLConstant.TOK_METADATA_CREATE);
+
createAutoAlignedTimeSeriesOperator.setPrefixPath(parseFullPath(ctx.fullPath()));
+ parseAutoAlignedMeasurements(
+ ctx.autoAlignedMeasurements(), createAutoAlignedTimeSeriesOperator);
+ return createAutoAlignedTimeSeriesOperator;
+ }
+
+ public void parseAutoAlignedMeasurements(
+ IoTDBSqlParser.AutoAlignedMeasurementsContext ctx,
+ CreateAutoAlignedTimeSeriesOperator createAutoAlignedTimeSeriesOperator)
{
+ for (int i = 0; i < ctx.nodeNameWithoutWildcard().size(); i++) {
+
createAutoAlignedTimeSeriesOperator.addMeasurement(ctx.nodeNameWithoutWildcard(i).getText());
+ parseAttributeClauses(ctx.attributeClauses(i),
createAutoAlignedTimeSeriesOperator);
+ }
+ }
+
public void parseAttributeClauses(
IoTDBSqlParser.AttributeClausesContext ctx,
CreateTimeSeriesOperator createTimeSeriesOperator) {
@@ -394,6 +414,44 @@ public class IoTDBSqlVisitor extends
IoTDBSqlParserBaseVisitor<Operator> {
}
}
+ public void parseAttributeClauses(
+ IoTDBSqlParser.AttributeClausesContext ctx,
+ CreateAutoAlignedTimeSeriesOperator createAutoAlignedTimeSeriesOperator)
{
+ if (ctx.alias() != null) {
+ throw new SQLParserException("create aligned timeseries: alias is not
supported yet.");
+ }
+
+ String dataTypeString = ctx.dataType.getText().toUpperCase();
+ TSDataType dataType = TSDataType.valueOf(dataTypeString);
+ createAutoAlignedTimeSeriesOperator.addDataType(dataType);
+
+ TSEncoding encoding =
IoTDBDescriptor.getInstance().getDefaultEncodingByType(dataType);
+ if (Objects.nonNull(ctx.encoding)) {
+ String encodingString = ctx.encoding.getText().toUpperCase();
+ encoding = TSEncoding.valueOf(encodingString);
+ }
+ createAutoAlignedTimeSeriesOperator.addEncoding(encoding);
+
+ CompressionType compressor =
TSFileDescriptor.getInstance().getConfig().getCompressor();
+ if (ctx.compressor != null) {
+ String compressorString = ctx.compressor.getText().toUpperCase();
+ compressor = CompressionType.valueOf(compressorString);
+ }
+ createAutoAlignedTimeSeriesOperator.addCompressor(compressor);
+
+ if (ctx.propertyClause(0) != null) {
+ throw new SQLParserException("create aligned timeseries: property is not
supported yet.");
+ }
+
+ if (ctx.tagClause() != null) {
+ throw new SQLParserException("create aligned timeseries: tag is not
supported yet.");
+ }
+
+ if (ctx.attributeClause() != null) {
+ throw new SQLParserException("create aligned timeseries: attribute is
not supported yet.");
+ }
+ }
+
// Create Timeseries Of Schema Template
@Override
@@ -1599,6 +1657,7 @@ public class IoTDBSqlVisitor extends
IoTDBSqlParserBaseVisitor<Operator> {
boolean isTimeDefault = parseInsertColumnSpec(ctx.insertColumnsSpec(),
insertOp);
parseInsertValuesSpec(ctx.insertValuesSpec(), insertOp, isTimeDefault);
insertOp.setAligned(ctx.ALIGNED() != null);
+ insertOp.setAutoAligned(ctx.AUTOALIGNED() != null);
return insertOp;
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
index c91348f..5f1d408 100644
---
a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
+++
b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
@@ -148,6 +148,10 @@ public class AlignedTVList extends TVList {
return getAlignedValueByValueIndex(valueIndex, null, floatPrecision,
encodingList);
}
+ public List<List<BitMap>> getBitMaps() {
+ return this.bitMaps;
+ }
+
public TsPrimitiveType getAlignedValue(
List<Integer> timeDuplicatedIndexList,
Integer floatPrecision,
diff --git
a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/AutoAlignedChunkGroupWriterImpl.java
b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/AutoAlignedChunkGroupWriterImpl.java
new file mode 100644
index 0000000..16a51ec
--- /dev/null
+++
b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/AutoAlignedChunkGroupWriterImpl.java
@@ -0,0 +1,301 @@
+/*
+ * 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.tsfile.write.chunk;
+
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.encoding.encoder.Encoder;
+import org.apache.iotdb.tsfile.encoding.encoder.TSEncodingBuilder;
+import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
+import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.record.datapoint.DataPoint;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.*;
+
+public class AutoAlignedChunkGroupWriterImpl implements IChunkGroupWriter {
+ private static final Logger LOG =
LoggerFactory.getLogger(AutoAlignedChunkGroupWriterImpl.class);
+
+ private final String deviceId;
+
+ // measurementID -> ValueChunkWriter
+ private Map<String, ValueChunkWriter> valueChunkWriterMap = new
LinkedHashMap<>();
+
+ private TimeChunkWriter timeChunkWriter;
+
+ private Set<String> writenMeasurementSet = new HashSet<>();
+
+ private long lastTime = -1;
+
+ public AutoAlignedChunkGroupWriterImpl(String deviceId) {
+ this.deviceId = deviceId;
+ String timeMeasurementId = "";
+ CompressionType compressionType =
TSFileDescriptor.getInstance().getConfig().getCompressor();
+ TSEncoding tsEncoding =
+
TSEncoding.valueOf(TSFileDescriptor.getInstance().getConfig().getTimeEncoder());
+ TSDataType timeType =
TSFileDescriptor.getInstance().getConfig().getTimeSeriesDataType();
+ Encoder encoder =
TSEncodingBuilder.getEncodingBuilder(tsEncoding).getEncoder(timeType);
+ timeChunkWriter = new TimeChunkWriter(timeMeasurementId, compressionType,
tsEncoding, encoder);
+ }
+
+ @Override
+ public void tryToAddSeriesWriter(MeasurementSchema measurementSchema) {
+ if
(!valueChunkWriterMap.containsKey(measurementSchema.getMeasurementId())) {
+ ValueChunkWriter valueChunkWriter =
+ new ValueChunkWriter(
+ measurementSchema.getMeasurementId(),
+ measurementSchema.getCompressor(),
+ measurementSchema.getType(),
+ measurementSchema.getEncodingType(),
+ measurementSchema.getValueEncoder());
+ valueChunkWriterMap.put(measurementSchema.getMeasurementId(),
valueChunkWriter);
+ tryToAddEmptyPageAndData(valueChunkWriter);
+ }
+ }
+
+ @Override
+ public void tryToAddSeriesWriter(List<MeasurementSchema> measurementSchemas)
{
+ for (MeasurementSchema schema : measurementSchemas) {
+ if (!valueChunkWriterMap.containsKey(schema.getMeasurementId())) {
+ ValueChunkWriter valueChunkWriter =
+ new ValueChunkWriter(
+ schema.getMeasurementId(),
+ schema.getCompressor(),
+ schema.getType(),
+ schema.getEncodingType(),
+ schema.getValueEncoder());
+ valueChunkWriterMap.put(schema.getMeasurementId(), valueChunkWriter);
+ tryToAddEmptyPageAndData(valueChunkWriter);
+ }
+ }
+ }
+
+ @Override
+ public int write(long time, List<DataPoint> data) throws
WriteProcessException, IOException {
+ checkIsHistoryData("", time);
+
+ for (DataPoint point : data) {
+ writenMeasurementSet.add(point.getMeasurementId());
+ boolean isNull = point.getValue() == null;
+ ValueChunkWriter valueChunkWriter =
valueChunkWriterMap.get(point.getMeasurementId());
+ switch (point.getType()) {
+ case BOOLEAN:
+ valueChunkWriter.write(time, (boolean) point.getValue(), isNull);
+ break;
+ case INT32:
+ valueChunkWriter.write(time, (int) point.getValue(), isNull);
+ break;
+ case INT64:
+ valueChunkWriter.write(time, (long) point.getValue(), isNull);
+ break;
+ case FLOAT:
+ valueChunkWriter.write(time, (float) point.getValue(), isNull);
+ break;
+ case DOUBLE:
+ valueChunkWriter.write(time, (double) point.getValue(), isNull);
+ break;
+ case TEXT:
+ valueChunkWriter.write(time, (Binary) point.getValue(), isNull);
+ break;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Data type %s is not supported.",
point.getType()));
+ }
+ }
+ writeEmptyDataInOneRow(time);
+ timeChunkWriter.write(time);
+ lastTime = time;
+ if (checkPageSizeAndMayOpenANewPage()) {
+ writePageToPageBuffer();
+ }
+ return 1;
+ }
+
+ @Override
+ public int write(Tablet tablet) throws WriteProcessException, IOException {
+ int pointCount = 0;
+ List<MeasurementSchema> measurementSchemas = tablet.getSchemas();
+ for (int row = 0; row < tablet.rowSize; row++) {
+ long time = tablet.timestamps[row];
+ checkIsHistoryData("", time);
+ for (int columnIndex = 0; columnIndex < measurementSchemas.size();
columnIndex++) {
+
writenMeasurementSet.add(measurementSchemas.get(columnIndex).getMeasurementId());
+ boolean isNull = false;
+ // check isNull by bitMap in tablet
+ if (tablet.bitMaps != null
+ && tablet.bitMaps[columnIndex] != null
+ && !tablet.bitMaps[columnIndex].isMarked(row)) {
+ isNull = true;
+ }
+ ValueChunkWriter valueChunkWriter =
+
valueChunkWriterMap.get(measurementSchemas.get(columnIndex).getMeasurementId());
+ switch (measurementSchemas.get(columnIndex).getType()) {
+ case BOOLEAN:
+ valueChunkWriter.write(time, ((boolean[])
tablet.values[columnIndex])[row], isNull);
+ break;
+ case INT32:
+ valueChunkWriter.write(time, ((int[])
tablet.values[columnIndex])[row], isNull);
+ break;
+ case INT64:
+ valueChunkWriter.write(time, ((long[])
tablet.values[columnIndex])[row], isNull);
+ break;
+ case FLOAT:
+ valueChunkWriter.write(time, ((float[])
tablet.values[columnIndex])[row], isNull);
+ break;
+ case DOUBLE:
+ valueChunkWriter.write(time, ((double[])
tablet.values[columnIndex])[row], isNull);
+ break;
+ case TEXT:
+ valueChunkWriter.write(time, ((Binary[])
tablet.values[columnIndex])[row], isNull);
+ break;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format(
+ "Data type %s is not supported.",
+ measurementSchemas.get(columnIndex).getType()));
+ }
+ }
+ writeEmptyDataInOneRow(time);
+ timeChunkWriter.write(time);
+ lastTime = time;
+ if (checkPageSizeAndMayOpenANewPage()) {
+ writePageToPageBuffer();
+ }
+ pointCount++;
+ }
+ return pointCount;
+ }
+
+ @Override
+ public long flushToFileWriter(TsFileIOWriter tsfileWriter) throws
IOException {
+ LOG.debug("start flush device id:{}", deviceId);
+ // make sure all the pages have been compressed into buffers, so that we
can get correct
+ // groupWriter.getCurrentChunkGroupSize().
+ sealAllChunks();
+ long currentChunkGroupSize = getCurrentChunkGroupSize();
+ timeChunkWriter.writeToFileWriter(tsfileWriter);
+ for (ValueChunkWriter valueChunkWriter : valueChunkWriterMap.values()) {
+ valueChunkWriter.writeToFileWriter(tsfileWriter);
+ }
+ return currentChunkGroupSize;
+ }
+
+ @Override
+ public long updateMaxGroupMemSize() {
+ long bufferSize = timeChunkWriter.estimateMaxSeriesMemSize();
+ for (ValueChunkWriter valueChunkWriter : valueChunkWriterMap.values()) {
+ bufferSize += valueChunkWriter.estimateMaxSeriesMemSize();
+ }
+ return bufferSize;
+ }
+
+ @Override
+ public long getCurrentChunkGroupSize() {
+ long size = timeChunkWriter.getCurrentChunkSize();
+ for (ValueChunkWriter valueChunkWriter : valueChunkWriterMap.values()) {
+ size += valueChunkWriter.getCurrentChunkSize();
+ }
+ return size;
+ }
+
+ public void tryToAddEmptyPageAndData(ValueChunkWriter valueChunkWriter) {
+ // add empty page
+ for (int i = 0; i < timeChunkWriter.getNumOfPages(); i++) {
+ valueChunkWriter.writeEmptyPageToPageBuffer();
+ }
+
+ // add empty data of currentPage
+ for (long i = 0; i <
timeChunkWriter.getPageWriter().getStatistics().getCount(); i++) {
+ valueChunkWriter.write(0, 0, true);
+ }
+ }
+
+ private void writeEmptyDataInOneRow(long time) {
+ for (Map.Entry<String, ValueChunkWriter> entry :
valueChunkWriterMap.entrySet()) {
+ if (!writenMeasurementSet.contains(entry.getKey())) {
+ entry.getValue().write(time, 0, true);
+ }
+ }
+ writenMeasurementSet.clear();
+ }
+
+ /**
+ * check occupied memory size, if it exceeds the PageSize threshold,
construct a page and put it
+ * to pageBuffer
+ */
+ private boolean checkPageSizeAndMayOpenANewPage() {
+ if (timeChunkWriter.checkPageSizeAndMayOpenANewPage()) {
+ return true;
+ }
+ for (ValueChunkWriter writer : valueChunkWriterMap.values()) {
+ if (writer.checkPageSizeAndMayOpenANewPage()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private void writePageToPageBuffer() {
+ timeChunkWriter.writePageToPageBuffer();
+ for (ValueChunkWriter valueChunkWriter : valueChunkWriterMap.values()) {
+ valueChunkWriter.writePageToPageBuffer();
+ }
+ }
+
+ private void sealAllChunks() {
+ timeChunkWriter.sealCurrentPage();
+ for (ValueChunkWriter valueChunkWriter : valueChunkWriterMap.values()) {
+ valueChunkWriter.sealCurrentPage();
+ }
+ }
+
+ private void checkIsHistoryData(String measurementId, long time) throws
WriteProcessException {
+ if (time <= lastTime) {
+ throw new WriteProcessException(
+ "Not allowed to write out-of-order data in timeseries "
+ + deviceId
+ + TsFileConstant.PATH_SEPARATOR
+ + measurementId
+ + ", time should later than "
+ + lastTime);
+ }
+ }
+
+ public List<String> getMeasurements() {
+ return new ArrayList<>(valueChunkWriterMap.keySet());
+ }
+
+ public Long getLastTime() {
+ return this.lastTime;
+ }
+
+ public void setLastTime(Long lastTime) {
+ this.lastTime = lastTime;
+ }
+}
diff --git
a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/AutoAlignedChunkWriterImpl.java
b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/AutoAlignedChunkWriterImpl.java
new file mode 100644
index 0000000..7bb2e4d
--- /dev/null
+++
b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/AutoAlignedChunkWriterImpl.java
@@ -0,0 +1,239 @@
+/*
+ * 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.tsfile.write.chunk;
+
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+import org.apache.iotdb.tsfile.encoding.encoder.Encoder;
+import org.apache.iotdb.tsfile.encoding.encoder.TSEncodingBuilder;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.utils.TsPrimitiveType;
+import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
+import org.apache.iotdb.tsfile.write.schema.VectorMeasurementSchema;
+import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+public class AutoAlignedChunkWriterImpl implements IChunkWriter {
+
+ private final TimeChunkWriter timeChunkWriter;
+ private final List<ValueChunkWriter> valueChunkWriterList;
+ private int valueIndex;
+
+ /** @param schema schema of this measurement */
+ public AutoAlignedChunkWriterImpl(VectorMeasurementSchema schema) {
+ timeChunkWriter =
+ new TimeChunkWriter(
+ schema.getMeasurementId(),
+ schema.getCompressor(),
+ schema.getTimeTSEncoding(),
+ schema.getTimeEncoder());
+
+ List<String> valueMeasurementIdList = schema.getSubMeasurementsList();
+ List<TSDataType> valueTSDataTypeList =
schema.getSubMeasurementsTSDataTypeList();
+ List<TSEncoding> valueTSEncodingList =
schema.getSubMeasurementsTSEncodingList();
+ List<Encoder> valueEncoderList = schema.getSubMeasurementsEncoderList();
+
+ valueChunkWriterList = new ArrayList<>(valueMeasurementIdList.size());
+ for (int i = 0; i < valueMeasurementIdList.size(); i++) {
+ valueChunkWriterList.add(
+ new ValueChunkWriter(
+ valueMeasurementIdList.get(i),
+ schema.getCompressor(),
+ valueTSDataTypeList.get(i),
+ valueTSEncodingList.get(i),
+ valueEncoderList.get(i)));
+ }
+
+ this.valueIndex = 0;
+ }
+
+ public AutoAlignedChunkWriterImpl(List<IMeasurementSchema> schemaList) {
+ TSEncoding timeEncoding =
+
TSEncoding.valueOf(TSFileDescriptor.getInstance().getConfig().getTimeEncoder());
+ TSDataType timeType =
TSFileDescriptor.getInstance().getConfig().getTimeSeriesDataType();
+ timeChunkWriter =
+ new TimeChunkWriter(
+ "",
+ schemaList.get(0).getCompressor(),
+ timeEncoding,
+
TSEncodingBuilder.getEncodingBuilder(timeEncoding).getEncoder(timeType));
+
+ valueChunkWriterList = new ArrayList<>(schemaList.size());
+ for (int i = 0; i < schemaList.size(); i++) {
+ valueChunkWriterList.add(
+ new ValueChunkWriter(
+ schemaList.get(i).getMeasurementId(),
+ schemaList.get(i).getCompressor(),
+ schemaList.get(i).getType(),
+ schemaList.get(i).getEncodingType(),
+ schemaList.get(i).getValueEncoder()));
+ }
+
+ this.valueIndex = 0;
+ }
+
+ public void write(long time, int value, boolean isNull) {
+ valueChunkWriterList.get(valueIndex++).write(time, value, isNull);
+ }
+
+ public void write(long time, long value, boolean isNull) {
+ valueChunkWriterList.get(valueIndex++).write(time, value, isNull);
+ }
+
+ public void write(long time, boolean value, boolean isNull) {
+ valueChunkWriterList.get(valueIndex++).write(time, value, isNull);
+ }
+
+ public void write(long time, float value, boolean isNull) {
+ valueChunkWriterList.get(valueIndex++).write(time, value, isNull);
+ }
+
+ public void write(long time, double value, boolean isNull) {
+ valueChunkWriterList.get(valueIndex++).write(time, value, isNull);
+ }
+
+ public void write(long time, Binary value, boolean isNull) {
+ valueChunkWriterList.get(valueIndex++).write(time, value, isNull);
+ }
+
+ public void write(long time, TsPrimitiveType[] points) {
+ valueIndex = 0;
+ for (TsPrimitiveType point : points) {
+ ValueChunkWriter writer = valueChunkWriterList.get(valueIndex++);
+ switch (writer.getDataType()) {
+ case INT64:
+ writer.write(time, point != null ? point.getLong() : Long.MAX_VALUE,
point == null);
+ break;
+ case INT32:
+ writer.write(time, point != null ? point.getInt() :
Integer.MAX_VALUE, point == null);
+ break;
+ case FLOAT:
+ writer.write(time, point != null ? point.getFloat() :
Float.MAX_VALUE, point == null);
+ break;
+ case DOUBLE:
+ writer.write(time, point != null ? point.getDouble() :
Double.MAX_VALUE, point == null);
+ break;
+ case BOOLEAN:
+ writer.write(time, point != null ? point.getBoolean() : false, point
== null);
+ break;
+ case TEXT:
+ writer.write(
+ time,
+ point != null ? point.getBinary() : new
Binary("".getBytes(StandardCharsets.UTF_8)),
+ point == null);
+ break;
+ }
+ }
+ write(time);
+ }
+
+ public void write(long time) {
+ valueIndex = 0;
+ timeChunkWriter.write(time);
+ if (checkPageSizeAndMayOpenANewPage()) {
+ writePageToPageBuffer();
+ }
+ }
+
+ /**
+ * check occupied memory size, if it exceeds the PageSize threshold,
construct a page and put it
+ * to pageBuffer
+ */
+ private boolean checkPageSizeAndMayOpenANewPage() {
+ if (timeChunkWriter.checkPageSizeAndMayOpenANewPage()) {
+ return true;
+ }
+ for (ValueChunkWriter writer : valueChunkWriterList) {
+ if (writer.checkPageSizeAndMayOpenANewPage()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private void writePageToPageBuffer() {
+ timeChunkWriter.writePageToPageBuffer();
+ for (ValueChunkWriter valueChunkWriter : valueChunkWriterList) {
+ valueChunkWriter.writePageToPageBuffer();
+ }
+ }
+
+ @Override
+ public void writeToFileWriter(TsFileIOWriter tsfileWriter) throws
IOException {
+ timeChunkWriter.writeToFileWriter(tsfileWriter);
+ for (ValueChunkWriter valueChunkWriter : valueChunkWriterList) {
+ valueChunkWriter.writeToFileWriter(tsfileWriter);
+ }
+ }
+
+ @Override
+ public long estimateMaxSeriesMemSize() {
+ long estimateMaxSeriesMemSize = timeChunkWriter.estimateMaxSeriesMemSize();
+ for (ValueChunkWriter valueChunkWriter : valueChunkWriterList) {
+ estimateMaxSeriesMemSize += valueChunkWriter.estimateMaxSeriesMemSize();
+ }
+ return estimateMaxSeriesMemSize;
+ }
+
+ public long getSerializedChunkSize() {
+ long currentChunkSize = timeChunkWriter.getCurrentChunkSize();
+ for (ValueChunkWriter valueChunkWriter : valueChunkWriterList) {
+ currentChunkSize += valueChunkWriter.getCurrentChunkSize();
+ }
+ return currentChunkSize;
+ }
+
+ @Override
+ public void sealCurrentPage() {
+ timeChunkWriter.sealCurrentPage();
+ for (ValueChunkWriter valueChunkWriter : valueChunkWriterList) {
+ valueChunkWriter.sealCurrentPage();
+ }
+ }
+
+ @Override
+ public void clearPageWriter() {
+ timeChunkWriter.clearPageWriter();
+ for (ValueChunkWriter valueChunkWriter : valueChunkWriterList) {
+ valueChunkWriter.clearPageWriter();
+ }
+ }
+
+ /** Used for compaction to control the target chunk size. */
+ public boolean checkIsChunkSizeOverThreshold(long threshold) {
+ if (timeChunkWriter.estimateMaxSeriesMemSize() > threshold) {
+ return true;
+ }
+ for (ValueChunkWriter valueChunkWriter : valueChunkWriterList) {
+ if (valueChunkWriter.estimateMaxSeriesMemSize() > threshold) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ public TSDataType getCurrentValueChunkType() {
+ return valueChunkWriterList.get(valueIndex).getDataType();
+ }
+}
