This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 9ee5aedc1d [IOTDB-2801] New storage engine framework (#5357)
9ee5aedc1d is described below
commit 9ee5aedc1d49540f7434a3ceef59705224670beb
Author: Haonan <[email protected]>
AuthorDate: Sat Apr 9 17:57:49 2022 +0800
[IOTDB-2801] New storage engine framework (#5357)
---
.../apache/iotdb/db/engine/StorageEngineV2.java | 638 +++++++++++++++++++++
.../iotdb/db/engine/memtable/AbstractMemTable.java | 198 +++++++
.../apache/iotdb/db/engine/memtable/IMemTable.java | 12 +
.../db/engine/storagegroup/TsFileProcessor.java | 209 +++++++
.../storagegroup/VirtualStorageGroupProcessor.java | 234 ++++++++
.../sql/planner/plan/node/write/InsertNode.java | 25 +-
.../planner/plan/node/write/InsertTabletNode.java | 6 +-
.../java/org/apache/iotdb/db/utils/MemUtils.java | 54 ++
.../db/writelog/node/ExclusiveWriteLogNode.java | 46 ++
.../iotdb/db/writelog/node/WriteLogNode.java | 9 +
10 files changed, 1421 insertions(+), 10 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java
b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java
new file mode 100644
index 0000000000..538ae1ef67
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java
@@ -0,0 +1,638 @@
+/*
+ * 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;
+
+import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.commons.concurrent.ThreadName;
+import org.apache.iotdb.commons.consensus.ConsensusGroupId;
+import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.commons.exception.ShutdownException;
+import org.apache.iotdb.commons.service.IService;
+import org.apache.iotdb.commons.service.ServiceType;
+import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.conf.ServerConfigConsistent;
+import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
+import org.apache.iotdb.db.engine.flush.CloseFileListener;
+import org.apache.iotdb.db.engine.flush.FlushListener;
+import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy;
+import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy.DirectFlushPolicy;
+import org.apache.iotdb.db.engine.storagegroup.TsFileProcessor;
+import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
+import org.apache.iotdb.db.exception.BatchProcessException;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.StorageGroupProcessorException;
+import org.apache.iotdb.db.exception.TsFileProcessorException;
+import org.apache.iotdb.db.exception.WriteProcessException;
+import org.apache.iotdb.db.exception.WriteProcessRejectException;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.exception.runtime.StorageEngineFailureException;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.write.InsertRowNode;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.write.InsertTabletNode;
+import org.apache.iotdb.db.rescon.SystemInfo;
+import org.apache.iotdb.db.utils.ThreadUtils;
+import org.apache.iotdb.db.utils.UpgradeUtils;
+import org.apache.iotdb.rpc.RpcUtils;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.service.rpc.thrift.TSStatus;
+import org.apache.iotdb.tsfile.utils.FilePathUtils;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.apache.commons.io.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.ConcurrentModificationException;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class StorageEngineV2 implements IService {
+ private static final Logger logger =
LoggerFactory.getLogger(StorageEngineV2.class);
+
+ private static final IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
+ private static final long TTL_CHECK_INTERVAL = 60 * 1000L;
+
+ /**
+ * Time range for dividing storage group, the time unit is the same with
IoTDB's
+ * TimestampPrecision
+ */
+ @ServerConfigConsistent private static long timePartitionInterval = -1;
+ /** whether enable data partition if disabled, all data belongs to partition
0 */
+ @ServerConfigConsistent private static boolean enablePartition =
config.isEnablePartition();
+
+ private final boolean enableMemControl = config.isEnableMemControl();
+
+ /**
+ * a folder (system/storage_groups/ by default) that persist system info.
Each Storage Processor
+ * will have a subfolder under the systemDir.
+ */
+ private final String systemDir =
+ FilePathUtils.regularizePath(config.getSystemDir()) + "storage_groups";
+
+ /** DataRegionId -> DataRegion */
+ private final ConcurrentHashMap<ConsensusGroupId,
VirtualStorageGroupProcessor> dataRegionMap =
+ new ConcurrentHashMap<>();
+
+ private AtomicBoolean isAllSgReady = new AtomicBoolean(false);
+
+ private ScheduledExecutorService ttlCheckThread;
+ private ScheduledExecutorService seqMemtableTimedFlushCheckThread;
+ private ScheduledExecutorService unseqMemtableTimedFlushCheckThread;
+ private ScheduledExecutorService tsFileTimedCloseCheckThread;
+
+ private TsFileFlushPolicy fileFlushPolicy = new DirectFlushPolicy();
+ private ExecutorService recoveryThreadPool;
+ // add customized listeners here for flush and close events
+ private List<CloseFileListener> customCloseFileListeners = new ArrayList<>();
+ private List<FlushListener> customFlushListeners = new ArrayList<>();
+
+ private StorageEngineV2() {}
+
+ public static StorageEngineV2 getInstance() {
+ return InstanceHolder.INSTANCE;
+ }
+
+ private static void initTimePartition() {
+ timePartitionInterval =
+ convertMilliWithPrecision(
+ IoTDBDescriptor.getInstance().getConfig().getPartitionInterval() *
1000L);
+ }
+
+ public static long convertMilliWithPrecision(long milliTime) {
+ long result = milliTime;
+ String timePrecision =
IoTDBDescriptor.getInstance().getConfig().getTimestampPrecision();
+ switch (timePrecision) {
+ case "ns":
+ result = milliTime * 1000_000L;
+ break;
+ case "us":
+ result = milliTime * 1000L;
+ break;
+ default:
+ break;
+ }
+ return result;
+ }
+
+ public static long getTimePartitionInterval() {
+ if (timePartitionInterval == -1) {
+ initTimePartition();
+ }
+ return timePartitionInterval;
+ }
+
+ @TestOnly
+ public static void setTimePartitionInterval(long timePartitionInterval) {
+ StorageEngineV2.timePartitionInterval = timePartitionInterval;
+ }
+
+ public static long getTimePartition(long time) {
+ return enablePartition ? time / timePartitionInterval : 0;
+ }
+
+ public static boolean isEnablePartition() {
+ return enablePartition;
+ }
+
+ @TestOnly
+ public static void setEnablePartition(boolean enablePartition) {
+ StorageEngineV2.enablePartition = enablePartition;
+ }
+
+ /** block insertion if the insertion is rejected by memory control */
+ public static void blockInsertionIfReject(TsFileProcessor tsFileProcessor)
+ throws WriteProcessRejectException {
+ long startTime = System.currentTimeMillis();
+ while (SystemInfo.getInstance().isRejected()) {
+ if (tsFileProcessor != null && tsFileProcessor.shouldFlush()) {
+ break;
+ }
+ try {
+ TimeUnit.MILLISECONDS.sleep(config.getCheckPeriodWhenInsertBlocked());
+ if (System.currentTimeMillis() - startTime >
config.getMaxWaitingTimeWhenInsertBlocked()) {
+ throw new WriteProcessRejectException(
+ "System rejected over " + (System.currentTimeMillis() -
startTime) + "ms");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ public boolean isAllSgReady() {
+ return isAllSgReady.get();
+ }
+
+ public void setAllSgReady(boolean allSgReady) {
+ isAllSgReady.set(allSgReady);
+ }
+
+ public void recover() {
+ setAllSgReady(false);
+ recoveryThreadPool =
+ IoTDBThreadPoolFactory.newFixedThreadPool(
+ Runtime.getRuntime().availableProcessors(),
"Recovery-Thread-Pool");
+ try {
+ getLocalDataRegion();
+ } catch (Exception e) {
+ throw new StorageEngineFailureException("StorageEngine failed to
recover.", e);
+ }
+ List<Future<Void>> futures = new LinkedList<>();
+ asyncRecover(recoveryThreadPool, futures);
+
+ // operations after all virtual storage groups are recovered
+ Thread recoverEndTrigger =
+ new Thread(
+ () -> {
+ for (Future<Void> future : futures) {
+ try {
+ future.get();
+ } catch (ExecutionException e) {
+ throw new StorageEngineFailureException("StorageEngine
failed to recover.", e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new StorageEngineFailureException("StorageEngine
failed to recover.", e);
+ }
+ }
+ recoveryThreadPool.shutdown();
+ setAllSgReady(true);
+ });
+ recoverEndTrigger.start();
+ }
+
+ private void getLocalDataRegion() throws MetadataException,
StorageGroupProcessorException {
+ File system = SystemFileFactory.INSTANCE.getFile(systemDir);
+ File[] sgDirs = system.listFiles();
+ for (File sgDir : sgDirs) {
+ if (!sgDir.isDirectory()) {
+ continue;
+ }
+ String sg = sgDir.getName();
+ // TODO: need to get TTL Info from config node
+ long ttl = Integer.MAX_VALUE;
+ for (File dataRegionDir : sgDir.listFiles()) {
+ if (!dataRegionDir.isDirectory()) {
+ continue;
+ }
+ ConsensusGroupId dataRegionId = new
DataRegionId(Integer.parseInt(dataRegionDir.getName()));
+ VirtualStorageGroupProcessor dataRegion =
+ buildNewStorageGroupProcessor(sg, dataRegionDir.getName(), ttl);
+ dataRegionMap.putIfAbsent(dataRegionId, dataRegion);
+ }
+ }
+ }
+
+ private void asyncRecover(ExecutorService pool, List<Future<Void>> futures) {
+ for (VirtualStorageGroupProcessor processor : dataRegionMap.values()) {
+ Callable<Void> recoverVsgTask =
+ () -> {
+ processor.setReady(true);
+ return null;
+ };
+ futures.add(pool.submit(recoverVsgTask));
+ }
+ }
+
+ @Override
+ public void start() {
+ // build time Interval to divide time partition
+ if (!enablePartition) {
+ timePartitionInterval = Long.MAX_VALUE;
+ } else {
+ initTimePartition();
+ }
+
+ // create systemDir
+ try {
+ FileUtils.forceMkdir(SystemFileFactory.INSTANCE.getFile(systemDir));
+ } catch (IOException e) {
+ throw new StorageEngineFailureException(e);
+ }
+
+ // recover upgrade process
+ UpgradeUtils.recoverUpgrade();
+
+ recover();
+
+ ttlCheckThread =
IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor("TTL-Check");
+ ttlCheckThread.scheduleAtFixedRate(
+ this::checkTTL, TTL_CHECK_INTERVAL, TTL_CHECK_INTERVAL,
TimeUnit.MILLISECONDS);
+ logger.info("start ttl check thread successfully.");
+
+ startTimedService();
+ }
+
+ private void checkTTL() {
+ try {
+ for (VirtualStorageGroupProcessor dataRegion : dataRegionMap.values()) {
+ if (dataRegion != null) {
+ dataRegion.checkFilesTTL();
+ }
+ }
+ } catch (ConcurrentModificationException e) {
+ // ignore
+ } catch (Exception e) {
+ logger.error("An error occurred when checking TTL", e);
+ }
+ }
+
+ private void startTimedService() {
+ // timed flush sequence memtable
+ if (config.isEnableTimedFlushSeqMemtable()) {
+ seqMemtableTimedFlushCheckThread =
+ IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
+ ThreadName.TIMED_FlUSH_SEQ_MEMTABLE.getName());
+ seqMemtableTimedFlushCheckThread.scheduleAtFixedRate(
+ this::timedFlushSeqMemTable,
+ config.getSeqMemtableFlushCheckInterval(),
+ config.getSeqMemtableFlushCheckInterval(),
+ TimeUnit.MILLISECONDS);
+ logger.info("start sequence memtable timed flush check thread
successfully.");
+ }
+ // timed flush unsequence memtable
+ if (config.isEnableTimedFlushUnseqMemtable()) {
+ unseqMemtableTimedFlushCheckThread =
+ IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
+ ThreadName.TIMED_FlUSH_UNSEQ_MEMTABLE.getName());
+ unseqMemtableTimedFlushCheckThread.scheduleAtFixedRate(
+ this::timedFlushUnseqMemTable,
+ config.getUnseqMemtableFlushCheckInterval(),
+ config.getUnseqMemtableFlushCheckInterval(),
+ TimeUnit.MILLISECONDS);
+ logger.info("start unsequence memtable timed flush check thread
successfully.");
+ }
+ // timed close tsfile
+ if (config.isEnableTimedCloseTsFile()) {
+ tsFileTimedCloseCheckThread =
+ IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
+ ThreadName.TIMED_CLOSE_TSFILE.getName());
+ tsFileTimedCloseCheckThread.scheduleAtFixedRate(
+ this::timedCloseTsFileProcessor,
+ config.getCloseTsFileCheckInterval(),
+ config.getCloseTsFileCheckInterval(),
+ TimeUnit.MILLISECONDS);
+ logger.info("start tsfile timed close check thread successfully.");
+ }
+ }
+
+ private void timedFlushSeqMemTable() {
+ try {
+ for (VirtualStorageGroupProcessor dataRegion : dataRegionMap.values()) {
+ if (dataRegion != null) {
+ dataRegion.timedFlushSeqMemTable();
+ }
+ }
+ } catch (Exception e) {
+ logger.error("An error occurred when timed flushing sequence memtables",
e);
+ }
+ }
+
+ private void timedFlushUnseqMemTable() {
+ try {
+ for (VirtualStorageGroupProcessor dataRegion : dataRegionMap.values()) {
+ if (dataRegion != null) {
+ dataRegion.timedFlushUnseqMemTable();
+ }
+ }
+ } catch (Exception e) {
+ logger.error("An error occurred when timed flushing unsequence
memtables", e);
+ }
+ }
+
+ private void timedCloseTsFileProcessor() {
+ try {
+ for (VirtualStorageGroupProcessor dataRegion : dataRegionMap.values()) {
+ if (dataRegion != null) {
+ dataRegion.timedCloseTsFileProcessor();
+ }
+ }
+ } catch (Exception e) {
+ logger.error("An error occurred when timed closing tsfiles interval", e);
+ }
+ }
+
+ @Override
+ public void stop() {
+ for (VirtualStorageGroupProcessor vsg : dataRegionMap.values()) {
+ if (vsg != null) {
+ ThreadUtils.stopThreadPool(
+ vsg.getTimedCompactionScheduleTask(),
ThreadName.COMPACTION_SCHEDULE);
+ ThreadUtils.stopThreadPool(vsg.getWALTrimScheduleTask(),
ThreadName.WAL_TRIM);
+ }
+ }
+ syncCloseAllProcessor();
+ ThreadUtils.stopThreadPool(ttlCheckThread, ThreadName.TTL_CHECK_SERVICE);
+ ThreadUtils.stopThreadPool(
+ seqMemtableTimedFlushCheckThread, ThreadName.TIMED_FlUSH_SEQ_MEMTABLE);
+ ThreadUtils.stopThreadPool(
+ unseqMemtableTimedFlushCheckThread,
ThreadName.TIMED_FlUSH_UNSEQ_MEMTABLE);
+ ThreadUtils.stopThreadPool(tsFileTimedCloseCheckThread,
ThreadName.TIMED_CLOSE_TSFILE);
+ recoveryThreadPool.shutdownNow();
+ // TODO(Removed from new wal)
+ // for (PartialPath storageGroup :
IoTDB.schemaEngine.getAllStorageGroupPaths()) {
+ // this.releaseWalDirectByteBufferPoolInOneStorageGroup(storageGroup);
+ // }
+ dataRegionMap.clear();
+ }
+
+ @Override
+ public void shutdown(long milliseconds) throws ShutdownException {
+ try {
+ for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
dataRegionMap.values()) {
+ ThreadUtils.stopThreadPool(
+ virtualStorageGroupProcessor.getTimedCompactionScheduleTask(),
+ ThreadName.COMPACTION_SCHEDULE);
+ ThreadUtils.stopThreadPool(
+ virtualStorageGroupProcessor.getWALTrimScheduleTask(),
ThreadName.WAL_TRIM);
+ }
+ forceCloseAllProcessor();
+ } catch (TsFileProcessorException e) {
+ throw new ShutdownException(e);
+ }
+ shutdownTimedService(ttlCheckThread, "TTlCheckThread");
+ shutdownTimedService(seqMemtableTimedFlushCheckThread,
"SeqMemtableTimedFlushCheckThread");
+ shutdownTimedService(unseqMemtableTimedFlushCheckThread,
"UnseqMemtableTimedFlushCheckThread");
+ shutdownTimedService(tsFileTimedCloseCheckThread,
"TsFileTimedCloseCheckThread");
+ recoveryThreadPool.shutdownNow();
+ dataRegionMap.clear();
+ }
+
+ private void shutdownTimedService(ScheduledExecutorService pool, String
poolName) {
+ if (pool != null) {
+ pool.shutdownNow();
+ try {
+ pool.awaitTermination(30, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ logger.warn("{} still doesn't exit after 30s", poolName);
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ private void stopTimedServiceAndThrow(ScheduledExecutorService pool, String
poolName)
+ throws ShutdownException {
+ if (pool != null) {
+ pool.shutdownNow();
+ try {
+ pool.awaitTermination(30, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ logger.warn("{} still doesn't exit after 30s", poolName);
+ throw new ShutdownException(e);
+ }
+ }
+ }
+
+ @Override
+ public ServiceType getID() {
+ return ServiceType.STORAGE_ENGINE_SERVICE;
+ }
+
+ /**
+ * build a new storage group processor
+ *
+ * @param virtualStorageGroupId virtual storage group id e.g. 1
+ * @param logicalStorageGroupName logical storage group name e.g. root.sg1
+ */
+ public VirtualStorageGroupProcessor buildNewStorageGroupProcessor(
+ String logicalStorageGroupName, String virtualStorageGroupId, long ttl)
+ throws StorageGroupProcessorException {
+ VirtualStorageGroupProcessor processor;
+ logger.info(
+ "construct a processor instance, the storage group is {}, Thread is
{}",
+ logicalStorageGroupName,
+ Thread.currentThread().getId());
+ processor =
+ new VirtualStorageGroupProcessor(
+ systemDir + File.separator + logicalStorageGroupName,
+ virtualStorageGroupId,
+ fileFlushPolicy,
+ logicalStorageGroupName);
+ processor.setDataTTL(ttl);
+ processor.setCustomFlushListeners(customFlushListeners);
+ processor.setCustomCloseFileListeners(customCloseFileListeners);
+ return processor;
+ }
+
+ /** This function is just for unit test. */
+ @TestOnly
+ public synchronized void reset() {
+ dataRegionMap.clear();
+ }
+
+ /**
+ * insert an InsertRowNode to a storage group.
+ *
+ * @param insertRowNode
+ */
+ // TODO:(New insert)
+ public void insert(ConsensusGroupId dataRegionId, InsertRowNode
insertRowNode)
+ throws StorageEngineException, MetadataException {
+ if (enableMemControl) {
+ try {
+ blockInsertionIfReject(null);
+ } catch (WriteProcessException e) {
+ throw new StorageEngineException(e);
+ }
+ }
+
+ VirtualStorageGroupProcessor dataRegion = dataRegionMap.get(dataRegionId);
+
+ try {
+ dataRegion.insert(insertRowNode);
+ } catch (WriteProcessException e) {
+ throw new StorageEngineException(e);
+ }
+ }
+
+ /** insert an InsertTabletNode to a storage group */
+ // TODO:(New insert)
+ public void insertTablet(ConsensusGroupId dataRegionId, InsertTabletNode
insertTabletNode)
+ throws StorageEngineException, BatchProcessException {
+ 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 dataRegion = dataRegionMap.get(dataRegionId);
+ dataRegion.insertTablet(insertTabletNode);
+ }
+
+ /** flush command Sync asyncCloseOneProcessor all file node processors. */
+ public void syncCloseAllProcessor() {
+ logger.info("Start closing all storage group processor");
+ for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
dataRegionMap.values()) {
+ if (virtualStorageGroupProcessor != null) {
+ virtualStorageGroupProcessor.syncCloseAllWorkingTsFileProcessors();
+ }
+ }
+ }
+
+ public void forceCloseAllProcessor() throws TsFileProcessorException {
+ logger.info("Start force closing all storage group processor");
+ for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
dataRegionMap.values()) {
+ if (virtualStorageGroupProcessor != null) {
+ virtualStorageGroupProcessor.forceCloseAllWorkingTsFileProcessors();
+ }
+ }
+ }
+
+ public void setTTL(List<ConsensusGroupId> dataRegionIdList, long dataTTL) {
+ for (ConsensusGroupId dataRegionId : dataRegionIdList) {
+ VirtualStorageGroupProcessor dataRegion =
dataRegionMap.get(dataRegionId);
+ if (dataRegion != null) {
+ dataRegion.setDataTTL(dataTTL);
+ }
+ }
+ }
+
+ public void setFileFlushPolicy(TsFileFlushPolicy fileFlushPolicy) {
+ this.fileFlushPolicy = fileFlushPolicy;
+ }
+
+ /**
+ * Get a map indicating which storage groups have working TsFileProcessors
and its associated
+ * partitionId and whether it is sequence or not.
+ *
+ * @return storage group -> a list of partitionId-isSequence pairs
+ */
+ public Map<String, List<Pair<Long, Boolean>>>
getWorkingStorageGroupPartitions() {
+ Map<String, List<Pair<Long, Boolean>>> res = new ConcurrentHashMap<>();
+ for (Entry<ConsensusGroupId, VirtualStorageGroupProcessor> entry :
dataRegionMap.entrySet()) {
+ VirtualStorageGroupProcessor virtualStorageGroupProcessor =
entry.getValue();
+ if (virtualStorageGroupProcessor != null) {
+ List<Pair<Long, Boolean>> partitionIdList = new ArrayList<>();
+ for (TsFileProcessor tsFileProcessor :
+ virtualStorageGroupProcessor.getWorkSequenceTsFileProcessors()) {
+ Pair<Long, Boolean> tmpPair = new
Pair<>(tsFileProcessor.getTimeRangeId(), true);
+ partitionIdList.add(tmpPair);
+ }
+
+ for (TsFileProcessor tsFileProcessor :
+ virtualStorageGroupProcessor.getWorkUnsequenceTsFileProcessors()) {
+ Pair<Long, Boolean> tmpPair = new
Pair<>(tsFileProcessor.getTimeRangeId(), false);
+ partitionIdList.add(tmpPair);
+ }
+
+ res.put(virtualStorageGroupProcessor.getStorageGroupPath(),
partitionIdList);
+ }
+ }
+
+ return res;
+ }
+
+ /**
+ * Add a listener to listen flush start/end events. Notice that this
addition only applies to
+ * TsFileProcessors created afterwards.
+ *
+ * @param listener
+ */
+ public void registerFlushListener(FlushListener listener) {
+ customFlushListeners.add(listener);
+ }
+
+ /**
+ * Add a listener to listen file close events. Notice that this addition
only applies to
+ * TsFileProcessors created afterwards.
+ *
+ * @param listener
+ */
+ public void registerCloseFileListener(CloseFileListener listener) {
+ customCloseFileListeners.add(listener);
+ }
+
+ // When registering a new region, the coordinator needs to register the
corresponding region with
+ // the local engine before adding the corresponding consensusGroup to the
consensus layer
+ public VirtualStorageGroupProcessor createDataRegion(DataRegionId regionId,
String sg, long ttl)
+ throws StorageEngineException {
+ try {
+ VirtualStorageGroupProcessor dataRegion =
+ buildNewStorageGroupProcessor(sg, regionId.toString(), ttl);
+ dataRegionMap.put(regionId, dataRegion);
+ } catch (StorageGroupProcessorException e) {
+ throw new StorageEngineException(e);
+ }
+ return null;
+ }
+
+ public VirtualStorageGroupProcessor getDataRegion(DataRegionId regionId) {
+ return dataRegionMap.get(regionId);
+ }
+
+ static class InstanceHolder {
+
+ private static final StorageEngineV2 INSTANCE = new StorageEngineV2();
+
+ private InstanceHolder() {
+ // forbidding instantiation
+ }
+ }
+}
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 06706b1361..aa5c5cc09b 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,8 @@ 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.InsertRowNode;
+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;
@@ -180,6 +182,49 @@ public abstract class AbstractMemTable implements
IMemTable {
}
}
+ @Override
+ public void insert(InsertRowNode insertRowNode) {
+ // if this insert plan isn't from storage engine (mainly from test), we
should set a temp device
+ // id for it
+ if (insertRowNode.getDeviceID() == null) {
+ insertRowNode.setDeviceID(
+
DeviceIDFactory.getInstance().getDeviceID(insertRowNode.getDevicePath()));
+ }
+
+ // updatePlanIndexes(insertRowNode.getIndex());
+ updatePlanIndexes(0);
+ String[] measurements = insertRowNode.getMeasurements();
+ Object[] values = insertRowNode.getValues();
+
+ List<IMeasurementSchema> schemaList = new ArrayList<>();
+ List<TSDataType> dataTypes = new ArrayList<>();
+ for (int i = 0; i < insertRowNode.getMeasurements().length; i++) {
+ if (measurements[i] == null) {
+ continue;
+ }
+ IMeasurementSchema schema = insertRowNode.getMeasurementSchemas()[i];
+ schemaList.add(schema);
+ dataTypes.add(schema.getType());
+ }
+ memSize += MemUtils.getRecordsSize(dataTypes, values, disableMemControl);
+ write(insertRowNode.getDeviceID(), schemaList, insertRowNode.getTime(),
values);
+
+ int pointsInserted = insertRowNode.getMeasurements().length;
+
+ totalPointsNum += pointsInserted;
+
+ if
(MetricConfigDescriptor.getInstance().getMetricConfig().getEnableMetric()) {
+ MetricsService.getInstance()
+ .getMetricManager()
+ .count(
+ pointsInserted,
+ Metric.QUANTITY.toString(),
+ MetricLevel.IMPORTANT,
+ Tag.NAME.toString(),
+ METRIC_POINT_IN);
+ }
+ }
+
@Override
public void insertAlignedRow(InsertRowPlan insertRowPlan) {
// if this insert plan isn't from storage engine, we should set a temp
device id for it
@@ -226,6 +271,52 @@ public abstract class AbstractMemTable implements
IMemTable {
}
}
+ @Override
+ public void insertAlignedRow(InsertRowNode insertRowNode) {
+ // if this insert node isn't from storage engine, we should set a temp
device id for it
+ if (insertRowNode.getDeviceID() == null) {
+ insertRowNode.setDeviceID(
+
DeviceIDFactory.getInstance().getDeviceID(insertRowNode.getDevicePath()));
+ }
+
+ // updatePlanIndexes(insertRowNode.getIndex());
+ updatePlanIndexes(0);
+ String[] measurements = insertRowNode.getMeasurements();
+ List<IMeasurementSchema> schemaList = new ArrayList<>();
+ List<TSDataType> dataTypes = new ArrayList<>();
+ for (int i = 0; i < insertRowNode.getMeasurements().length; i++) {
+ if (measurements[i] == null) {
+ continue;
+ }
+ IMeasurementSchema schema = insertRowNode.getMeasurementSchemas()[i];
+ schemaList.add(schema);
+ dataTypes.add(schema.getType());
+ }
+ if (schemaList.isEmpty()) {
+ return;
+ }
+ memSize +=
+ MemUtils.getAlignedRecordsSize(dataTypes, insertRowNode.getValues(),
disableMemControl);
+ writeAlignedRow(
+ insertRowNode.getDeviceID(),
+ schemaList,
+ insertRowNode.getTime(),
+ insertRowNode.getValues());
+ int pointsInserted = insertRowNode.getMeasurements().length;
+ totalPointsNum += pointsInserted;
+
+ if
(MetricConfigDescriptor.getInstance().getMetricConfig().getEnableMetric()) {
+ MetricsService.getInstance()
+ .getMetricManager()
+ .count(
+ pointsInserted,
+ Metric.QUANTITY.toString(),
+ MetricLevel.IMPORTANT,
+ Tag.NAME.toString(),
+ METRIC_POINT_IN);
+ }
+ }
+
@Override
public void insertTablet(InsertTabletPlan insertTabletPlan, int start, int
end)
throws WriteProcessException {
@@ -278,6 +369,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,
@@ -328,6 +471,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.getMeasurementSchemas().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 +527,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.getMeasurementSchemas().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 69aeb47ffe..9b5a2b3580 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,8 @@ 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.InsertRowNode;
+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;
@@ -100,6 +102,10 @@ public interface IMemTable {
void insertAlignedRow(InsertRowPlan insertRowPlan);
+ void insert(InsertRowNode insertRowNode);
+
+ void insertAlignedRow(InsertRowNode insertRowNode);
+
/**
* 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,
@@ -115,6 +121,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 df6e20bb88..5fd5aad7f5 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,8 @@ 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.InsertRowNode;
+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;
@@ -257,6 +259,64 @@ public class TsFileProcessor {
tsFileResource.updatePlanIndexes(insertRowPlan.getIndex());
}
+ /**
+ * insert data in an InsertRowNode into the workingMemtable.
+ *
+ * @param insertRowNode physical plan of insertion
+ */
+ public void insert(InsertRowNode insertRowNode) throws WriteProcessException
{
+
+ if (workMemTable == null) {
+ if (enableMemControl) {
+ workMemTable = new PrimitiveMemTable(enableMemControl);
+ MemTableManager.getInstance().addMemtableNumber();
+ } else {
+ workMemTable =
MemTableManager.getInstance().getAvailableMemTable(storageGroupName);
+ }
+ }
+
+ long[] memIncrements = null;
+ if (enableMemControl) {
+ if (insertRowNode.isAligned()) {
+ // memIncrements = checkAlignedMemCostAndAddToTspInfo(insertRowNode);
+ } else {
+ // memIncrements = checkMemCostAndAddToTspInfo(insertRowNode);
+ }
+ }
+
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
+ try {
+ getLogNode().write(insertRowNode);
+ } catch (Exception e) {
+ if (enableMemControl && memIncrements != null) {
+ rollbackMemoryInfo(memIncrements);
+ }
+ throw new WriteProcessException(
+ String.format(
+ "%s: %s write WAL failed",
+ storageGroupName,
tsFileResource.getTsFile().getAbsolutePath()),
+ e);
+ }
+ }
+
+ if (insertRowNode.isAligned()) {
+ workMemTable.insertAlignedRow(insertRowNode);
+ } else {
+ workMemTable.insert(insertRowNode);
+ }
+
+ // update start time of this memtable
+ tsFileResource.updateStartTime(
+ insertRowNode.getDeviceID().toStringID(), insertRowNode.getTime());
+ // 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(
+ insertRowNode.getDeviceID().toStringID(), insertRowNode.getTime());
+ }
+ // tsFileResource.updatePlanIndexes(insertRowNode.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
@@ -340,6 +400,93 @@ 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);
+ }
+
+ try {
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
+ // TODO(WAL)
+ // Start and end should be removed from new WAL
+ // insertTabletNode.setStart(start);
+ // insertTabletNode.setEnd(end);
+ getLogNode().write(insertTabletNode);
+ }
+ } 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 +618,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.getMeasurementSchemas()[i].getMeasurementId();
+ 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 +680,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 2f9de94ab4..3b95fad455 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,8 @@ 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.InsertRowNode;
+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;
@@ -116,6 +118,7 @@ import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -197,6 +200,8 @@ public class VirtualStorageGroupProcessor {
private AtomicInteger upgradeFileCount = new AtomicInteger();
+ private AtomicBoolean isSettling = new AtomicBoolean();
+
/** virtual storage group id */
private String virtualStorageGroupId;
/** logical storage group name */
@@ -452,6 +457,14 @@ public class VirtualStorageGroupProcessor {
return ret;
}
+ public AtomicBoolean getIsSettling() {
+ return isSettling;
+ }
+
+ public void setSettling(boolean isSettling) {
+ this.isSettling.set(isSettling);
+ }
+
/** this class is used to store recovering context */
private class RecoveryContext {
/** number of files to be recovered */
@@ -909,6 +922,44 @@ public class VirtualStorageGroupProcessor {
}
}
+ // TODO: (New Insert)
+ public void insert(InsertRowNode insertRowNode)
+ throws WriteProcessException, TriggerExecutionException {
+ // reject insertions that are out of ttl
+ if (!isAlive(insertRowNode.getTime())) {
+ throw new OutOfTTLException(insertRowNode.getTime(),
(System.currentTimeMillis() - dataTTL));
+ }
+ writeLock("InsertRow");
+ try {
+ // init map
+ long timePartitionId =
StorageEngine.getTimePartition(insertRowNode.getTime());
+
+ lastFlushTimeManager.ensureFlushedTimePartition(timePartitionId);
+
+ boolean isSequence =
+ insertRowNode.getTime()
+ > lastFlushTimeManager.getFlushedTime(
+ timePartitionId,
insertRowNode.getDevicePath().getFullPath());
+
+ // is unsequence and user set config to discard out of order data
+ if (!isSequence
+ &&
IoTDBDescriptor.getInstance().getConfig().isEnableDiscardOutOfOrderData()) {
+ return;
+ }
+
+ lastFlushTimeManager.ensureLastTimePartition(timePartitionId);
+
+ // fire trigger before insertion
+ // TriggerEngine.fire(TriggerEvent.BEFORE_INSERT, insertRowNode);
+ // insert to sequence or unSequence file
+ insertToTsFileProcessor(insertRowNode, isSequence, timePartitionId);
+ // fire trigger after insertion
+ // TriggerEngine.fire(TriggerEvent.AFTER_INSERT, insertRowNode);
+ } finally {
+ writeUnlock();
+ }
+ }
+
/**
* Insert a tablet (rows belonging to the same devices) into this storage
group.
*
@@ -1028,6 +1079,102 @@ 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];
+ // always in some time partition
+ // 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;
@@ -1095,6 +1242,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;
@@ -1145,6 +1354,31 @@ public class VirtualStorageGroupProcessor {
}
}
+ private void insertToTsFileProcessor(
+ InsertRowNode insertRowNode, boolean sequence, long timePartitionId)
+ throws WriteProcessException {
+ TsFileProcessor tsFileProcessor =
getOrCreateTsFileProcessor(timePartitionId, sequence);
+ if (tsFileProcessor == null) {
+ return;
+ }
+
+ tsFileProcessor.insert(insertRowNode);
+
+ // try to update the latest time of the device of this tsRecord
+ lastFlushTimeManager.updateLastTime(
+ timePartitionId, insertRowNode.getDevicePath().getFullPath(),
insertRowNode.getTime());
+
+ long globalLatestFlushTime =
+
lastFlushTimeManager.getGlobalFlushedTime(insertRowNode.getDevicePath().getFullPath());
+
+ // tryToUpdateInsertLastCache(insertRowNode, globalLatestFlushTime);
+
+ // check memtable size and may asyncTryToFlush the work memtable
+ if (tsFileProcessor.shouldFlush()) {
+ fileFlushPolicy.apply(this, tsFileProcessor, sequence);
+ }
+ }
+
private void tryToUpdateInsertLastCache(InsertRowPlan 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 96ccd482c7..f0ec0e1a4b 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
@@ -39,7 +39,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;
@@ -61,12 +62,12 @@ public abstract class InsertNode extends PlanNode {
PlanNodeId id,
PartialPath devicePath,
boolean isAligned,
- MeasurementSchema[] measurements,
+ MeasurementSchema[] measurementSchemas,
TSDataType[] dataTypes) {
super(id);
this.devicePath = devicePath;
this.isAligned = isAligned;
- this.measurements = measurements;
+ this.measurementSchemas = measurementSchemas;
this.dataTypes = dataTypes;
}
@@ -94,12 +95,22 @@ public abstract class InsertNode extends PlanNode {
isAligned = aligned;
}
- public MeasurementSchema[] getMeasurements() {
- return measurements;
+ public MeasurementSchema[] getMeasurementSchemas() {
+ return measurementSchemas;
+ }
+
+ public void setMeasurementSchemas(MeasurementSchema[] measurementSchemas) {
+ this.measurementSchemas = measurementSchemas;
}
- public void setMeasurements(MeasurementSchema[] measurements) {
- this.measurements = measurements;
+ public String[] getMeasurements() {
+ if (measurements == null) {
+ measurements = new String[measurementSchemas.length];
+ for (int i = 0; i < measurementSchemas.length; i++) {
+ measurements[i] = measurementSchemas[i].getMeasurementId();
+ }
+ }
+ return measurements;
}
public TSDataType[] getDataTypes() {
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 f48cf01272..700fdbab93 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
@@ -62,13 +62,13 @@ public class InsertTabletNode extends InsertNode {
PlanNodeId id,
PartialPath devicePath,
boolean isAligned,
- MeasurementSchema[] measurements,
+ MeasurementSchema[] measurementSchemas,
TSDataType[] dataTypes,
long[] times,
BitMap[] bitMaps,
Object[] columns,
int rowCount) {
- super(id, devicePath, isAligned, measurements, dataTypes);
+ super(id, devicePath, isAligned, measurementSchemas, dataTypes);
this.times = times;
this.bitMaps = bitMaps;
this.columns = columns;
@@ -228,7 +228,7 @@ public class InsertTabletNode extends InsertNode {
getPlanNodeId(),
devicePath,
isAligned,
- measurements,
+ measurementSchemas,
dataTypes,
subTimes,
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 854fb0348a..66bfc6ab84 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;
@@ -105,6 +106,33 @@ public class MemUtils {
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(
+ 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.
@@ -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
diff --git
a/server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java
b/server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java
index 5283cafaca..32c77afaf4 100644
---
a/server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java
+++
b/server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.conf.directories.DirectoryManager;
import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNode;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.utils.MmapUtil;
import org.apache.iotdb.db.utils.ThreadUtils;
@@ -133,6 +134,33 @@ public class ExclusiveWriteLogNode implements
WriteLogNode, Comparable<Exclusive
}
}
+ @Override
+ public void write(PlanNode node) throws IOException {
+ if (deleted.get()) {
+ throw new IOException("WAL node deleted");
+ }
+ lock.lock();
+ try {
+ putLog(node);
+ if (bufferedLogNum >= config.getFlushWalThreshold()) {
+ sync();
+ }
+ } catch (BufferOverflowException e) {
+ // if the size of a single plan bigger than logBufferWorking
+ // we need to clear the buffer to drop something wrong that has written.
+ logBufferWorking.clear();
+ // TODO(WAL)
+ // int neededSize = node.getSerializedSize();
+ // throw new IOException(
+ // "Log cannot fit into the buffer, please increase
wal_buffer_size to more than
+ // "
+ // + neededSize * 2,
+ // e);
+ } finally {
+ lock.unlock();
+ }
+ }
+
private void putLog(PhysicalPlan plan) {
try {
plan.serialize(logBufferWorking);
@@ -149,6 +177,24 @@ public class ExclusiveWriteLogNode implements
WriteLogNode, Comparable<Exclusive
bufferedLogNum++;
}
+ private void putLog(PlanNode node) {
+ try {
+ // TODO(WAL)
+ // node.serialize(logBufferWorking);
+ } catch (BufferOverflowException e) {
+ bufferOverflowNum++;
+ if (bufferOverflowNum > 200) {
+ logger.info(
+ "WAL bytebuffer overflows too many times. If this occurs
frequently, please increase wal_buffer_size.");
+ bufferOverflowNum = 0;
+ }
+ sync();
+ // TODO(WAL)
+ // node.serialize(logBufferWorking);
+ }
+ bufferedLogNum++;
+ }
+
@Override
public void close() {
sync();
diff --git
a/server/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java
b/server/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java
index 952238f57e..e5b1f82e93 100644
--- a/server/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.db.writelog.node;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNode;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.writelog.io.ILogReader;
@@ -36,6 +37,14 @@ public interface WriteLogNode {
*/
void write(PhysicalPlan plan) throws IOException;
+ /**
+ * Write a wal for a PlanNode. First, the PhysicalPlan will be conveyed to
byte[]. Then the byte[]
+ * will be put into a cache. When the cache is full, the logs in the cache
will be synced to disk.
+ *
+ * @param plan - a PhysicalPlan
+ */
+ void write(PlanNode plan) throws IOException;
+
/** Sync and close streams. */
void close() throws IOException;
