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:11 UTC
[iotdb] 04/11: dev
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 fe594fb46254b8df15c91c8ecf6ecb9c6e8a5763
Author: HTHou <hh...@outlook.com>
AuthorDate: Thu Mar 24 21:43:32 2022 +0800
dev
---
.../org/apache/iotdb/db/engine/StorageEngine.java | 17 --------
.../db/engine/storagegroup/TsFileProcessor.java | 32 +++++++--------
.../planner/plan/node/write/InsertTabletNode.java | 16 ++++++++
.../db/writelog/node/ExclusiveWriteLogNode.java | 46 ++++++++++++++++++++++
.../iotdb/db/writelog/node/WriteLogNode.java | 9 +++++
5 files changed, 87 insertions(+), 33 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 b877735..2c0075d 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,7 +53,6 @@ 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;
@@ -649,22 +648,6 @@ 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/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index 9f78c97..482f5eb 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
@@ -380,22 +380,22 @@ public class TsFileProcessor {
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 (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
+ insertTabletNode.setStart(start);
+ insertTabletNode.setEnd(end);
+ // TODO(WAL)
+ 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()) {
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 5bcab88..3b03c63 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
@@ -89,4 +89,20 @@ public class InsertTabletNode extends InsertNode {
public BitMap[] getBitMaps() {
return bitMaps;
}
+
+ public int getStart() {
+ return start;
+ }
+
+ public void setStart(int start) {
+ this.start = start;
+ }
+
+ public int getEnd() {
+ return end;
+ }
+
+ public void setEnd(int end) {
+ this.end = end;
+ }
}
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 5283caf..32c77af 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 952238f..e5b1f82 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;