JackieTien97 commented on code in PR #18580:
URL: https://github.com/apache/iotdb/pull/18580#discussion_r3931349257
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/template/Template.java:
##########
@@ -214,6 +215,35 @@ public void deserialize(ByteBuffer buffer) {
}
}
+ public void deserialize(InputStream inputStream) throws IOException {
+ id = ReadWriteIOUtils.readInt(inputStream);
+ name = ReadWriteIOUtils.readString(inputStream);
+ isDirectAligned = ReadWriteIOUtils.readBool(inputStream);
+ int schemaSize = ReadWriteIOUtils.readInt(inputStream);
+ if (schemaSize < 0) {
+ throw new IOException(String.format(SchemaMessages.INVALID_INPUT,
schemaSize));
+ }
+ schemaMap = new ConcurrentHashMap<>(schemaSize);
+ for (int i = 0; i < schemaSize; i++) {
+ String schemaName = ReadWriteIOUtils.readString(inputStream);
+ byte flag = ReadWriteIOUtils.readByte(inputStream);
+ IMeasurementSchema measurementSchema;
+ if (flag == (byte) 0) {
+ measurementSchema =
+ new MeasurementSchema(
+ ReadWriteIOUtils.readString(inputStream),
+ TSDataType.deserializeFrom(inputStream),
+ TSEncoding.deserialize(ReadWriteIOUtils.readByte(inputStream)),
+
CompressionType.deserialize(ReadWriteIOUtils.readByte(inputStream)));
+ } else if (flag == (byte) 1) {
+ measurementSchema =
VectorMeasurementSchema.deserializeFrom(inputStream);
Review Comment:
[P1] Preserve snapshots containing per-column-compressor vector schemas.
This changes template recovery from
`VectorMeasurementSchema.partialDeserializeFrom(ByteBuffer)` to
`VectorMeasurementSchema.deserializeFrom(InputStream)`. In the TsFile version
used by this branch, the stream overload allocates `measurementSize + 1`
compressor bytes, performs a single `read`, and then compares the result with
`measurementSize`
([implementation](https://github.com/apache/tsfile/blob/7d25839f81bfaf6bfcac47e3e993ed67da546171/java/tsfile/src/main/java/org/apache/tsfile/write/schema/VectorMeasurementSchema.java#L420-L425)).
A normal buffered/file stream therefore reads all `measurementSize + 1` bytes
and throws; a short read of `measurementSize` would instead leave one byte
behind and misalign subsequent data. I reproduced this with a template
containing `new VectorMeasurementSchema("vector", new String[]{"s1", "s2"}, new
TSDataType[]{INT32, BOOLEAN})`: the existing ByteBuffer path accepts it and
snapshot serialization succeeds, but this path fails with `Unexpected end of
stream when reading compressors`. Because the write path accepts this state, a
later ConfigNode snapshot recovery can fail. Please read exactly
`measurementSize + 1` bytes with a loop/readFully, either by fixing and
updating TsFile or via an IoTDB-side decoder, and add a vector-template
snapshot regression test.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/CNPhysicalPlanGenerator.java:
##########
@@ -513,15 +517,22 @@ private void generateDatabasePhysicalPlan() {
}
private void generateTemplatePlan() {
- try (final BufferedInputStream bufferedInputStream =
- new BufferedInputStream(templateInputStream)) {
- final ByteBuffer byteBuffer =
ByteBuffer.wrap(IOUtils.toByteArray(bufferedInputStream));
+ // The template snapshot file is read into a direct buffer instead of the
heap, so generating
Review Comment:
[nit] Update this comment and the PR description to match the final
implementation. This code no longer reads the file into a direct buffer:
`templateInputStream` is the bounded stream created by `SnapshotStreamFactory`,
and `Template.deserialize(InputStream)` consumes it incrementally. The PR
description also still describes direct-buffer and snapshot
`ByteBufferInputStream` paths. Please describe this as bounded streaming
instead.
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/snapshot/ByteBufferInputStream.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.commons.snapshot;
+
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.Objects;
+
+/**
+ * An {@link InputStream} over a {@link ByteBuffer} (typically a direct buffer
filled from a
+ * snapshot file). Unlike {@link java.io.ByteArrayInputStream}, the underlying
bytes live outside
+ * the heap, so a snapshot file can be parsed without first copying it into
heap memory. {@code
+ * mark}/{@code reset} behave like {@link java.io.ByteArrayInputStream}: they
are position-based and
+ * independent of the read limit.
+ */
+public final class ByteBufferInputStream extends InputStream {
+
+ private final ByteBuffer buffer;
+ private int markPosition = -1;
+
+ public ByteBufferInputStream(final ByteBuffer buffer) {
+ this.buffer = Objects.requireNonNull(buffer);
+ }
+
+ @Override
+ public int read() {
+ return buffer.hasRemaining() ? buffer.get() & 0xFF : -1;
+ }
+
+ @Override
+ public int read(final byte[] b, final int off, final int len) {
+ Objects.checkFromIndexSize(off, len, b.length);
+ if (len == 0) {
+ return 0;
+ }
+ if (!buffer.hasRemaining()) {
+ return -1;
+ }
+ final int toRead = Math.min(len, buffer.remaining());
+ buffer.get(b, off, toRead);
+ return toRead;
+ }
+
+ @Override
+ public long skip(final long n) {
+ if (n <= 0) {
+ return 0;
+ }
+ final long skipped = Math.min(n, buffer.remaining());
+ buffer.position(buffer.position() + (int) skipped);
+ return skipped;
+ }
+
+ @Override
+ public int available() {
+ return buffer.remaining();
+ }
+
+ @Override
+ public boolean markSupported() {
+ return true;
+ }
+
+ @Override
+ public synchronized void mark(final int readLimit) {
+ markPosition = buffer.position();
+ }
+
+ @Override
+ public synchronized void reset() {
+ // Like ByteArrayInputStream, reset is position-based and independent of
the read limit. When
Review Comment:
[nit] Match the stated ByteArrayInputStream reset contract, or correct the
documentation. `ByteArrayInputStream` initializes its mark to the initial
offset, so `reset()` before an explicit `mark()` rewinds to that offset. Here
`markPosition` starts at -1 and reset is a no-op, despite the class-level claim
that mark/reset behave like ByteArrayInputStream. If the class remains,
initialize the mark from the initial buffer position or explicitly document the
different behavior.
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/snapshot/SnapshotStreamFactory.java:
##########
@@ -0,0 +1,125 @@
+/*
+ * 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.commons.snapshot;
+
+import org.apache.iotdb.commons.i18n.CommonMessages;
+
+import java.io.BufferedInputStream;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.lang.ref.SoftReference;
+
+/**
+ * Creates the buffered streams used to write and read ConfigNode snapshot
files.
+ *
+ * <p>The buffer size is bounded by {@link #bufferSizeMax}, which is
configurable through {@code
+ * config_node_snapshot_buffer_size_max} (0 disables buffering). This replaces
the previous fixed
+ * 32MB buffer of {@code PartitionInfo}: memory-constrained deployments can
lower the cap, while the
+ * default keeps snapshot I/O fast for large partition tables. Write buffers
are pooled per thread
+ * through {@link SoftReference}s, so consecutive snapshots on the same thread
reuse the array
+ * instead of re-allocating it.
+ */
+public final class SnapshotStreamFactory {
+
+ /** Default upper bound of a snapshot stream buffer, 4MB. */
+ public static final long DEFAULT_BUFFER_SIZE_MAX = 4 * 1024 * 1024L;
+
+ private static volatile long bufferSizeMax = DEFAULT_BUFFER_SIZE_MAX;
+
+ /** Thread-local pool of reusable write buffers, kept only as long as GC
allows. */
+ private static final ThreadLocal<SoftReference<byte[]>> WRITE_BUFFER_POOL =
+ ThreadLocal.withInitial(() -> new SoftReference<>(null));
+
+ private SnapshotStreamFactory() {
+ // Utility class
+ }
+
+ /**
+ * Set the upper bound of snapshot stream buffers, in bytes. Values below or
equal to zero disable
+ * buffering entirely (the raw stream is returned unchanged). Thread-safe;
takes effect on the
+ * next stream creation.
+ */
+ public static void setBufferSizeMax(final long sizeInBytes) {
+ if (sizeInBytes > Integer.MAX_VALUE) {
+ throw new IllegalArgumentException(
+ String.format(
+ CommonMessages
+
.EXCEPTION_SNAPSHOT_BUFFER_SIZE_MUST_NOT_EXCEED_ARG_BYTES_BUT_WAS_ARG_D1DA6F7E,
+ Integer.MAX_VALUE,
+ sizeInBytes));
+ }
+ bufferSizeMax = Math.max(0L, sizeInBytes);
+ }
+
+ public static long getBufferSizeMax() {
+ return bufferSizeMax;
+ }
+
+ /**
+ * Wrap {@code raw} with a buffered output stream whose buffer is at most
{@link #bufferSizeMax}
+ * bytes. The buffer is reusable, so the same thread writing several
snapshots does not repeatedly
+ * allocate it.
+ *
+ * @param raw the raw stream to buffer
+ * @return a buffered stream, or {@code raw} itself if buffering is disabled
+ */
+ public static OutputStream createOutputStream(final OutputStream raw) {
+ final int bufferSize = (int) bufferSizeMax;
+ return bufferSize <= 0 ? raw : new ReusableBufferedOutputStream(raw,
bufferSize);
+ }
+
+ /**
+ * Wrap {@code raw} with a buffered input stream whose buffer is sized to at
most {@code
+ * fileSize}, capped by {@link #bufferSizeMax}. Reading never allocates more
than the configured
+ * cap, no matter how large the snapshot file is.
+ *
+ * @param raw the raw stream to buffer
+ * @param fileSize size of the file being read, in bytes
+ * @return a buffered stream, or {@code raw} itself if buffering is disabled
or the file is empty
+ */
+ public static InputStream createInputStream(final InputStream raw, final
long fileSize) {
+ final long bufferSize = Math.min(fileSize, bufferSizeMax);
+ return bufferSize <= 0 ? raw : new BufferedInputStream(raw, (int)
bufferSize);
+ }
+
+ /**
+ * Borrow a reusable buffer of at least {@code minSize} bytes. If the
thread's pool holds a large
+ * enough buffer, it is handed out (and removed from the pool, so concurrent
borrowers never see
+ * the same array); otherwise a new buffer is allocated.
+ */
+ static byte[] acquireBuffer(final int minSize) {
+ final SoftReference<byte[]> reference = WRITE_BUFFER_POOL.get();
+ final byte[] cached = reference == null ? null : reference.get();
+ if (cached != null && cached.length >= minSize) {
Review Comment:
[P3] Do not reuse a write buffer larger than the current cap. If this thread
cached a 4 MB buffer and `setBufferSizeMax(8192)` is called later, this
condition returns the 4 MB array to the next stream. That contradicts both the
documented next-stream behavior and the central guarantee that the backing
buffer is at most `bufferSizeMax`. Please reuse only a buffer compatible with
the current configured size, or discard/clear oversized cached arrays when the
cap changes, and cover the decrease-after-release case in a test.
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/snapshot/ByteBufferInputStream.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.commons.snapshot;
+
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.Objects;
+
+/**
+ * An {@link InputStream} over a {@link ByteBuffer} (typically a direct buffer
filled from a
+ * snapshot file). Unlike {@link java.io.ByteArrayInputStream}, the underlying
bytes live outside
+ * the heap, so a snapshot file can be parsed without first copying it into
heap memory. {@code
+ * mark}/{@code reset} behave like {@link java.io.ByteArrayInputStream}: they
are position-based and
+ * independent of the read limit.
+ */
+public final class ByteBufferInputStream extends InputStream {
Review Comment:
[nit] Remove this unused snapshot stream utility. At the current PR head, no
production source imports
`org.apache.iotdb.commons.snapshot.ByteBufferInputStream`; NodeInfo and the
template paths now use FileChannel/InputStream streaming. This class and its
dedicated tests appear to be remnants of the earlier direct-buffer
implementation and add public API surface without a caller. Please remove them
unless this PR also adds a production use.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
