This is an automated email from the ASF dual-hosted git repository. jackietien pushed a commit to branch object_type in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit aaa3bec02ef0ea87c6119ef3d4703b18a5af3d79 Author: JackieTien97 <[email protected]> AuthorDate: Mon Jul 7 18:20:33 2025 +0800 Add Object Read Example --- .../java/org/apache/iotdb/ObjectReadExample.java | 80 ++++++++++++++++++++++ .../java/org/apache/iotdb/rpc/TSStatusCode.java | 3 + .../apache/iotdb/db/conf/DataNodeMemoryConfig.java | 5 +- .../execution/memory/LocalMemoryManager.java | 1 - .../unary/scalar/ReadObjectColumnTransformer.java | 13 ++-- .../db/storageengine/rescon/disk/TierManager.java | 11 +++ .../commons/exception/ObjectFileNotExist.java} | 25 ++----- 7 files changed, 111 insertions(+), 27 deletions(-) diff --git a/example/session/src/main/java/org/apache/iotdb/ObjectReadExample.java b/example/session/src/main/java/org/apache/iotdb/ObjectReadExample.java new file mode 100644 index 00000000000..4fd68322fe0 --- /dev/null +++ b/example/session/src/main/java/org/apache/iotdb/ObjectReadExample.java @@ -0,0 +1,80 @@ +/* + * 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; + +import org.apache.iotdb.isession.ITableSession; +import org.apache.iotdb.isession.SessionDataSet; +import org.apache.iotdb.rpc.IoTDBConnectionException; +import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.session.TableSessionBuilder; + +import org.apache.commons.codec.digest.DigestUtils; +import org.apache.tsfile.utils.Binary; + +import java.util.Collections; + +public class ObjectReadExample { + private static final String LOCAL_URL = "127.0.0.1:6667"; + + public static void main(String[] args) { + + // don't specify database in constructor + try (ITableSession session = + new TableSessionBuilder() + .nodeUrls(Collections.singletonList(LOCAL_URL)) + .username("root") + .password("root") + .database("test1") + .thriftMaxFrameSize(256 * 1024 * 1024) + .build()) { + try (SessionDataSet dataSet = + session.executeQueryStatement("select READ_OBJECT(file) from test1 where time = 1")) { + SessionDataSet.DataIterator iterator = dataSet.iterator(); + while (iterator.next()) { + Binary binary = iterator.getBlob(1); + System.out.println(DigestUtils.md5Hex(binary.getValues())); + } + } + + try (SessionDataSet dataSet = + session.executeQueryStatement("select READ_OBJECT(file) from test1 where time = 2")) { + SessionDataSet.DataIterator iterator = dataSet.iterator(); + while (iterator.next()) { + Binary binary = iterator.getBlob(1); + System.out.println(DigestUtils.md5Hex(binary.getValues())); + } + } + + try (SessionDataSet dataSet = + session.executeQueryStatement("select READ_OBJECT(file) from test1")) { + SessionDataSet.DataIterator iterator = dataSet.iterator(); + while (iterator.next()) { + Binary binary = iterator.getBlob(1); + System.out.println(DigestUtils.md5Hex(binary.getValues())); + } + } + + } catch (IoTDBConnectionException e) { + e.printStackTrace(); + } catch (StatementExecutionException e) { + e.printStackTrace(); + } + } +} diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java index d64d761631c..a04a50115b3 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java @@ -145,6 +145,9 @@ public enum TSStatusCode { PLAN_FAILED_NETWORK_PARTITION(721), CANNOT_FETCH_FI_STATE(722), + // OBJECT + OBJECT_NOT_EXISTS(740), + // Arithmetic NUMERIC_VALUE_OUT_OF_RANGE(750), DIVISION_BY_ZERO(751), diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java index efb6758066f..551d8a9031d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java @@ -60,8 +60,7 @@ public class DataNodeMemoryConfig { private int queryThreadCount = Runtime.getRuntime().availableProcessors(); /** Max bytes of each FragmentInstance for DataExchange */ - private long maxBytesPerFragmentInstance = - Runtime.getRuntime().maxMemory() * 3 / 10 * 200 / 1001 / queryThreadCount; + private long maxBytesPerFragmentInstance = Runtime.getRuntime().maxMemory() * 3 / 10 * 200 / 1001; /** The memory manager of on heap */ private MemoryManager onHeapMemoryManager; @@ -483,7 +482,7 @@ public class DataNodeMemoryConfig { operatorsMemorySize += partForOperators; } // set max bytes per fragment instance - setMaxBytesPerFragmentInstance(dataExchangeMemorySize / getQueryThreadCount()); + setMaxBytesPerFragmentInstance(dataExchangeMemorySize); bloomFilterCacheMemoryManager = queryEngineMemoryManager.getOrCreateMemoryManager( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/memory/LocalMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/memory/LocalMemoryManager.java index 03766ab5276..b6ce8b52110 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/memory/LocalMemoryManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/memory/LocalMemoryManager.java @@ -30,7 +30,6 @@ public class LocalMemoryManager { private final MemoryPool queryPool; public LocalMemoryManager() { - // TODO @spricoder: why this pool is only used for query data exchange queryPool = new MemoryPool( "read", diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/unary/scalar/ReadObjectColumnTransformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/unary/scalar/ReadObjectColumnTransformer.java index 6332ab43b3b..9aa4187f3d6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/unary/scalar/ReadObjectColumnTransformer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/unary/scalar/ReadObjectColumnTransformer.java @@ -23,10 +23,12 @@ import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext; import org.apache.iotdb.db.queryengine.transformation.dag.column.ColumnTransformer; import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.UnaryColumnTransformer; +import org.apache.iotdb.db.storageengine.rescon.disk.TierManager; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.block.column.Column; import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.read.common.type.Type; import org.apache.tsfile.utils.Binary; @@ -40,6 +42,8 @@ import java.util.Optional; public class ReadObjectColumnTransformer extends UnaryColumnTransformer { + private static final TierManager TIER_MANAGER = TierManager.getInstance(); + private final Optional<FragmentInstanceContext> fragmentInstanceContext; private long offset = 0; private long length = -1; @@ -104,8 +108,7 @@ public class ReadObjectColumnTransformer extends UnaryColumnTransformer { } private Binary readObject(Binary binary) { - File file = new File(getObjectPathFromBinary(binary)); - // TODO: allocate memory + File file = getObjectPathFromBinary(binary); long fileSize = file.length(); if (offset >= fileSize) { throw new UnsupportedOperationException("offset is greater than object size"); @@ -126,7 +129,9 @@ public class ReadObjectColumnTransformer extends UnaryColumnTransformer { return new Binary(bytes); } - private String getObjectPathFromBinary(Binary binary) { - return binary.toString().split(",")[0]; + private File getObjectPathFromBinary(Binary binary) { + byte[] bytes = binary.getValues(); + return TIER_MANAGER.getObjectFile( + new String(bytes, 8, bytes.length - 8, TSFileConfig.STRING_CHARSET)); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java index 75c5d3b3c96..f3fcd2d3503 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.storageengine.rescon.disk; import org.apache.iotdb.commons.conf.IoTDBConstant; +import org.apache.iotdb.commons.exception.ObjectFileNotExist; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.DiskSpaceInsufficientException; @@ -323,6 +324,16 @@ public class TierManager { return tierDiskSpace; } + public File getObjectFile(String relativePath) { + for (String folder : objectDirs) { + File file = new File(folder, relativePath); + if (file.exists()) { + return file; + } + } + throw new ObjectFileNotExist(relativePath); + } + private enum DiskSpaceType { TOTAL, USABLE, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/memory/LocalMemoryManager.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/ObjectFileNotExist.java similarity index 52% copy from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/memory/LocalMemoryManager.java copy to iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/ObjectFileNotExist.java index 03766ab5276..05add08b218 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/memory/LocalMemoryManager.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/ObjectFileNotExist.java @@ -17,28 +17,15 @@ * under the License. */ -package org.apache.iotdb.db.queryengine.execution.memory; +package org.apache.iotdb.commons.exception; -import org.apache.iotdb.db.conf.IoTDBDescriptor; +import static org.apache.iotdb.rpc.TSStatusCode.OBJECT_NOT_EXISTS; -/** - * Manages memory of a data node. The memory is divided into two memory pools so that the memory for - * read and for write can be isolated. - */ -public class LocalMemoryManager { - - private final MemoryPool queryPool; +public class ObjectFileNotExist extends IoTDBRuntimeException { - public LocalMemoryManager() { - // TODO @spricoder: why this pool is only used for query data exchange - queryPool = - new MemoryPool( - "read", - IoTDBDescriptor.getInstance().getMemoryConfig().getDataExchangeMemoryManager(), - IoTDBDescriptor.getInstance().getMemoryConfig().getMaxBytesPerFragmentInstance()); - } + private static final String ERROR_MSG = "Object file %s does not exist"; - public MemoryPool getQueryPool() { - return queryPool; + public ObjectFileNotExist(String relativeObjectPath) { + super(String.format(ERROR_MSG, relativeObjectPath), OBJECT_NOT_EXISTS.getStatusCode()); } }
