CRZbulabula commented on code in PR #18580:
URL: https://github.com/apache/iotdb/pull/18580#discussion_r3931555233
##########
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:
Thanks for the thorough analysis. This PR has been re-scoped to only the
PartitionInfo snapshot buffering, so the template snapshot changes this comment
refers to (Template.deserialize(InputStream) and the CNPhysicalPlanGenerator
streaming path) have been reverted and are no longer part of this PR. The
vector-schema concern therefore no longer applies here. Resolving.
##########
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:
Fixed. acquireBuffer now reuses a pooled buffer only when it also fits the
current bufferSizeMax (cached.length <= cap), so a buffer released under a
larger cap is never handed out again after the cap is lowered. Added
testBufferNotReusedAfterCapDecrease covering the decrease-after-release case,
and testWriteBufferReuse now pins an explicit 128KB cap. Resolving.
##########
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:
Removed. ByteBufferInputStream and its tests were dropped together with the
earlier direct-buffer implementation when the PR was re-scoped to PartitionInfo
only, so there is no unused public API surface. No longer applies. Resolving.
##########
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:
Removed along with ByteBufferInputStream itself (see the other thread). No
longer applies. Resolving.
--
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]
