This is an automated email from the ASF dual-hosted git repository. tanxinyu pushed a commit to branch master_performance in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5cef83ff3cb07579c4116d2c8557c1280f5d2a82 Author: Alan Choo <[email protected]> AuthorDate: Thu Nov 25 21:32:09 2021 +0800 [IOTDB-2027] Ignore too many WAL BufferOverflow log (#4467) --- .../apache/iotdb/db/qp/physical/PhysicalPlan.java | 4 ++ .../db/writelog/node/ExclusiveWriteLogNode.java | 8 ++++ .../apache/iotdb/db/writelog/WriteLogNodeTest.java | 49 ++++++++++++++++++++++ 3 files changed, 61 insertions(+) diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java index 1951ba0..f77021a 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java @@ -78,6 +78,7 @@ import org.slf4j.LoggerFactory; import java.io.DataOutputStream; import java.io.IOException; +import java.nio.BufferOverflowException; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collections; @@ -198,6 +199,9 @@ public abstract class PhysicalPlan { } catch (UnsupportedOperationException e) { // ignore and throw throw e; + } catch (BufferOverflowException e) { + buffer.reset(); + throw e; } catch (Exception e) { logger.error( "Rollback buffer entry because error occurs when serializing this physical plan.", e); 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 5b281c6..a64d0f5 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 @@ -76,6 +76,8 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive private final AtomicBoolean deleted = new AtomicBoolean(false); + private int bufferOverflowNum = 0; + /** * constructor of ExclusiveWriteLogNode. * @@ -129,6 +131,12 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive try { plan.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(); plan.serialize(logBufferWorking); } diff --git a/server/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeTest.java b/server/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeTest.java index 1b24a05..dc6904e 100644 --- a/server/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeTest.java +++ b/server/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeTest.java @@ -399,4 +399,53 @@ public class WriteLogNodeTest { MmapUtil.clean((MappedByteBuffer) byteBuffer); } } + + @Test + public void testBufferOverflowAndRewrite() throws IOException, IllegalPathException { + String identifier = "root.logTestDevice"; + + InsertRowPlan insertPlan = + new InsertRowPlan( + new PartialPath(identifier), + 100, + new String[] {"s1", "s2", "s3", "s4"}, + new TSDataType[] { + TSDataType.DOUBLE, TSDataType.INT64, TSDataType.TEXT, TSDataType.BOOLEAN + }, + new String[] {"1.0", "15", "str", "false"}); + + // get InsertRowPlan byte size + ByteBuffer tmpBuffer = + ByteBuffer.allocate(IoTDBDescriptor.getInstance().getConfig().getWalBufferSize() / 2); + insertPlan.serialize(tmpBuffer); + int size = tmpBuffer.position(); + // allocate buffers + ByteBuffer[] byteBuffers = new ByteBuffer[2]; + byteBuffers[0] = ByteBuffer.allocateDirect(size + 1); + byteBuffers[1] = ByteBuffer.allocateDirect(size + 1); + WriteLogNode logNode = new ExclusiveWriteLogNode(identifier); + logNode.initBuffer(byteBuffers); + // write InsertRowPlan to WAL buffer + logNode.write(insertPlan); + insertPlan.setTime(200); + logNode.write(insertPlan); + + logNode.close(); + + File walFile = + new File(config.getWalDir() + File.separator + identifier + File.separator + "wal1"); + assertTrue(walFile.exists()); + + ILogReader reader = logNode.getLogReader(); + insertPlan.setTime(100); + assertEquals(insertPlan, reader.next()); + insertPlan.setTime(200); + assertEquals(insertPlan, reader.next()); + reader.close(); + + ByteBuffer[] array = logNode.delete(); + for (ByteBuffer byteBuffer : array) { + MmapUtil.clean((MappedByteBuffer) byteBuffer); + } + } }
