You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ha...@apache.org on 2022/03/27 15:45:08 UTC
[iotdb] 01/11: InsertTabletNode
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch InsertTabletNode
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 284c3a1b3ce1cdecc1e5d1d65d9a2fe369ae67d5
Author: HTHou <hh...@outlook.com>
AuthorDate: Thu Mar 24 14:16:14 2022 +0800
InsertTabletNode
---
.../org/apache/iotdb/db/engine/StorageEngine.java | 16 ++
.../iotdb/db/engine/memtable/AbstractMemTable.java | 108 ++++++++++++
.../apache/iotdb/db/engine/memtable/IMemTable.java | 7 +
.../db/engine/storagegroup/TsFileProcessor.java | 149 +++++++++++++++++
.../storagegroup/VirtualStorageGroupProcessor.java | 183 +++++++++++++++++++++
.../sql/planner/plan/node/write/InsertNode.java | 32 +++-
.../planner/plan/node/write/InsertTabletNode.java | 16 ++
.../java/org/apache/iotdb/db/utils/MemUtils.java | 56 ++++++-
8 files changed, 565 insertions(+), 2 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
index 2c0075d..674c5cf 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
@@ -53,6 +53,7 @@ import org.apache.iotdb.db.metadata.idtable.entry.DeviceIDFactory;
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.mpp.sql.planner.plan.node.write.InsertTabletNode;
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.InsertRowsOfOneDevicePlan;
@@ -648,6 +649,21 @@ public class StorageEngine implements IService {
virtualStorageGroupProcessor.insertTablet(insertTabletPlan);
}
+ /** insert an InsertTabletNode to a storage group */
+ public void insertTabletV2(VirtualStorageGroupProcessor virtualStorageGroupProcessor, InsertTabletNode insertTabletNode)
+ throws StorageEngineException, BatchProcessException, MetadataException {
+ if (enableMemControl) {
+ try {
+ blockInsertionIfReject(null);
+ } catch (WriteProcessRejectException e) {
+ TSStatus[] results = new TSStatus[insertTabletNode.getRowCount()];
+ Arrays.fill(results, RpcUtils.getStatus(TSStatusCode.WRITE_PROCESS_REJECT));
+ throw new BatchProcessException(results);
+ }
+ }
+ virtualStorageGroupProcessor.insertTablet(insertTabletNode);
+ }
+
/** flush command Sync asyncCloseOneProcessor all file node processors. */
public void syncCloseAllProcessor() {
logger.info("Start closing all storage group processor");
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 06706b1..75787c1 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
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.idtable.entry.DeviceIDFactory;
import org.apache.iotdb.db.metadata.idtable.entry.IDeviceID;
import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.service.metrics.Metric;
@@ -279,6 +280,58 @@ public abstract class AbstractMemTable implements IMemTable {
}
@Override
+ public void insertTablet(InsertTabletNode insertTabletNode, int start, int end)
+ throws WriteProcessException {
+ // TODO: PlanIndex
+ //updatePlanIndexes(insertTabletPlan.getIndex());
+ updatePlanIndexes(0);
+ try {
+ write(insertTabletNode, start, end);
+ memSize += MemUtils.getTabletSize(insertTabletNode, start, end, disableMemControl);
+ int pointsInserted = insertTabletNode.getDataTypes().length * (end - start);
+ totalPointsNum += pointsInserted;
+ if (MetricConfigDescriptor.getInstance().getMetricConfig().getEnableMetric()) {
+ MetricsService.getInstance()
+ .getMetricManager()
+ .count(
+ pointsInserted,
+ Metric.QUANTITY.toString(),
+ MetricLevel.IMPORTANT,
+ Tag.NAME.toString(),
+ METRIC_POINT_IN);
+ }
+ } catch (RuntimeException e) {
+ throw new WriteProcessException(e);
+ }
+ }
+
+ @Override
+ public void insertAlignedTablet(InsertTabletNode insertTabletNode, int start, int end)
+ throws WriteProcessException {
+ // TODO: PlanIndex
+ //updatePlanIndexes(insertTabletPlan.getIndex());
+ updatePlanIndexes(0);
+ try {
+ writeAlignedTablet(insertTabletNode, start, end);
+ memSize += MemUtils.getAlignedTabletSize(insertTabletNode, start, end, disableMemControl);
+ int pointsInserted = insertTabletNode.getDataTypes().length * (end - start);
+ totalPointsNum += pointsInserted;
+ if (MetricConfigDescriptor.getInstance().getMetricConfig().getEnableMetric()) {
+ MetricsService.getInstance()
+ .getMetricManager()
+ .count(
+ pointsInserted,
+ Metric.QUANTITY.toString(),
+ MetricLevel.IMPORTANT,
+ Tag.NAME.toString(),
+ METRIC_POINT_IN);
+ }
+ } catch (RuntimeException e) {
+ throw new WriteProcessException(e);
+ }
+ }
+
+ @Override
public void write(
IDeviceID deviceId,
List<IMeasurementSchema> schemaList,
@@ -328,6 +381,32 @@ public abstract class AbstractMemTable implements IMemTable {
end);
}
+ public void write(InsertTabletNode insertTabletNode, int start, int end) {
+ // if this insert plan isn't from storage engine, we should set a temp device id for it
+ if (insertTabletNode.getDeviceID() == null) {
+ insertTabletNode.setDeviceID(
+ DeviceIDFactory.getInstance().getDeviceID(insertTabletNode.getDevicePath()));
+ }
+
+ List<IMeasurementSchema> schemaList = new ArrayList<>();
+ for (int i = 0; i < insertTabletNode.getMeasurements().length; i++) {
+ if (insertTabletNode.getColumns()[i] == null) {
+ continue;
+ }
+ IMeasurementSchema schema = insertTabletNode.getMeasurementSchemas()[i];
+ schemaList.add(schema);
+ }
+ IWritableMemChunkGroup memChunkGroup =
+ createMemChunkGroupIfNotExistAndGet(insertTabletNode.getDeviceID(), schemaList);
+ memChunkGroup.writeValues(
+ insertTabletNode.getTimes(),
+ insertTabletNode.getColumns(),
+ insertTabletNode.getBitMaps(),
+ schemaList,
+ start,
+ end);
+ }
+
@Override
public void writeAlignedTablet(InsertTabletPlan insertTabletPlan, int start, int end) {
// if this insert plan isn't from storage engine, we should set a temp device id for it
@@ -358,6 +437,35 @@ public abstract class AbstractMemTable implements IMemTable {
end);
}
+ public void writeAlignedTablet(InsertTabletNode insertTabletNode, int start, int end) {
+ // if this insert plan isn't from storage engine, we should set a temp device id for it
+ if (insertTabletNode.getDeviceID() == null) {
+ insertTabletNode.setDeviceID(
+ DeviceIDFactory.getInstance().getDeviceID(insertTabletNode.getDevicePath()));
+ }
+
+ List<IMeasurementSchema> schemaList = new ArrayList<>();
+ for (int i = 0; i < insertTabletNode.getMeasurements().length; i++) {
+ if (insertTabletNode.getColumns()[i] == null) {
+ continue;
+ }
+ IMeasurementSchema schema = insertTabletNode.getMeasurementSchemas()[i];
+ schemaList.add(schema);
+ }
+ if (schemaList.isEmpty()) {
+ return;
+ }
+ IWritableMemChunkGroup memChunkGroup =
+ createAlignedMemChunkGroupIfNotExistAndGet(insertTabletNode.getDeviceID(), schemaList);
+ memChunkGroup.writeValues(
+ insertTabletNode.getTimes(),
+ insertTabletNode.getColumns(),
+ insertTabletNode.getBitMaps(),
+ schemaList,
+ start,
+ end);
+ }
+
@Override
public boolean checkIfChunkDoesNotExist(IDeviceID deviceId, String measurement) {
IWritableMemChunkGroup memChunkGroup = memTableMap.get(deviceId);
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..ce3747f 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
@@ -25,6 +25,7 @@ import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.idtable.entry.IDeviceID;
import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.tsfile.utils.Pair;
@@ -115,6 +116,12 @@ public interface IMemTable {
void insertAlignedTablet(InsertTabletPlan insertTabletPlan, int start, int end)
throws WriteProcessException;
+ void insertTablet(InsertTabletNode insertTabletNode, int start, int end)
+ throws WriteProcessException;
+
+ void insertAlignedTablet(InsertTabletNode insertTabletNode, int start, int end)
+ throws WriteProcessException;
+
ReadOnlyMemChunk query(
PartialPath fullPath, long ttlLowerBound, List<Pair<Modification, IMemTable>> modsToMemtable)
throws IOException, QueryProcessException, MetadataException;
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 df6e20b..e09b90e 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
@@ -46,6 +46,7 @@ import org.apache.iotdb.db.metadata.idtable.entry.DeviceIDFactory;
import org.apache.iotdb.db.metadata.idtable.entry.IDeviceID;
import org.apache.iotdb.db.metadata.path.AlignedPath;
import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.query.context.QueryContext;
@@ -340,6 +341,92 @@ public class TsFileProcessor {
tsFileResource.updatePlanIndexes(insertTabletPlan.getIndex());
}
+ /**
+ * insert batch data of insertTabletPlan into the workingMemtable. 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, 3, null, 5} will be {1, 3, 5, null, 5}
+ *
+ * @param insertTabletNode insert a tablet of a device
+ * @param start start index of rows to be inserted in insertTabletPlan
+ * @param end end index of rows to be inserted in insertTabletPlan
+ * @param results result array
+ */
+ public void insertTablet(
+ InsertTabletNode insertTabletNode, int start, int end, TSStatus[] results)
+ throws WriteProcessException {
+
+ if (workMemTable == null) {
+ if (enableMemControl) {
+ workMemTable = new PrimitiveMemTable(enableMemControl);
+ MemTableManager.getInstance().addMemtableNumber();
+ } else {
+ workMemTable = MemTableManager.getInstance().getAvailableMemTable(storageGroupName);
+ }
+ }
+
+ long[] memIncrements = null;
+ try {
+ if (enableMemControl) {
+ if (insertTabletNode.isAligned()) {
+ memIncrements = checkAlignedMemCostAndAddToTsp(insertTabletNode, start, end);
+ } else {
+ memIncrements = checkMemCostAndAddToTspInfo(insertTabletNode, start, end);
+ }
+ }
+ } catch (WriteProcessException e) {
+ for (int i = start; i < end; i++) {
+ results[i] = RpcUtils.getStatus(TSStatusCode.WRITE_PROCESS_REJECT, e.getMessage());
+ }
+ throw new WriteProcessException(e);
+ }
+
+ // TODO(WAL)
+// try {
+// if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
+// insertTabletPlan.setStart(start);
+// insertTabletPlan.setEnd(end);
+// getLogNode().write(insertTabletPlan);
+// }
+// } catch (Exception e) {
+// for (int i = start; i < end; i++) {
+// results[i] = RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage());
+// }
+// if (enableMemControl && memIncrements != null) {
+// rollbackMemoryInfo(memIncrements);
+// }
+// throw new WriteProcessException(e);
+// }
+
+ try {
+ if (insertTabletNode.isAligned()) {
+ workMemTable.insertAlignedTablet(insertTabletNode, start, end);
+ } else {
+ workMemTable.insertTablet(insertTabletNode, start, end);
+ }
+ } catch (WriteProcessException e) {
+ for (int i = start; i < end; i++) {
+ results[i] = RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage());
+ }
+ throw new WriteProcessException(e);
+ }
+
+ for (int i = start; i < end; i++) {
+ results[i] = RpcUtils.SUCCESS_STATUS;
+ }
+ tsFileResource.updateStartTime(
+ insertTabletNode.getDeviceID().toStringID(), insertTabletNode.getTimes()[start]);
+
+ // for sequence tsfile, we update the endTime only when the file is prepared to be closed.
+ // for unsequence tsfile, we have to update the endTime for each insertion.
+ if (!sequence) {
+ tsFileResource.updateEndTime(
+ insertTabletNode.getDeviceID().toStringID(), insertTabletNode.getTimes()[end - 1]);
+ }
+ // TODO: PlanIndex
+ tsFileResource.updatePlanIndexes(0);
+// tsFileResource.updatePlanIndexes(insertTabletPlan.getIndex());
+ }
+
@SuppressWarnings("squid:S3776") // high Cognitive Complexity
private long[] checkMemCostAndAddToTspInfo(InsertRowPlan insertRowPlan)
throws WriteProcessException {
@@ -471,6 +558,38 @@ public class TsFileProcessor {
return memIncrements;
}
+ private long[] checkMemCostAndAddToTspInfo(InsertTabletNode insertTabletNode, int start, int end)
+ throws WriteProcessException {
+ if (start >= end) {
+ return new long[] {0, 0, 0};
+ }
+ long[] memIncrements = new long[3]; // memTable, text, chunk metadata
+
+ // get device id
+ IDeviceID deviceID = null;
+ try {
+ deviceID = getDeviceID(insertTabletNode.getDevicePath().getFullPath());
+ } catch (IllegalPathException e) {
+ throw new WriteProcessException(e);
+ }
+
+ for (int i = 0; i < insertTabletNode.getDataTypes().length; i++) {
+ // skip failed Measurements
+ TSDataType dataType = insertTabletNode.getDataTypes()[i];
+ String measurement = insertTabletNode.getMeasurements()[i];
+ Object column = insertTabletNode.getColumns()[i];
+ if (dataType == null || column == null || measurement == null) {
+ continue;
+ }
+ updateMemCost(dataType, measurement, deviceID, start, end, memIncrements, column);
+ }
+ long memTableIncrement = memIncrements[0];
+ long textDataIncrement = memIncrements[1];
+ long chunkMetadataIncrement = memIncrements[2];
+ updateMemoryInfo(memTableIncrement, chunkMetadataIncrement, textDataIncrement);
+ return memIncrements;
+ }
+
private long[] checkAlignedMemCostAndAddToTsp(
InsertTabletPlan insertTabletPlan, int start, int end) throws WriteProcessException {
if (start >= end) {
@@ -501,6 +620,36 @@ public class TsFileProcessor {
return memIncrements;
}
+ private long[] checkAlignedMemCostAndAddToTsp(
+ InsertTabletNode insertTabletNode, int start, int end) throws WriteProcessException {
+ if (start >= end) {
+ return new long[] {0, 0, 0};
+ }
+ long[] memIncrements = new long[3]; // memTable, text, chunk metadata
+
+ // get device id
+ IDeviceID deviceID = null;
+ try {
+ deviceID = getDeviceID(insertTabletNode.getDevicePath().getFullPath());
+ } catch (IllegalPathException e) {
+ throw new WriteProcessException(e);
+ }
+
+ updateAlignedMemCost(
+ insertTabletNode.getDataTypes(),
+ deviceID,
+ insertTabletNode.getMeasurements(),
+ start,
+ end,
+ memIncrements,
+ insertTabletNode.getColumns());
+ long memTableIncrement = memIncrements[0];
+ long textDataIncrement = memIncrements[1];
+ long chunkMetadataIncrement = memIncrements[2];
+ updateMemoryInfo(memTableIncrement, chunkMetadataIncrement, textDataIncrement);
+ return memIncrements;
+ }
+
private void updateMemCost(
TSDataType dataType,
String measurement,
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/VirtualStorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/VirtualStorageGroupProcessor.java
index 3cd3ed7..a17a2e2 100755
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/VirtualStorageGroupProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/VirtualStorageGroupProcessor.java
@@ -61,6 +61,7 @@ import org.apache.iotdb.db.metadata.idtable.IDTable;
import org.apache.iotdb.db.metadata.idtable.IDTableManager;
import org.apache.iotdb.db.metadata.mnode.IMeasurementMNode;
import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowsOfOneDevicePlan;
@@ -1103,6 +1104,126 @@ public class VirtualStorageGroupProcessor {
}
}
+ /**
+ * Insert a tablet (rows belonging to the same devices) into this storage group.
+ *
+ * @throws BatchProcessException if some of the rows failed to be inserted
+ */
+ @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
+ public void insertTablet(InsertTabletNode insertTabletNode)
+ throws BatchProcessException, TriggerExecutionException {
+
+ writeLock("insertTablet");
+ try {
+ TSStatus[] results = new TSStatus[insertTabletNode.getRowCount()];
+ Arrays.fill(results, RpcUtils.SUCCESS_STATUS);
+ boolean noFailure = true;
+
+ /*
+ * assume that batch has been sorted by client
+ */
+ int loc = 0;
+ while (loc < insertTabletNode.getRowCount()) {
+ long currTime = insertTabletNode.getTimes()[loc];
+ // skip points that do not satisfy TTL
+ if (!isAlive(currTime)) {
+ results[loc] =
+ RpcUtils.getStatus(
+ TSStatusCode.OUT_OF_TTL_ERROR,
+ "time " + currTime + " in current line is out of TTL: " + dataTTL);
+ loc++;
+ noFailure = false;
+ } else {
+ break;
+ }
+ }
+ // loc pointing at first legal position
+ if (loc == insertTabletNode.getRowCount()) {
+ throw new BatchProcessException(results);
+ }
+
+// TODO(Trigger)// fire trigger before insertion
+// final int firePosition = loc;
+// TriggerEngine.fire(TriggerEvent.BEFORE_INSERT, insertTabletPlan, firePosition);
+
+ // before is first start point
+ int before = loc;
+ // before time partition
+ long beforeTimePartition =
+ StorageEngine.getTimePartition(insertTabletNode.getTimes()[before]);
+ // init map
+ long lastFlushTime =
+ lastFlushTimeManager.ensureFlushedTimePartitionAndInit(
+ beforeTimePartition, insertTabletNode.getDevicePath().getFullPath(), Long.MIN_VALUE);
+ // if is sequence
+ boolean isSequence = false;
+ while (loc < insertTabletNode.getRowCount()) {
+ long time = insertTabletNode.getTimes()[loc];
+ long curTimePartition = StorageEngine.getTimePartition(time);
+ // start next partition
+ if (curTimePartition != beforeTimePartition) {
+ // insert last time partition
+ if (isSequence
+ || !IoTDBDescriptor.getInstance().getConfig().isEnableDiscardOutOfOrderData()) {
+ noFailure =
+ insertTabletToTsFileProcessor(
+ insertTabletNode, before, loc, isSequence, results, beforeTimePartition)
+ && noFailure;
+ }
+ // re initialize
+ before = loc;
+ beforeTimePartition = curTimePartition;
+ lastFlushTime =
+ lastFlushTimeManager.ensureFlushedTimePartitionAndInit(
+ beforeTimePartition,
+ insertTabletNode.getDevicePath().getFullPath(),
+ Long.MIN_VALUE);
+
+ isSequence = false;
+ }
+ // still in this partition
+ else {
+ // judge if we should insert sequence
+ if (!isSequence && time > lastFlushTime) {
+ // insert into unsequence and then start sequence
+ if (!IoTDBDescriptor.getInstance().getConfig().isEnableDiscardOutOfOrderData()) {
+ noFailure =
+ insertTabletToTsFileProcessor(
+ insertTabletNode, before, loc, false, results, beforeTimePartition)
+ && noFailure;
+ }
+ before = loc;
+ isSequence = true;
+ }
+ loc++;
+ }
+ }
+
+ // do not forget last part
+ if (before < loc
+ && (isSequence
+ || !IoTDBDescriptor.getInstance().getConfig().isEnableDiscardOutOfOrderData())) {
+ noFailure =
+ insertTabletToTsFileProcessor(
+ insertTabletNode, before, loc, isSequence, results, beforeTimePartition)
+ && noFailure;
+ }
+ long globalLatestFlushedTime =
+ lastFlushTimeManager.getGlobalFlushedTime(insertTabletNode.getDevicePath().getFullPath());
+ // TODO:LAST CACHE
+// tryToUpdateBatchInsertLastCache(insertTabletNode, globalLatestFlushedTime);
+
+ if (!noFailure) {
+ throw new BatchProcessException(results);
+ }
+
+// TODO: trigger // fire trigger after insertion
+// TriggerEngine.fire(TriggerEvent.AFTER_INSERT, insertTabletPlan, firePosition);
+ } finally {
+ writeUnlock();
+ }
+ }
+
/** @return whether the given time falls in ttl */
private boolean isAlive(long time) {
return dataTTL == Long.MAX_VALUE || (System.currentTimeMillis() - time) <= dataTTL;
@@ -1170,6 +1291,68 @@ public class VirtualStorageGroupProcessor {
return true;
}
+ /**
+ * insert batch to tsfile processor thread-safety that the caller need to guarantee 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, 3, null, 5} will be {1, 3, 5, null, 5}
+ *
+ * @param insertTabletNode insert a tablet of a device
+ * @param sequence whether is sequence
+ * @param start start index of rows to be inserted in insertTabletPlan
+ * @param end end index of rows to be inserted in insertTabletPlan
+ * @param results result array
+ * @param timePartitionId time partition id
+ * @return false if any failure occurs when inserting the tablet, true otherwise
+ */
+ private boolean insertTabletToTsFileProcessor(
+ InsertTabletNode insertTabletNode,
+ int start,
+ int end,
+ boolean sequence,
+ TSStatus[] results,
+ long timePartitionId) {
+ // return when start >= end
+ if (start >= end) {
+ return true;
+ }
+
+ TsFileProcessor tsFileProcessor = getOrCreateTsFileProcessor(timePartitionId, sequence);
+ if (tsFileProcessor == null) {
+ for (int i = start; i < end; i++) {
+ results[i] =
+ RpcUtils.getStatus(
+ TSStatusCode.INTERNAL_SERVER_ERROR,
+ "can not create TsFileProcessor, timePartitionId: " + timePartitionId);
+ }
+ return false;
+ }
+
+ try {
+ tsFileProcessor.insertTablet(insertTabletNode, start, end, results);
+ } catch (WriteProcessRejectException e) {
+ logger.warn("insert to TsFileProcessor rejected, {}", e.getMessage());
+ return false;
+ } catch (WriteProcessException e) {
+ logger.error("insert to TsFileProcessor error ", e);
+ return false;
+ }
+
+ lastFlushTimeManager.ensureLastTimePartition(timePartitionId);
+ // try to update the latest time of the device of this tsRecord
+ if (sequence) {
+ lastFlushTimeManager.updateLastTime(
+ timePartitionId,
+ insertTabletNode.getDevicePath().getFullPath(),
+ insertTabletNode.getTimes()[end - 1]);
+ }
+
+ // check memtable size and may async try to flush the work memtable
+ if (tsFileProcessor.shouldFlush()) {
+ fileFlushPolicy.apply(this, tsFileProcessor, sequence);
+ }
+ return true;
+ }
+
private void tryToUpdateBatchInsertLastCache(InsertTabletPlan plan, Long latestFlushedTime) {
if (!IoTDBDescriptor.getInstance().getConfig().isLastCacheEnabled()) {
return;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/write/InsertNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/write/InsertNode.java
index 9edaf5c..3108166 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/write/InsertNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/write/InsertNode.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.db.mpp.sql.analyze.Analysis;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeId;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import java.util.List;
@@ -37,7 +38,8 @@ public abstract class InsertNode extends PlanNode {
protected PartialPath devicePath;
protected boolean isAligned;
- protected MeasurementSchema[] measurements;
+ protected MeasurementSchema[] measurementSchemas;
+ protected String[] measurements;
protected TSDataType[] dataTypes;
// TODO(INSERT) need to change it to a function handle to update last time value
// protected IMeasurementMNode[] measurementMNodes;
@@ -59,4 +61,32 @@ public abstract class InsertNode extends PlanNode {
public boolean needSplit() {
return true;
}
+
+ public PartialPath getDevicePath() {
+ return devicePath;
+ }
+
+ public boolean isAligned() {
+ return isAligned;
+ }
+
+ public IDeviceID getDeviceID() {
+ return deviceID;
+ }
+
+ public TSDataType[] getDataTypes() {
+ return dataTypes;
+ }
+
+ public String[] getMeasurements() {
+ return measurements;
+ }
+
+ public void setDeviceID(IDeviceID deviceID) {
+ this.deviceID = deviceID;
+ }
+
+ public MeasurementSchema[] getMeasurementSchemas() {
+ return measurementSchemas;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/write/InsertTabletNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/write/InsertTabletNode.java
index 5ce7bbb..5bcab88 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/write/InsertTabletNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/write/InsertTabletNode.java
@@ -73,4 +73,20 @@ public class InsertTabletNode extends InsertNode {
public List<InsertNode> splitByPartition(Analysis analysis) {
return null;
}
+
+ public int getRowCount() {
+ return times.length;
+ }
+
+ public long[] getTimes() {
+ return times;
+ }
+
+ public Object[] getColumns() {
+ return columns;
+ }
+
+ public BitMap[] getBitMaps() {
+ return bitMaps;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/MemUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/MemUtils.java
index 854fb03..11df91c 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/MemUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/MemUtils.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.utils;
import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.write.InsertTabletNode;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
@@ -110,7 +111,34 @@ public class MemUtils {
* before inserting.
*/
public static long getTabletSize(
- InsertTabletPlan insertTabletPlan, int start, int end, boolean addingTextDataSize) {
+ InsertTabletNode insertTabletNode, int start, int end, boolean addingTextDataSize) {
+ if (start >= end) {
+ return 0L;
+ }
+ long memSize = 0;
+ for (int i = 0; i < insertTabletNode.getMeasurements().length; i++) {
+ if (insertTabletNode.getMeasurements()[i] == null) {
+ continue;
+ }
+ // time column memSize
+ memSize += (end - start) * 8L;
+ if (insertTabletNode.getDataTypes()[i] == TSDataType.TEXT && addingTextDataSize) {
+ for (int j = start; j < end; j++) {
+ memSize += getBinarySize(((Binary[]) insertTabletNode.getColumns()[i])[j]);
+ }
+ } else {
+ memSize += (end - start) * insertTabletNode.getDataTypes()[i].getDataTypeSize();
+ }
+ }
+ return memSize;
+ }
+
+ /**
+ * If mem control enabled, do not add text data size here, the size will be added to memtable
+ * before inserting.
+ */
+ public static long getTabletSize(
+ InsertTabletPlan insertTabletPlan, int start, int end, boolean addingTextDataSize) {
if (start >= end) {
return 0L;
}
@@ -158,6 +186,32 @@ public class MemUtils {
return memSize;
}
+ public static long getAlignedTabletSize(
+ InsertTabletNode insertTabletNode, int start, int end, boolean addingTextDataSize) {
+ if (start >= end) {
+ return 0L;
+ }
+ long memSize = 0;
+ for (int i = 0; i < insertTabletNode.getMeasurements().length; i++) {
+ if (insertTabletNode.getMeasurements()[i] == null) {
+ continue;
+ }
+ TSDataType valueType;
+ // value columns memSize
+ valueType = insertTabletNode.getDataTypes()[i];
+ if (valueType == TSDataType.TEXT && addingTextDataSize) {
+ for (int j = start; j < end; j++) {
+ memSize += getBinarySize(((Binary[]) insertTabletNode.getColumns()[i])[j]);
+ }
+ } else {
+ memSize += (long) (end - start) * valueType.getDataTypeSize();
+ }
+ }
+ // time and index column memSize for vector
+ memSize += (end - start) * (8L + 4L);
+ return memSize;
+ }
+
/** Calculate how much memory will be used if the given record is written to sequence file. */
public static long getTsRecordMem(TSRecord record) {
long memUsed = 8; // time