You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by xi...@apache.org on 2023/05/29 09:23:09 UTC
[iotdb] branch tiered_storage_sync_code updated (340148a7684 -> 64dd4fe33bd)
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a change to branch tiered_storage_sync_code
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from 340148a7684 Merge branch 'master' into tiered_storage
add 0819828ee7a support hot modify data_dirs
add 8132effa340 Merge branch 'tiered_storage' of github.com:apache/iotdb into tiered_storage
new 64dd4fe33bd remove some files
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../iotdb/commons/concurrent/ThreadName.java | 2 -
.../apache/iotdb/commons/service/ServiceType.java | 1 -
object-storage/README.md | 21 -
object-storage/pom.xml | 192 ---------
.../iotdb/os/HybridFileInputFactoryDecorator.java | 56 ---
.../apache/iotdb/os/cache/CacheFileManager.java | 116 ------
.../apache/iotdb/os/cache/CacheRecoverTask.java | 89 ----
.../org/apache/iotdb/os/cache/OSFileCache.java | 104 -----
.../org/apache/iotdb/os/cache/OSFileCacheKey.java | 86 ----
.../apache/iotdb/os/cache/OSFileCacheValue.java | 133 ------
.../org/apache/iotdb/os/cache/OSFileChannel.java | 185 ---------
.../org/apache/iotdb/os/cache/OSInputStream.java | 96 -----
.../apache/iotdb/os/conf/ObjectStorageConfig.java | 117 ------
.../iotdb/os/conf/ObjectStorageDescriptor.java | 37 --
.../apache/iotdb/os/conf/provider/AWSS3Config.java | 22 -
.../iotdb/os/conf/provider/OSProviderConfig.java | 58 ---
.../apache/iotdb/os/conf/provider/TestConfig.java | 33 --
.../iotdb/os/exception/ObjectStorageException.java | 41 --
.../org/apache/iotdb/os/fileSystem/OSFile.java | 461 ---------------------
.../apache/iotdb/os/fileSystem/OSTsFileInput.java | 81 ----
.../apache/iotdb/os/fileSystem/OSTsFileOutput.java | 74 ----
.../java/org/apache/iotdb/os/fileSystem/OSURI.java | 78 ----
.../java/org/apache/iotdb/os/io/IMetaData.java | 25 --
.../apache/iotdb/os/io/ObjectStorageConnector.java | 50 ---
.../org/apache/iotdb/os/io/aws/S3MetaData.java | 42 --
.../iotdb/os/io/aws/S3ObjectStorageConnector.java | 248 -----------
.../os/io/test/TestObjectStorageConnector.java | 142 -------
.../iotdb/os/utils/ObjectStorageConstant.java | 25 --
.../apache/iotdb/os/utils/ObjectStorageType.java | 44 --
.../org/apache/iotdb/os/cache/OSFileCacheTest.java | 137 ------
.../apache/iotdb/os/cache/OSFileChannelTest.java | 219 ----------
.../apache/iotdb/os/cache/OSInputStreamTest.java | 181 --------
object-storage/src/test/resources/logback.xml | 41 --
pom.xml | 1 -
server/pom.xml | 5 -
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 108 +----
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 76 +---
.../iotdb/db/conf/directories/TierManager.java | 15 +-
.../execute/recover/CompactionRecoverManager.java | 1 -
.../execute/recover/CompactionRecoverTask.java | 1 -
.../db/engine/migration/LocalMigrationTask.java | 88 ----
.../iotdb/db/engine/migration/MigrationCause.java | 24 --
.../iotdb/db/engine/migration/MigrationTask.java | 109 -----
.../db/engine/migration/MigrationTaskManager.java | 211 ----------
.../db/engine/migration/RemoteMigrationTask.java | 76 ----
.../java/org/apache/iotdb/db/service/DataNode.java | 16 +-
.../iotdb/db/service/metrics/SystemMetrics.java | 1 -
47 files changed, 11 insertions(+), 3958 deletions(-)
delete mode 100644 object-storage/README.md
delete mode 100644 object-storage/pom.xml
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/HybridFileInputFactoryDecorator.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/cache/CacheFileManager.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/cache/CacheRecoverTask.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCache.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheKey.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheValue.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileChannel.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/cache/OSInputStream.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/conf/ObjectStorageConfig.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/conf/ObjectStorageDescriptor.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/conf/provider/AWSS3Config.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/conf/provider/OSProviderConfig.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/conf/provider/TestConfig.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/exception/ObjectStorageException.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSFile.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSTsFileInput.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSTsFileOutput.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSURI.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/io/IMetaData.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/io/ObjectStorageConnector.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/io/aws/S3MetaData.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/io/aws/S3ObjectStorageConnector.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/io/test/TestObjectStorageConnector.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/utils/ObjectStorageConstant.java
delete mode 100644 object-storage/src/main/java/org/apache/iotdb/os/utils/ObjectStorageType.java
delete mode 100644 object-storage/src/test/java/org/apache/iotdb/os/cache/OSFileCacheTest.java
delete mode 100644 object-storage/src/test/java/org/apache/iotdb/os/cache/OSFileChannelTest.java
delete mode 100644 object-storage/src/test/java/org/apache/iotdb/os/cache/OSInputStreamTest.java
delete mode 100644 object-storage/src/test/resources/logback.xml
delete mode 100644 server/src/main/java/org/apache/iotdb/db/engine/migration/LocalMigrationTask.java
delete mode 100644 server/src/main/java/org/apache/iotdb/db/engine/migration/MigrationCause.java
delete mode 100644 server/src/main/java/org/apache/iotdb/db/engine/migration/MigrationTask.java
delete mode 100644 server/src/main/java/org/apache/iotdb/db/engine/migration/MigrationTaskManager.java
delete mode 100644 server/src/main/java/org/apache/iotdb/db/engine/migration/RemoteMigrationTask.java
[iotdb] 01/01: remove some files
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch tiered_storage_sync_code
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 64dd4fe33bda5be3b3099bd13bb07e7574483ebf
Author: Jinrui.Zhang <xi...@gmail.com>
AuthorDate: Mon May 29 17:22:06 2023 +0800
remove some files
---
.../iotdb/commons/concurrent/ThreadName.java | 2 -
.../apache/iotdb/commons/service/ServiceType.java | 1 -
object-storage/README.md | 21 -
object-storage/pom.xml | 192 ---------
.../iotdb/os/HybridFileInputFactoryDecorator.java | 56 ---
.../apache/iotdb/os/cache/CacheFileManager.java | 116 ------
.../apache/iotdb/os/cache/CacheRecoverTask.java | 89 ----
.../org/apache/iotdb/os/cache/OSFileCache.java | 104 -----
.../org/apache/iotdb/os/cache/OSFileCacheKey.java | 86 ----
.../apache/iotdb/os/cache/OSFileCacheValue.java | 133 ------
.../org/apache/iotdb/os/cache/OSFileChannel.java | 185 ---------
.../org/apache/iotdb/os/cache/OSInputStream.java | 96 -----
.../apache/iotdb/os/conf/ObjectStorageConfig.java | 117 ------
.../iotdb/os/conf/ObjectStorageDescriptor.java | 37 --
.../apache/iotdb/os/conf/provider/AWSS3Config.java | 22 -
.../iotdb/os/conf/provider/OSProviderConfig.java | 58 ---
.../apache/iotdb/os/conf/provider/TestConfig.java | 33 --
.../iotdb/os/exception/ObjectStorageException.java | 41 --
.../org/apache/iotdb/os/fileSystem/OSFile.java | 461 ---------------------
.../apache/iotdb/os/fileSystem/OSTsFileInput.java | 81 ----
.../apache/iotdb/os/fileSystem/OSTsFileOutput.java | 74 ----
.../java/org/apache/iotdb/os/fileSystem/OSURI.java | 78 ----
.../java/org/apache/iotdb/os/io/IMetaData.java | 25 --
.../apache/iotdb/os/io/ObjectStorageConnector.java | 50 ---
.../org/apache/iotdb/os/io/aws/S3MetaData.java | 42 --
.../iotdb/os/io/aws/S3ObjectStorageConnector.java | 248 -----------
.../os/io/test/TestObjectStorageConnector.java | 142 -------
.../iotdb/os/utils/ObjectStorageConstant.java | 25 --
.../apache/iotdb/os/utils/ObjectStorageType.java | 44 --
.../org/apache/iotdb/os/cache/OSFileCacheTest.java | 137 ------
.../apache/iotdb/os/cache/OSFileChannelTest.java | 219 ----------
.../apache/iotdb/os/cache/OSInputStreamTest.java | 181 --------
object-storage/src/test/resources/logback.xml | 41 --
pom.xml | 1 -
server/pom.xml | 5 -
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 106 +----
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 68 ---
.../iotdb/db/conf/directories/TierManager.java | 13 -
.../execute/recover/CompactionRecoverManager.java | 1 -
.../execute/recover/CompactionRecoverTask.java | 1 -
.../db/engine/migration/LocalMigrationTask.java | 88 ----
.../iotdb/db/engine/migration/MigrationCause.java | 24 --
.../iotdb/db/engine/migration/MigrationTask.java | 109 -----
.../db/engine/migration/MigrationTaskManager.java | 211 ----------
.../db/engine/migration/RemoteMigrationTask.java | 76 ----
.../java/org/apache/iotdb/db/service/DataNode.java | 16 +-
.../iotdb/db/service/metrics/SystemMetrics.java | 1 -
47 files changed, 6 insertions(+), 3951 deletions(-)
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
index e19e62eee81..718c5a6bfc1 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
@@ -71,8 +71,6 @@ public enum ThreadName {
PIPE_PROCESSOR_EXECUTOR_POOL("Pipe-Processor-Executor-Pool"),
PIPE_CONNECTOR_EXECUTOR_POOL("Pipe-Connector-Executor-Pool"),
PIPE_SUBTASK_CALLBACK_EXECUTOR_POOL("Pipe-SubTask-Callback-Executor-Pool"),
- MIGRATION_SCHEDULER("Migration-Scheduler"),
- MIGRATION("Migration-Executor-Pool"),
PIPE_META_SYNC_SERVICE("Pipe-Meta-Sync-Service"),
PIPE_WAL_RESOURCE_TTL_CHECKER_SERVICE("Pipe-WAL-Resource-TTL-Checker-Service"),
;
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java b/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java
index d350305af87..7bde2af3102 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java
@@ -75,7 +75,6 @@ public enum ServiceType {
IOT_CONSENSUS_SERVICE("IoTConsensus Service", "IoTConsensusRPCService"),
PIPE_PLUGIN_CLASSLOADER_MANAGER_SERVICE(
"Pipe Plugin Classloader Manager Service", "PipePluginClassLoader"),
- MIGRATION_SERVICE("Migration Manager", "Migration Manager"),
PIPE_RUNTIME_AGENT("Pipe Runtime Agent", "PipeRuntimeAgent"),
MLNode_RPC_SERVICE("Rpc Service for MLNode", "MLNodeRPCService");
private final String name;
diff --git a/object-storage/README.md b/object-storage/README.md
deleted file mode 100644
index 2d6f52d4a0e..00000000000
--- a/object-storage/README.md
+++ /dev/null
@@ -1,21 +0,0 @@
-<!--
-
- 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.
-
--->
-# TsFile-ObjectStorage-Connector
\ No newline at end of file
diff --git a/object-storage/pom.xml b/object-storage/pom.xml
deleted file mode 100644
index c2906dce676..00000000000
--- a/object-storage/pom.xml
+++ /dev/null
@@ -1,192 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-
- 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.
-
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
- <parent>
- <groupId>org.apache.iotdb</groupId>
- <artifactId>iotdb-parent</artifactId>
- <version>1.2.0-SNAPSHOT</version>
- <relativePath>../pom.xml</relativePath>
- </parent>
- <artifactId>os-tsfile</artifactId>
- <packaging>jar</packaging>
- <name>IoTDB ObjectStorage-Tsfile</name>
- <url>http://maven.apache.org</url>
- <properties>
- <os.test.skip>false</os.test.skip>
- <os.it.skip>${os.test.skip}</os.it.skip>
- <os.ut.skip>${os.test.skip}</os.ut.skip>
- </properties>
- <dependencyManagement>
- <dependencies>
- <dependency>
- <groupId>software.amazon.awssdk</groupId>
- <artifactId>bom</artifactId>
- <version>2.20.1</version>
- <type>pom</type>
- <scope>import</scope>
- </dependency>
- </dependencies>
- </dependencyManagement>
- <dependencies>
- <dependency>
- <groupId>org.apache.iotdb</groupId>
- <artifactId>tsfile</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>software.amazon.awssdk</groupId>
- <artifactId>s3</artifactId>
- </dependency>
- <dependency>
- <groupId>com.github.ben-manes.caffeine</groupId>
- <artifactId>caffeine</artifactId>
- </dependency>
- <dependency>
- <groupId>org.mockito</groupId>
- <artifactId>mockito-core</artifactId>
- <scope>test</scope>
- </dependency>
- </dependencies>
- <build>
- <plugins>
- <!--
- Generate an OSGI compatible MANIFEST file.
- -->
- <plugin>
- <groupId>org.apache.felix</groupId>
- <artifactId>maven-bundle-plugin</artifactId>
- <version>${felix.version}</version>
- <executions>
- <execution>
- <id>bundle-manifest</id>
- <phase>process-classes</phase>
- <goals>
- <goal>manifest</goal>
- </goals>
- </execution>
- </executions>
- <configuration>
- <exportScr>true</exportScr>
- <instructions>
- <_include>-bnd.bnd</_include>
- <_removeheaders>Bnd-LastModified,Built-By</_removeheaders>
- <Embed-Dependency>dependencies</Embed-Dependency>
- </instructions>
- </configuration>
- </plugin>
- <!--
- Use the MANIFEST file generated by the maven-bundle-plugin.
- -->
- <plugin>
- <artifactId>maven-jar-plugin</artifactId>
- <configuration>
- <archive>
- <manifestFile>${project.build.outputDirectory}/META-INF/MANIFEST.MF</manifestFile>
- </archive>
- </configuration>
- </plugin>
- <!--using `mvn test` to run UT, `mvn verify` to run ITs
- Reference: https://antoniogoncalves.org/2012/12/13/lets-turn-integration-tests-with-maven-to-a-first-class-citizen/-->
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-surefire-plugin</artifactId>
- <configuration>
- <skipTests>${os.ut.skip}</skipTests>
- </configuration>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-failsafe-plugin</artifactId>
- <executions>
- <execution>
- <id>run-integration-tests</id>
- <phase>integration-test</phase>
- <goals>
- <goal>integration-test</goal>
- <goal>verify</goal>
- </goals>
- </execution>
- </executions>
- <configuration>
- <skipTests>${os.test.skip}</skipTests>
- <skipITs>${os.it.skip}</skipITs>
- </configuration>
- </plugin>
- </plugins>
- </build>
- <profiles>
- <profile>
- <id>skipObjectStorageTests</id>
- <activation>
- <property>
- <name>skipTests</name>
- <value>true</value>
- </property>
- </activation>
- <properties>
- <os.test.skip>true</os.test.skip>
- <os.ut.skip>true</os.ut.skip>
- <os.it.skip>true</os.it.skip>
- </properties>
- </profile>
- <profile>
- <id>skipUT_ObjectStorageTests</id>
- <activation>
- <property>
- <name>skipUTs</name>
- <value>true</value>
- </property>
- </activation>
- <properties>
- <os.ut.skip>true</os.ut.skip>
- </properties>
- </profile>
- <profile>
- <id>get-jar-with-dependencies</id>
- <build>
- <plugins>
- <plugin>
- <artifactId>maven-assembly-plugin</artifactId>
- <version>${maven.assembly.version}</version>
- <configuration>
- <descriptorRefs>
- <descriptorRef>jar-with-dependencies</descriptorRef>
- </descriptorRefs>
- </configuration>
- <executions>
- <execution>
- <id>make-assembly</id>
- <!-- this is used for inheritance merges -->
- <phase>package</phase>
- <!-- bind to the packaging phase -->
- <goals>
- <goal>single</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
- </profile>
- </profiles>
-</project>
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/HybridFileInputFactoryDecorator.java b/object-storage/src/main/java/org/apache/iotdb/os/HybridFileInputFactoryDecorator.java
deleted file mode 100644
index 975fa86f986..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/HybridFileInputFactoryDecorator.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * 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.os;
-
-import org.apache.iotdb.os.conf.ObjectStorageConfig;
-import org.apache.iotdb.os.conf.ObjectStorageDescriptor;
-import org.apache.iotdb.tsfile.fileSystem.fileInputFactory.FileInputFactory;
-import org.apache.iotdb.tsfile.fileSystem.fileInputFactory.HybridFileInputFactory;
-import org.apache.iotdb.tsfile.read.reader.TsFileInput;
-import org.apache.iotdb.tsfile.utils.FSUtils;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
-import java.io.IOException;
-
-public class HybridFileInputFactoryDecorator implements FileInputFactory {
- private static final Logger logger =
- LoggerFactory.getLogger(HybridFileInputFactoryDecorator.class);
- private static final ObjectStorageConfig config =
- ObjectStorageDescriptor.getInstance().getConfig();
- private static final FileInputFactory fileInputFactory = new HybridFileInputFactory();
-
- private int dataNodeId;
-
- public HybridFileInputFactoryDecorator(int dataNodeId) {
- this.dataNodeId = dataNodeId;
- }
-
- @Override
- public TsFileInput getTsFileInput(String filePath) throws IOException {
- File file = new File(filePath);
- if (!file.exists()) {
- return fileInputFactory.getTsFileInput(
- FSUtils.parseLocalTsFile2OSFile(file, config.getBucketName(), dataNodeId).getPath());
- }
- return fileInputFactory.getTsFileInput(filePath);
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/cache/CacheFileManager.java b/object-storage/src/main/java/org/apache/iotdb/os/cache/CacheFileManager.java
deleted file mode 100644
index e2910f05226..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/cache/CacheFileManager.java
+++ /dev/null
@@ -1,116 +0,0 @@
-/*
- * 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.os.cache;
-
-import org.apache.iotdb.os.conf.ObjectStorageConfig;
-import org.apache.iotdb.os.conf.ObjectStorageDescriptor;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.nio.channels.FileChannel;
-import java.nio.file.StandardOpenOption;
-import java.util.concurrent.atomic.AtomicLong;
-
-import static org.apache.iotdb.os.utils.ObjectStorageConstant.CACHE_FILE_SUFFIX;
-import static org.apache.iotdb.os.utils.ObjectStorageConstant.TMP_CACHE_FILE_SUFFIX;
-
-/** This class manages all write operations to the cache files */
-public class CacheFileManager {
- private static final Logger logger = LoggerFactory.getLogger(CacheFileManager.class);
- private static final ObjectStorageConfig config =
- ObjectStorageDescriptor.getInstance().getConfig();
- private final AtomicLong cacheFileId = new AtomicLong(0);
-
- private CacheFileManager() {
- for (String cacheDir : config.getCacheDirs()) {
- File cacheDirFile = new File(cacheDir);
- if (!cacheDirFile.exists()) {
- cacheDirFile.mkdirs();
- }
- }
- }
-
- private long getNextCacheFileId() {
- return cacheFileId.getAndIncrement();
- }
-
- private File getTmpCacheFile(long id) {
- long dirId = id % config.getCacheDirs().length;
- return new File(config.getCacheDirs()[(int) dirId], id + TMP_CACHE_FILE_SUFFIX);
- }
-
- private File getCacheFile(long id) {
- long dirId = id % config.getCacheDirs().length;
- return new File(config.getCacheDirs()[(int) dirId], id + CACHE_FILE_SUFFIX);
- }
-
- /** Persist data, return null when failing to persist data */
- public OSFileCacheValue persist(OSFileCacheKey key, byte[] data) {
- long cacheFileId;
- File tmpCacheFile;
- // create new tmp cache file
- try {
- do {
- cacheFileId = getNextCacheFileId();
- tmpCacheFile = getTmpCacheFile(cacheFileId);
- } while (!tmpCacheFile.createNewFile());
- } catch (IOException e) {
- logger.error("Fail to create cache file.", e);
- return null;
- }
- // write value into tmp cache file
- try (FileChannel channel = FileChannel.open(tmpCacheFile.toPath(), StandardOpenOption.WRITE)) {
- ByteBuffer meta = key.serialize();
- meta.flip();
- channel.write(meta);
- channel.write(ByteBuffer.wrap(data));
- } catch (IOException e) {
- logger.error("Fail to persist data to cache file {}.", tmpCacheFile, e);
- tmpCacheFile.delete();
- return null;
- }
- // rename tmp file to cache file
- File cacheFile = getCacheFile(cacheFileId);
- if (tmpCacheFile.renameTo(cacheFile)) {
- return new OSFileCacheValue(
- cacheFile, 0, key.serializeSize(), data.length, key.getStartPosition());
- } else {
- return null;
- }
- }
-
- /** This method is used by the recover procedure */
- void setCacheFileId(long startId) {
- cacheFileId.set(startId);
- }
-
- public static CacheFileManager getInstance() {
- return InstanceHolder.INSTANCE;
- }
-
- private static class InstanceHolder {
- private InstanceHolder() {}
-
- private static final CacheFileManager INSTANCE = new CacheFileManager();
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/cache/CacheRecoverTask.java b/object-storage/src/main/java/org/apache/iotdb/os/cache/CacheRecoverTask.java
deleted file mode 100644
index 08f505bd89c..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/cache/CacheRecoverTask.java
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
- * 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.os.cache;
-
-import org.apache.iotdb.os.conf.ObjectStorageConfig;
-import org.apache.iotdb.os.conf.ObjectStorageDescriptor;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
-import java.io.InputStream;
-import java.nio.file.Files;
-
-import static org.apache.iotdb.os.utils.ObjectStorageConstant.CACHE_FILE_SUFFIX;
-
-public class CacheRecoverTask implements Runnable {
- private static final Logger logger = LoggerFactory.getLogger(CacheRecoverTask.class);
- private static final ObjectStorageConfig config =
- ObjectStorageDescriptor.getInstance().getConfig();
- private static final OSFileCache cache = OSFileCache.getInstance();
-
- @Override
- public void run() {
- long maxCacheFileId = 0;
- for (String cacheDir : config.getCacheDirs()) {
- File cacheDirFile = new File(cacheDir);
- if (!cacheDirFile.exists()) {
- continue;
- }
- File[] cacheFiles = cacheDirFile.listFiles();
- if (cacheFiles == null) {
- continue;
- }
- for (File cacheFile : cacheFiles) {
- try {
- if (cacheFile.isDirectory()) {
- continue;
- }
- String cacheFileName = cacheFile.getName();
- if (!cacheFileName.endsWith(CACHE_FILE_SUFFIX)) {
- cacheFile.delete();
- continue;
- }
- // read meta and put it back to the cache
- try (InputStream in = Files.newInputStream(cacheFile.toPath())) {
- OSFileCacheKey key = OSFileCacheKey.deserialize(in);
- int metaSize = key.serializeSize();
- int dataSize = (int) cacheFile.length() - metaSize;
- if (dataSize != config.getCachePageSize()) {
- logger.debug(
- "Cache file {}'s data size doesn't match the cache page size, so delete it.",
- cacheFile);
- cacheFile.delete();
- continue;
- }
- OSFileCacheValue value =
- new OSFileCacheValue(cacheFile, 0, metaSize, dataSize, key.getStartPosition());
- cache.put(key, value);
- }
- // update max cache file id
- String cacheFileIdStr = cacheFileName.substring(0, cacheFileName.indexOf('.'));
- long cacheFileId = Long.parseLong(cacheFileIdStr);
- if (cacheFileId > maxCacheFileId) {
- maxCacheFileId = cacheFileId;
- }
- } catch (Exception e) {
- logger.error("Failed to recover cache file {}", cacheFile.getName(), e);
- }
- }
- }
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCache.java b/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCache.java
deleted file mode 100644
index a43c28d8016..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCache.java
+++ /dev/null
@@ -1,104 +0,0 @@
-/*
- * 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.os.cache;
-
-import org.apache.iotdb.os.conf.ObjectStorageConfig;
-import org.apache.iotdb.os.conf.ObjectStorageDescriptor;
-import org.apache.iotdb.os.io.ObjectStorageConnector;
-import org.apache.iotdb.os.utils.ObjectStorageType;
-
-import com.github.benmanes.caffeine.cache.CacheLoader;
-import com.github.benmanes.caffeine.cache.Caffeine;
-import com.github.benmanes.caffeine.cache.LoadingCache;
-import com.github.benmanes.caffeine.cache.Weigher;
-import org.checkerframework.checker.nullness.qual.NonNull;
-import org.checkerframework.checker.nullness.qual.Nullable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class OSFileCache {
- private static final Logger logger = LoggerFactory.getLogger(OSFileCache.class);
- private static final ObjectStorageConfig config =
- ObjectStorageDescriptor.getInstance().getConfig();
- private ObjectStorageConnector connector;
-
- /** manage all io operations to the cache files */
- private final CacheFileManager cacheFileManager = CacheFileManager.getInstance();
- /**
- * persistent LRU cache for remote TsFile, value is loaded successfully when it has been stored on
- * the disk
- */
- private final LoadingCache<OSFileCacheKey, OSFileCacheValue> remotePos2LocalCacheFile;
-
- OSFileCache(ObjectStorageType osType) {
- connector = ObjectStorageType.getConnector(osType);
- remotePos2LocalCacheFile =
- Caffeine.newBuilder()
- .maximumWeight(config.getCacheMaxDiskUsage())
- .weigher((Weigher<OSFileCacheKey, OSFileCacheValue>) (key, value) -> value.getLength())
- .removalListener(
- (key, value, cause) -> {
- if (value != null) {
- value.setShouldDelete();
- }
- })
- .build(new OSFileCacheLoader());
- }
-
- public OSFileCacheValue get(OSFileCacheKey key) {
- OSFileCacheValue value = remotePos2LocalCacheFile.get(key);
- // TODO try to simplify the logic here
- if (!value.getCacheFile().exists()) {
- logger.info("want {} but file deleted", key);
- remotePos2LocalCacheFile.invalidate(key);
- }
- return value;
- }
-
- /** This method is used by the recover procedure */
- void put(OSFileCacheKey key, OSFileCacheValue value) {
- remotePos2LocalCacheFile.put(key, value);
- }
-
- // test only
- void setConnector(ObjectStorageConnector connector) {
- this.connector = connector;
- }
-
- class OSFileCacheLoader implements CacheLoader<OSFileCacheKey, OSFileCacheValue> {
- @Override
- public @Nullable OSFileCacheValue load(@NonNull OSFileCacheKey key) throws Exception {
- byte[] data =
- connector.getRemoteObject(
- key.getFile().toOSURI(), key.getStartPosition(), config.getCachePageSize());
- return cacheFileManager.persist(key, data);
- }
- }
-
- public static OSFileCache getInstance() {
- return OSFileCache.InstanceHolder.INSTANCE;
- }
-
- private static class InstanceHolder {
- private InstanceHolder() {}
-
- private static final OSFileCache INSTANCE = new OSFileCache(config.getOsType());
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheKey.java b/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheKey.java
deleted file mode 100644
index 4bd6f47f085..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheKey.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * 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.os.cache;
-
-import org.apache.iotdb.os.fileSystem.OSFile;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.Serializable;
-import java.nio.ByteBuffer;
-import java.util.Objects;
-
-public class OSFileCacheKey implements Serializable {
- /** remote TsFile */
- private final OSFile file;
- /** start position in the remote TsFile */
- private final long startPosition;
-
- public OSFileCacheKey(OSFile file, long startPosition) {
- this.file = file;
- this.startPosition = startPosition;
- }
-
- public int serializeSize() {
- return Integer.BYTES + file.toString().getBytes().length + Long.BYTES;
- }
-
- public ByteBuffer serialize() {
- ByteBuffer buffer = ByteBuffer.allocate(serializeSize());
- ReadWriteIOUtils.write(file.toString(), buffer);
- ReadWriteIOUtils.write(startPosition, buffer);
- return buffer;
- }
-
- public static OSFileCacheKey deserialize(InputStream inputStream) throws IOException {
- String filePath = ReadWriteIOUtils.readString(inputStream);
- long startPosition = ReadWriteIOUtils.readLong(inputStream);
- return new OSFileCacheKey(new OSFile(filePath), startPosition);
- }
-
- public OSFile getFile() {
- return file;
- }
-
- public long getStartPosition() {
- return startPosition;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(file, startPosition);
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj) {
- return true;
- }
- if (obj == null || getClass() != obj.getClass()) {
- return false;
- }
- OSFileCacheKey that = (OSFileCacheKey) obj;
- return file.equals(that.file) && startPosition == that.startPosition;
- }
-
- public String toString() {
- return file.getName() + "," + startPosition;
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheValue.java b/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheValue.java
deleted file mode 100644
index ab9e4718909..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheValue.java
+++ /dev/null
@@ -1,133 +0,0 @@
-/*
- * 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.os.cache;
-
-import java.io.File;
-
-public class OSFileCacheValue {
- /** local cache file */
- private File cacheFile;
- // 如果每个块用一个文件来存储，则该值一直为 0
- // 如果使用一个大文件存储所有块，则该值为大文件中的起点
- /** start position in the local cache file */
- private long startPositionInCacheFile;
- /** cache data size */
- private int dataSize;
- /** cache key size */
- private int metaSize;
- /** start position in the remote TsFile */
- private long startPositionInOSFile;
- /** start position in the remote TsFile */
- private long endPositionInOSFile;
-
- private boolean shouldDelete;
- private int readCnt;
-
- public OSFileCacheValue(
- File cacheFile,
- long startPositionInCacheFile,
- int metaSize,
- int dataSize,
- long startPositionInOSFile) {
- this.cacheFile = cacheFile;
- this.startPositionInCacheFile = startPositionInCacheFile;
- this.metaSize = metaSize;
- this.dataSize = dataSize;
- this.startPositionInOSFile = startPositionInOSFile;
- this.endPositionInOSFile = startPositionInOSFile + dataSize;
- }
-
- public File getCacheFile() {
- return cacheFile;
- }
-
- public long getStartPositionInCacheFile() {
- return startPositionInCacheFile;
- }
-
- public int getMetaSize() {
- return metaSize;
- }
-
- public int getDataSize() {
- return dataSize;
- }
-
- // 如果每个块用一个文件来存储，则该值一直为该文件的大小
- // 如果使用一个大文件存储所有块，则该值为该块的实际长度
- public int getLength() {
- return metaSize + dataSize;
- }
-
- public long getStartPositionInOSFile() {
- return startPositionInOSFile;
- }
-
- public long getEndPositionInOSFile() {
- return endPositionInOSFile;
- }
-
- public long getEndPositionInCacheFile() {
- return startPositionInCacheFile + getLength();
- }
-
- /**
- * Convert position in the TsFile to the corresponding position in the cache file. Return -1 when
- * the position is outside the cache file range
- */
- public long convertTsFilePos2CachePos(long positionInTsFile) {
- if (positionInTsFile < startPositionInOSFile || positionInTsFile >= endPositionInOSFile) {
- return -1;
- }
- return startPositionInCacheFile + metaSize + (positionInTsFile - startPositionInOSFile);
- }
-
- /** Mark this value should be deleted, delete this value when no one is reading it. */
- public synchronized void setShouldDelete() {
- this.shouldDelete = true;
- if (readCnt == 0) {
- cacheFile.delete();
- }
- }
-
- /**
- * Try to get the read lock, return false when this cache value should be deleted or has been
- * deleted.
- */
- public synchronized boolean tryReadLock() {
- if (shouldDelete || !cacheFile.exists()) {
- return false;
- } else {
- this.readCnt++;
- return true;
- }
- }
-
- /**
- * Release the read lock, delete the cache value when no one else is reading it and this cache
- * value should be deleted.
- */
- public synchronized void readUnlock() {
- this.readCnt--;
- if (shouldDelete && readCnt == 0) {
- cacheFile.delete();
- }
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileChannel.java b/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileChannel.java
deleted file mode 100644
index db0649c3374..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileChannel.java
+++ /dev/null
@@ -1,185 +0,0 @@
-/*
- * 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.os.cache;
-
-import org.apache.iotdb.os.conf.ObjectStorageConfig;
-import org.apache.iotdb.os.conf.ObjectStorageDescriptor;
-import org.apache.iotdb.os.fileSystem.OSFile;
-import org.apache.iotdb.os.fileSystem.OSTsFileInput;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.ByteBuffer;
-import java.nio.channels.FileChannel;
-import java.nio.file.StandardOpenOption;
-
-public class OSFileChannel implements Closeable {
- private static final Logger logger = LoggerFactory.getLogger(OSTsFileInput.class);
- private static final ObjectStorageConfig config =
- ObjectStorageDescriptor.getInstance().getConfig();
- private final OSFileCache cache;
- private final OSFile osFile;
- private long position = 0;
-
- private OSFileBlock currentOSFileBlock;
-
- public OSFileChannel(OSFile osFile) throws IOException {
- this(osFile, OSFileCache.getInstance());
- }
-
- OSFileChannel(OSFile osFile, OSFileCache cache) throws IOException {
- this.cache = cache;
- this.osFile = osFile;
- }
-
- public static InputStream newInputStream(OSFileChannel channel) {
- return new OSInputStream(channel);
- }
-
- private void openNextCacheFile(int position) throws IOException {
- // close prev cache file
- closeCurrentOSFileBlock();
- // open next cache file
- OSFileCacheKey key = locateCacheFileFromPosition(position);
- currentOSFileBlock = new OSFileBlock(key);
- }
-
- private OSFileCacheKey locateCacheFileFromPosition(int position) throws IOException {
- if (position >= size()) {
- throw new IOException("EOF");
- }
- long startPosition = position - position % config.getCachePageSize();
- return new OSFileCacheKey(osFile, startPosition);
- }
-
- public long size() {
- return osFile.length();
- }
-
- public long position() {
- return position;
- }
-
- public synchronized void position(long newPosition) {
- if (newPosition < 0) {
- throw new IllegalArgumentException();
- }
- position = newPosition;
- }
-
- public synchronized int read(ByteBuffer dst) throws IOException {
- int readSize = read(dst, position);
- position(position + readSize);
- return readSize;
- }
-
- public synchronized int read(ByteBuffer dst, long position) throws IOException {
- int currentPosition = (int) position;
- dst.mark();
- int dstLimit = dst.limit();
- // determiner the ead range
- long startPos = position;
- long endPos = position + dst.remaining();
- if (startPos >= size()) {
- return -1;
- }
- if (endPos > size()) {
- endPos = size();
- }
- // read each cache file
- int totalReadBytes = 0;
- try {
- while (startPos < endPos) {
- if (currentOSFileBlock == null || !currentOSFileBlock.canRead(startPos)) {
- openNextCacheFile(currentPosition);
- }
- int readSize = currentOSFileBlock.read(dst, startPos, endPos);
- totalReadBytes += readSize;
- startPos += readSize;
- currentPosition += readSize;
- }
- } catch (IOException e) {
- dst.reset();
- throw e;
- } finally {
- dst.limit(dstLimit);
- }
- return totalReadBytes;
- }
-
- private void closeCurrentOSFileBlock() throws IOException {
- if (currentOSFileBlock != null) {
- currentOSFileBlock.close();
- }
- }
-
- @Override
- public void close() throws IOException {
- closeCurrentOSFileBlock();
- }
-
- private class OSFileBlock {
- private OSFileCacheValue cacheValue;
- private FileChannel fileChannel;
-
- public OSFileBlock(OSFileCacheKey cacheKey) throws IOException {
- do {
- cacheValue = cache.get(cacheKey);
- } while (!cacheValue.tryReadLock());
- fileChannel = FileChannel.open(cacheValue.getCacheFile().toPath(), StandardOpenOption.READ);
- }
-
- public boolean canRead(long positionInOSFile) {
- return cacheValue.getStartPositionInOSFile() <= positionInOSFile
- && positionInOSFile < cacheValue.getEndPositionInOSFile();
- }
-
- public int read(ByteBuffer dst, long startPos, long endPos) throws IOException {
- long readStartPosition = cacheValue.convertTsFilePos2CachePos(startPos);
- long expectedReadLength = endPos - startPos;
-
- int readSize =
- (int)
- Math.min(
- expectedReadLength, cacheValue.getEndPositionInCacheFile() - readStartPosition);
-
- dst.limit(dst.position() + readSize);
- long actualReadSize = fileChannel.read(dst, readStartPosition);
- if (actualReadSize != readSize) {
- throw new IOException(
- String.format(
- "Cache file %s may crash because cannot read enough information in the cash file.",
- cacheValue.getCacheFile()));
- }
- return readSize;
- }
-
- public void close() throws IOException {
- try {
- fileChannel.close();
- } finally {
- cacheValue.readUnlock();
- }
- }
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSInputStream.java b/object-storage/src/main/java/org/apache/iotdb/os/cache/OSInputStream.java
deleted file mode 100644
index 68166d0ce45..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSInputStream.java
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
- * 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.os.cache;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.ByteBuffer;
-
-public class OSInputStream extends InputStream {
- private final OSFileChannel channel;
-
- public OSInputStream(OSFileChannel channel) {
- super();
- this.channel = channel;
- }
-
- @Override
- public synchronized int read() throws IOException {
- byte[] b1 = new byte[1];
- int n = read(b1);
- if (n == 1) return b1[0] & 0xff;
- return -1;
- }
-
- @Override
- public synchronized int read(byte[] b) throws IOException {
- return read(b, 0, b.length);
- }
-
- @Override
- public synchronized int read(byte[] b, int off, int len) throws IOException {
- if (off < 0 || len < 0 || len > b.length - off) {
- throw new IndexOutOfBoundsException();
- }
- if (len == 0) {
- return 0;
- }
- ByteBuffer buffer = ByteBuffer.wrap(b);
- buffer.position(off);
- buffer.limit(off + len);
- return channel.read(buffer);
- }
-
- @Override
- public synchronized long skip(long n) throws IOException {
- if (n <= 0) {
- return 0;
- }
- if (n > channel.size() - channel.position()) {
- n = channel.size() - channel.position();
- }
- channel.position(channel.position() + n);
- return n;
- }
-
- @Override
- public synchronized int available() throws IOException {
- return (int) (channel.size() - channel.position());
- }
-
- @Override
- public synchronized void close() throws IOException {
- channel.close();
- }
-
- @Override
- public synchronized void mark(int readlimit) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public synchronized void reset() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public synchronized boolean markSupported() {
- throw new UnsupportedOperationException();
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/conf/ObjectStorageConfig.java b/object-storage/src/main/java/org/apache/iotdb/os/conf/ObjectStorageConfig.java
deleted file mode 100644
index b66b4871768..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/conf/ObjectStorageConfig.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
- * 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.os.conf;
-
-import org.apache.iotdb.os.conf.provider.AWSS3Config;
-import org.apache.iotdb.os.conf.provider.OSProviderConfig;
-import org.apache.iotdb.os.conf.provider.TestConfig;
-import org.apache.iotdb.os.utils.ObjectStorageType;
-
-import java.io.File;
-
-public class ObjectStorageConfig {
- private ObjectStorageType osType = ObjectStorageType.TEST;
-
- private OSProviderConfig providerConfig = new TestConfig();
-
- private String[] cacheDirs = {
- "data" + File.separator + "datanode" + File.separator + "data" + File.separator + "cache"
- };
-
- private long cacheMaxDiskUsage = 50 * 1024 * 1024L;
-
- private int cachePageSize = 10 * 1024 * 1024;
-
- ObjectStorageConfig() {}
-
- public ObjectStorageType getOsType() {
- return osType;
- }
-
- public void setOsType(ObjectStorageType osType) {
- this.osType = osType;
- switch (osType) {
- case AWS_S3:
- this.providerConfig = new AWSS3Config();
- break;
- default:
- this.providerConfig = new TestConfig();
- }
- }
-
- public OSProviderConfig getProviderConfig() {
- return providerConfig;
- }
-
- public String getEndpoint() {
- return providerConfig.getEndpoint();
- }
-
- public void setEndpoint(String endpoint) {
- providerConfig.setEndpoint(endpoint);
- }
-
- public String getBucketName() {
- return providerConfig.getBucketName();
- }
-
- public void setBucketName(String bucketName) {
- providerConfig.setBucketName(bucketName);
- }
-
- public String getAccessKeyId() {
- return providerConfig.getAccessKeyId();
- }
-
- public void setAccessKeyId(String accessKeyId) {
- providerConfig.setAccessKeyId(accessKeyId);
- }
-
- public String getAccessKeySecret() {
- return providerConfig.getAccessKeySecret();
- }
-
- public void setAccessKeySecret(String accessKeySecret) {
- providerConfig.setAccessKeySecret(accessKeySecret);
- }
-
- public String[] getCacheDirs() {
- return cacheDirs;
- }
-
- public void setCacheDirs(String[] cacheDirs) {
- this.cacheDirs = cacheDirs;
- }
-
- public long getCacheMaxDiskUsage() {
- return cacheMaxDiskUsage;
- }
-
- public void setCacheMaxDiskUsage(long cacheMaxDiskUsage) {
- this.cacheMaxDiskUsage = cacheMaxDiskUsage;
- }
-
- public int getCachePageSize() {
- return cachePageSize;
- }
-
- public void setCachePageSize(int cachePageSize) {
- this.cachePageSize = cachePageSize;
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/conf/ObjectStorageDescriptor.java b/object-storage/src/main/java/org/apache/iotdb/os/conf/ObjectStorageDescriptor.java
deleted file mode 100644
index 5c832c98409..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/conf/ObjectStorageDescriptor.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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.os.conf;
-
-public class ObjectStorageDescriptor {
- private final ObjectStorageConfig conf = new ObjectStorageConfig();
-
- public ObjectStorageConfig getConfig() {
- return conf;
- }
-
- public static ObjectStorageDescriptor getInstance() {
- return InstanceHolder.INSTANCE;
- }
-
- private static class InstanceHolder {
- private InstanceHolder() {}
-
- private static final ObjectStorageDescriptor INSTANCE = new ObjectStorageDescriptor();
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/conf/provider/AWSS3Config.java b/object-storage/src/main/java/org/apache/iotdb/os/conf/provider/AWSS3Config.java
deleted file mode 100644
index ff684bab7d0..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/conf/provider/AWSS3Config.java
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * 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.os.conf.provider;
-
-public class AWSS3Config extends OSProviderConfig {}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/conf/provider/OSProviderConfig.java b/object-storage/src/main/java/org/apache/iotdb/os/conf/provider/OSProviderConfig.java
deleted file mode 100644
index ca0fff2ec18..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/conf/provider/OSProviderConfig.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * 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.os.conf.provider;
-
-public abstract class OSProviderConfig {
- protected String endpoint;
- protected String bucketName = "iotdb_data";
- protected String accessKeyId;
- protected String accessKeySecret;
-
- public String getEndpoint() {
- return endpoint;
- }
-
- public void setEndpoint(String endpoint) {
- this.endpoint = endpoint;
- }
-
- public String getBucketName() {
- return bucketName;
- }
-
- public void setBucketName(String bucketName) {
- this.bucketName = bucketName;
- }
-
- public String getAccessKeyId() {
- return accessKeyId;
- }
-
- public void setAccessKeyId(String accessKeyId) {
- this.accessKeyId = accessKeyId;
- }
-
- public String getAccessKeySecret() {
- return accessKeySecret;
- }
-
- public void setAccessKeySecret(String accessKeySecret) {
- this.accessKeySecret = accessKeySecret;
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/conf/provider/TestConfig.java b/object-storage/src/main/java/org/apache/iotdb/os/conf/provider/TestConfig.java
deleted file mode 100644
index 558d01fbe8d..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/conf/provider/TestConfig.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * 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.os.conf.provider;
-
-import java.io.File;
-
-public class TestConfig extends OSProviderConfig {
- private String testDir = "data" + File.separator + "test_s3";
-
- public String getTestDir() {
- return testDir;
- }
-
- public void setTestDir(String testDir) {
- this.testDir = testDir;
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/exception/ObjectStorageException.java b/object-storage/src/main/java/org/apache/iotdb/os/exception/ObjectStorageException.java
deleted file mode 100644
index 2af0de14c05..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/exception/ObjectStorageException.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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.os.exception;
-
-public class ObjectStorageException extends Exception {
- public ObjectStorageException() {}
-
- public ObjectStorageException(String message) {
- super(message);
- }
-
- public ObjectStorageException(String message, Throwable cause) {
- super(message, cause);
- }
-
- public ObjectStorageException(Throwable cause) {
- super(cause);
- }
-
- public ObjectStorageException(
- String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
- super(message, cause, enableSuppression, writableStackTrace);
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSFile.java b/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSFile.java
deleted file mode 100644
index 96c9a4e7eb6..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSFile.java
+++ /dev/null
@@ -1,461 +0,0 @@
-/*
- * 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.os.fileSystem;
-
-import org.apache.iotdb.os.conf.ObjectStorageConfig;
-import org.apache.iotdb.os.conf.ObjectStorageDescriptor;
-import org.apache.iotdb.os.exception.ObjectStorageException;
-import org.apache.iotdb.os.io.ObjectStorageConnector;
-import org.apache.iotdb.os.utils.ObjectStorageType;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.BufferedInputStream;
-import java.io.BufferedOutputStream;
-import java.io.BufferedReader;
-import java.io.BufferedWriter;
-import java.io.File;
-import java.io.FileFilter;
-import java.io.FilenameFilter;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.net.MalformedURLException;
-import java.net.URI;
-import java.net.URL;
-import java.nio.file.Path;
-import java.util.Arrays;
-
-import static org.apache.iotdb.os.utils.ObjectStorageConstant.FILE_SEPARATOR;
-
-public class OSFile extends File {
- private static final Logger logger = LoggerFactory.getLogger(OSFile.class);
- private static final String UNSUPPORT_OPERATION =
- "Current object storage file doesn't support this operation.";
- private static final ObjectStorageConfig config =
- ObjectStorageDescriptor.getInstance().getConfig();
- private ObjectStorageConnector connector;
-
- private final OSURI osUri;
-
- private long length = 0L;
-
- public OSFile(OSURI osUri) {
- this(osUri, config.getOsType());
- }
-
- public OSFile(OSURI osUri, ObjectStorageType osType) {
- super(osUri.getURI().toString());
- this.osUri = osUri;
- connector = ObjectStorageType.getConnector(osType);
- }
-
- public OSFile(String pathname) {
- this(new OSURI(pathname));
- }
-
- public OSFile(String parent, String child) {
- this(new OSURI(concatPath(parent, child)));
- }
-
- public OSFile(File parent, String child) {
- this(new OSURI(concatPath(parent.toString(), child)));
- }
-
- public OSFile(URI uri) {
- this(new OSURI(uri));
- }
-
- private static String concatPath(String parent, String child) {
- if (parent.endsWith(FILE_SEPARATOR)) {
- return parent + child;
- } else {
- return parent + FILE_SEPARATOR + child;
- }
- }
-
- @Override
- public String getName() {
- return osUri.getKey();
- }
-
- @Override
- public String getParent() {
- File parent = getParentFile();
- return parent == null ? null : parent.toString();
- }
-
- @Override
- public File getParentFile() {
- int lastSeparatorIdx = osUri.getKey().lastIndexOf(FILE_SEPARATOR);
- if (lastSeparatorIdx <= 0) {
- return null;
- }
- return new OSFile(new OSURI(osUri.getBucket(), osUri.getKey().substring(0, lastSeparatorIdx)));
- }
-
- @Override
- public String getPath() {
- return osUri.toString();
- }
-
- @Override
- public boolean isAbsolute() {
- return true;
- }
-
- @Override
- public String getAbsolutePath() {
- return osUri.toString();
- }
-
- @Override
- public File getAbsoluteFile() {
- return this;
- }
-
- @Override
- public String getCanonicalPath() throws IOException {
- return osUri.toString();
- }
-
- @Override
- public File getCanonicalFile() throws IOException {
- return this;
- }
-
- @Override
- public URL toURL() throws MalformedURLException {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public URI toURI() {
- return osUri.getURI();
- }
-
- @Override
- public boolean canRead() {
- return this.exists();
- }
-
- @Override
- public boolean canWrite() {
- return this.exists();
- }
-
- @Override
- public boolean exists() {
- try {
- return connector.doesObjectExist(osUri);
- } catch (ObjectStorageException e) {
- logger.error("Fail to get object {}.", osUri, e);
- return false;
- }
- }
-
- @Override
- public boolean isDirectory() {
- return false;
- }
-
- @Override
- public boolean isFile() {
- return true;
- }
-
- @Override
- public boolean isHidden() {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public long lastModified() {
- try {
- return connector.getMetaData(osUri).lastModified();
- } catch (ObjectStorageException e) {
- logger.error("Fail to get lastModified of the object {}.", osUri, e);
- return 0;
- }
- }
-
- @Override
- public long length() {
- if (length == 0) {
- try {
- length = connector.getMetaData(osUri).length();
- } catch (ObjectStorageException e) {
- logger.error("Fail to get length of the object {}.", osUri, e);
- return 0;
- }
- }
- return length;
- }
-
- @Override
- public boolean createNewFile() throws IOException {
- try {
- return connector.createNewEmptyObject(osUri);
- } catch (ObjectStorageException e) {
- logger.error("Fail to create new object {}.", osUri, e);
- return false;
- }
- }
-
- @Override
- public boolean delete() {
- try {
- return connector.delete(osUri);
- } catch (ObjectStorageException e) {
- logger.error("Fail to delete object {}.", osUri, e);
- return false;
- }
- }
-
- @Override
- public void deleteOnExit() {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public String[] list() {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public String[] list(FilenameFilter filter) {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public File[] listFiles() {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public File[] listFiles(FilenameFilter filter) {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public File[] listFiles(FileFilter filter) {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public boolean mkdir() {
- return true;
- }
-
- @Override
- public boolean mkdirs() {
- return true;
- }
-
- @Override
- public boolean renameTo(File dest) {
- OSURI targetOSUri = ((OSFile) dest).osUri;
- try {
- return connector.renameTo(osUri, targetOSUri);
- } catch (ObjectStorageException e) {
- logger.error("Fail to rename object from {} to {}.", osUri, targetOSUri, e);
- return false;
- }
- }
-
- @Override
- public boolean setLastModified(long time) {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public boolean setReadOnly() {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public boolean setWritable(boolean writable, boolean ownerOnly) {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public boolean setWritable(boolean writable) {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public boolean setReadable(boolean readable, boolean ownerOnly) {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public boolean setReadable(boolean readable) {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public boolean setExecutable(boolean executable, boolean ownerOnly) {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public boolean setExecutable(boolean executable) {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public boolean canExecute() {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- @Override
- public long getTotalSpace() {
- // object storage has infinity space
- return Long.MAX_VALUE;
- }
-
- @Override
- public long getFreeSpace() {
- // object storage has infinity space
- return Long.MAX_VALUE;
- }
-
- @Override
- public long getUsableSpace() {
- // object storage has infinity space
- return Long.MAX_VALUE;
- }
-
- @Override
- public int compareTo(File pathname) {
- return this.toString().compareTo(pathname.toString());
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj) {
- return true;
- }
- if (obj == null || getClass() != obj.getClass()) {
- return false;
- }
- OSFile other = (OSFile) obj;
- return osUri.equals(other.osUri);
- }
-
- @Override
- public int hashCode() {
- return osUri.hashCode();
- }
-
- @Override
- public String toString() {
- return osUri.toString();
- }
-
- @Override
- public Path toPath() {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- public OSURI toOSURI() {
- return osUri;
- }
-
- public BufferedReader getBufferedReader() {
- try {
- return new BufferedReader(new InputStreamReader(connector.getInputStream(osUri)));
- } catch (ObjectStorageException e) {
- logger.error("Fail to open input stream for object {}.", osUri, e);
- return null;
- }
- }
-
- public BufferedWriter getBufferedWriter(boolean append) {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- public BufferedInputStream getBufferedInputStream() {
- try {
- return new BufferedInputStream(connector.getInputStream(osUri));
- } catch (ObjectStorageException e) {
- logger.error("Fail to open input stream for object {}.", osUri, e);
- return null;
- }
- }
-
- public BufferedOutputStream getBufferedOutputStream() {
- throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
- }
-
- public File[] listFilesBySuffix(String fileFolder, String suffix) {
- try {
- OSURI[] osUris = connector.list(new OSURI(fileFolder));
- return Arrays.stream(osUris)
- .filter(uri -> uri.toString().endsWith(suffix))
- .map(OSFile::new)
- .toArray(OSFile[]::new);
- } catch (ObjectStorageException e) {
- logger.error("Fail to list objects under the object {} by suffix {}.", osUri, suffix, e);
- return null;
- }
- }
-
- public File[] listFilesByPrefix(String fileFolder, String prefix) {
- try {
- OSURI[] osUris = connector.list(new OSURI(fileFolder));
- return Arrays.stream(osUris)
- .filter(uri -> uri.toString().startsWith(concatPath(fileFolder, prefix)))
- .map(OSFile::new)
- .toArray(OSFile[]::new);
- } catch (ObjectStorageException e) {
- logger.error("Fail to list objects under the object {} by prefix {}.", osUri, prefix, e);
- return null;
- }
- }
-
- public void putFile(File localFile) {
- try {
- connector.putLocalFile(osUri, localFile);
- } catch (ObjectStorageException e) {
- logger.error("Fail to put local file {} to the object storage {}.", localFile, osUri, e);
- }
- }
-
- public void copyTo(File destFile) {
- try {
- connector.copyObject(osUri, ((OSFile) destFile).toOSURI());
- } catch (ObjectStorageException e) {
- logger.error("Fail to copy file {} to {}.", osUri, destFile, e);
- }
- }
-
- public void deleteObjectsByPrefix() {
- try {
- connector.deleteObjectsByPrefix(osUri);
- } catch (ObjectStorageException e) {
- logger.error("Fail to delete objects by prefix {}.", osUri, e);
- }
- }
-
- // test only
- public void setConnector(ObjectStorageConnector connector) {
- this.connector = connector;
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSTsFileInput.java b/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSTsFileInput.java
deleted file mode 100644
index 3e83e56f726..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSTsFileInput.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * 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.os.fileSystem;
-
-import org.apache.iotdb.os.cache.OSFileChannel;
-import org.apache.iotdb.tsfile.read.reader.TsFileInput;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.ByteBuffer;
-
-public class OSTsFileInput implements TsFileInput {
- private OSFile file;
- private OSFileChannel channel;
-
- public OSTsFileInput(String fileURI) throws IOException {
- this(new OSFile(fileURI));
- }
-
- public OSTsFileInput(OSFile file) throws IOException {
- this.file = file;
- this.channel = new OSFileChannel(file);
- }
-
- @Override
- public long size() throws IOException {
- return channel.size();
- }
-
- @Override
- public long position() throws IOException {
- return channel.position();
- }
-
- @Override
- public TsFileInput position(long newPosition) throws IOException {
- channel.position(newPosition);
- return this;
- }
-
- @Override
- public int read(ByteBuffer dst) throws IOException {
- return channel.read(dst);
- }
-
- @Override
- public int read(ByteBuffer dst, long position) throws IOException {
- return channel.read(dst, position);
- }
-
- @Override
- public InputStream wrapAsInputStream() throws IOException {
- return OSFileChannel.newInputStream(channel);
- }
-
- @Override
- public void close() throws IOException {
- channel.close();
- }
-
- @Override
- public String getFilePath() {
- return file.getPath();
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSTsFileOutput.java b/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSTsFileOutput.java
deleted file mode 100644
index 6932623d271..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSTsFileOutput.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * 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.os.fileSystem;
-
-import org.apache.iotdb.tsfile.write.writer.TsFileOutput;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.io.OutputStream;
-import java.nio.ByteBuffer;
-
-public class OSTsFileOutput implements TsFileOutput {
- private static final Logger logger = LoggerFactory.getLogger(OSTsFileOutput.class);
-
- public OSTsFileOutput(String filePath, boolean overwrite) throws IOException {}
-
- @Override
- public void write(byte[] b) throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void write(byte b) throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void write(ByteBuffer b) throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public long getPosition() throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void close() throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public OutputStream wrapAsStream() throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void flush() throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void truncate(long size) throws IOException {
- throw new UnsupportedOperationException();
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSURI.java b/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSURI.java
deleted file mode 100644
index 66a5b10424e..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSURI.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * 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.os.fileSystem;
-
-import org.apache.iotdb.os.utils.ObjectStorageConstant;
-
-import java.net.URI;
-
-import static org.apache.iotdb.os.utils.ObjectStorageConstant.FILE_SEPARATOR;
-
-/** The OSURI format is os://{bucket}/{key} */
-public class OSURI {
- public static final String SCHEME = "os";
- public static final String OS_PREFIX = SCHEME + "://";
- private final URI uri;
-
- public OSURI(String bucket, String key) {
- this(OS_PREFIX + bucket + FILE_SEPARATOR + key);
- }
-
- public OSURI(String path) {
- this(URI.create(path));
- }
-
- public OSURI(URI uri) {
- if (!uri.getScheme().equals(SCHEME)) {
- throw new IllegalArgumentException();
- }
- this.uri = uri;
- }
-
- public URI getURI() {
- return uri;
- }
-
- public String getBucket() {
- return uri.getAuthority();
- }
-
- public String getKey() {
- return uri.getPath().substring(ObjectStorageConstant.FILE_SEPARATOR.length());
- }
-
- public String toString() {
- return uri.toString();
- }
-
- public boolean equals(Object obj) {
- if (this == obj) {
- return true;
- }
- if (obj == null || getClass() != obj.getClass()) {
- return false;
- }
- OSURI other = (OSURI) obj;
- return uri.equals(other.uri);
- }
-
- public int hashCode() {
- return uri.hashCode();
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/io/IMetaData.java b/object-storage/src/main/java/org/apache/iotdb/os/io/IMetaData.java
deleted file mode 100644
index 771f32f223e..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/io/IMetaData.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * 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.os.io;
-
-public interface IMetaData {
- long length();
-
- long lastModified();
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/io/ObjectStorageConnector.java b/object-storage/src/main/java/org/apache/iotdb/os/io/ObjectStorageConnector.java
deleted file mode 100644
index a8bb812011c..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/io/ObjectStorageConnector.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * 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.os.io;
-
-import org.apache.iotdb.os.exception.ObjectStorageException;
-import org.apache.iotdb.os.fileSystem.OSURI;
-
-import java.io.File;
-import java.io.InputStream;
-
-public interface ObjectStorageConnector {
- boolean doesObjectExist(OSURI osUri) throws ObjectStorageException;
-
- IMetaData getMetaData(OSURI osUri) throws ObjectStorageException;
-
- boolean createNewEmptyObject(OSURI osUri) throws ObjectStorageException;
-
- boolean delete(OSURI osUri) throws ObjectStorageException;
-
- boolean renameTo(OSURI fromOSUri, OSURI toOSUri) throws ObjectStorageException;
-
- InputStream getInputStream(OSURI osUri) throws ObjectStorageException;
-
- OSURI[] list(OSURI osUri) throws ObjectStorageException;
-
- void putLocalFile(OSURI osUri, File lcoalFile) throws ObjectStorageException;
-
- byte[] getRemoteObject(OSURI osUri, long position, int len) throws ObjectStorageException;
-
- void copyObject(OSURI srcUri, OSURI destUri) throws ObjectStorageException;
-
- void deleteObjectsByPrefix(OSURI prefixUri) throws ObjectStorageException;
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/io/aws/S3MetaData.java b/object-storage/src/main/java/org/apache/iotdb/os/io/aws/S3MetaData.java
deleted file mode 100644
index 35746be1dd6..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/io/aws/S3MetaData.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.os.io.aws;
-
-import org.apache.iotdb.os.io.IMetaData;
-
-public class S3MetaData implements IMetaData {
- private long length;
-
- private long lastModified;
-
- public S3MetaData(long length, long lastModified) {
- this.length = length;
- this.lastModified = lastModified;
- }
-
- @Override
- public long length() {
- return length;
- }
-
- @Override
- public long lastModified() {
- return lastModified;
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/io/aws/S3ObjectStorageConnector.java b/object-storage/src/main/java/org/apache/iotdb/os/io/aws/S3ObjectStorageConnector.java
deleted file mode 100644
index 7f4b78ccc85..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/io/aws/S3ObjectStorageConnector.java
+++ /dev/null
@@ -1,248 +0,0 @@
-/*
- * 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.os.io.aws;
-
-import org.apache.iotdb.os.conf.ObjectStorageDescriptor;
-import org.apache.iotdb.os.conf.provider.AWSS3Config;
-import org.apache.iotdb.os.exception.ObjectStorageException;
-import org.apache.iotdb.os.fileSystem.OSURI;
-import org.apache.iotdb.os.io.IMetaData;
-import org.apache.iotdb.os.io.ObjectStorageConnector;
-
-import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
-import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
-import software.amazon.awssdk.core.ResponseBytes;
-import software.amazon.awssdk.core.sync.RequestBody;
-import software.amazon.awssdk.regions.Region;
-import software.amazon.awssdk.services.s3.S3Client;
-import software.amazon.awssdk.services.s3.model.CopyObjectRequest;
-import software.amazon.awssdk.services.s3.model.Delete;
-import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
-import software.amazon.awssdk.services.s3.model.DeleteObjectResponse;
-import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest;
-import software.amazon.awssdk.services.s3.model.GetObjectRequest;
-import software.amazon.awssdk.services.s3.model.GetObjectResponse;
-import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
-import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
-import software.amazon.awssdk.services.s3.model.ListObjectsRequest;
-import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
-import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
-import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
-import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
-import software.amazon.awssdk.services.s3.model.PutObjectRequest;
-import software.amazon.awssdk.services.s3.model.S3Exception;
-
-import java.io.File;
-import java.io.InputStream;
-import java.util.stream.Collectors;
-
-public class S3ObjectStorageConnector implements ObjectStorageConnector {
- private static final String RANGE_FORMAT = "%d-%d";
- private final AWSS3Config s3config =
- (AWSS3Config) ObjectStorageDescriptor.getInstance().getConfig().getProviderConfig();
- private final S3Client s3Client =
- S3Client.builder()
- .region(Region.of(s3config.getEndpoint()))
- .credentialsProvider(
- StaticCredentialsProvider.create(
- AwsBasicCredentials.create(
- s3config.getAccessKeyId(), s3config.getAccessKeySecret())))
- .build();
-
- @Override
- public boolean doesObjectExist(OSURI osUri) throws ObjectStorageException {
- try {
- HeadObjectRequest req =
- HeadObjectRequest.builder().bucket(osUri.getBucket()).key(osUri.getKey()).build();
- s3Client.headObject(req);
- return true;
- } catch (NoSuchKeyException e) {
- return false;
- } catch (S3Exception e) {
- throw new ObjectStorageException(e);
- }
- }
-
- @Override
- public IMetaData getMetaData(OSURI osUri) throws ObjectStorageException {
- try {
- HeadObjectRequest req =
- HeadObjectRequest.builder().bucket(osUri.getBucket()).key(osUri.getKey()).build();
- HeadObjectResponse resp = s3Client.headObject(req);
- return new S3MetaData(resp.contentLength(), resp.lastModified().toEpochMilli());
- } catch (S3Exception e) {
- throw new ObjectStorageException(e);
- }
- }
-
- @Override
- public boolean createNewEmptyObject(OSURI osUri) throws ObjectStorageException {
- try {
- PutObjectRequest req =
- PutObjectRequest.builder().bucket(osUri.getBucket()).key(osUri.getKey()).build();
- s3Client.putObject(req, RequestBody.empty());
- return true;
- } catch (S3Exception e) {
- throw new ObjectStorageException(e);
- }
- }
-
- @Override
- public boolean delete(OSURI osUri) throws ObjectStorageException {
- try {
- DeleteObjectRequest req =
- DeleteObjectRequest.builder().bucket(osUri.getBucket()).key(osUri.getKey()).build();
- DeleteObjectResponse resp = s3Client.deleteObject(req);
- return resp.deleteMarker();
- } catch (S3Exception e) {
- throw new ObjectStorageException(e);
- }
- }
-
- @Override
- public boolean renameTo(OSURI fromOSUri, OSURI toOSUri) throws ObjectStorageException {
- try {
- CopyObjectRequest copyReq =
- CopyObjectRequest.builder()
- .sourceBucket(fromOSUri.getBucket())
- .sourceKey(fromOSUri.getKey())
- .destinationBucket(toOSUri.getBucket())
- .destinationKey(toOSUri.getKey())
- .build();
- s3Client.copyObject(copyReq);
-
- DeleteObjectRequest deleteReq =
- DeleteObjectRequest.builder()
- .bucket(fromOSUri.getBucket())
- .key(fromOSUri.getKey())
- .build();
- DeleteObjectResponse resp = s3Client.deleteObject(deleteReq);
- return resp.deleteMarker();
- } catch (S3Exception e) {
- throw new ObjectStorageException(e);
- }
- }
-
- public OSURI[] list(OSURI osUri) throws ObjectStorageException {
- try {
- ListObjectsRequest req =
- ListObjectsRequest.builder().bucket(osUri.getBucket()).prefix(osUri.getKey()).build();
- return s3Client.listObjects(req).contents().stream()
- .map(obj -> new OSURI(osUri.getBucket(), obj.key()))
- .toArray(OSURI[]::new);
- } catch (S3Exception e) {
- throw new ObjectStorageException(e);
- }
- }
-
- @Override
- public InputStream getInputStream(OSURI osUri) throws ObjectStorageException {
- try {
- GetObjectRequest req =
- GetObjectRequest.builder().bucket(osUri.getBucket()).key(osUri.getKey()).build();
- return s3Client.getObject(req);
- } catch (S3Exception e) {
- throw new ObjectStorageException(e);
- }
- }
-
- @Override
- public void putLocalFile(OSURI osUri, File lcoalFile) throws ObjectStorageException {
- try {
- PutObjectRequest req =
- PutObjectRequest.builder().bucket(osUri.getBucket()).key(osUri.getKey()).build();
- s3Client.putObject(req, RequestBody.fromFile(lcoalFile));
- } catch (S3Exception e) {
- throw new ObjectStorageException(e);
- }
- }
-
- @Override
- public byte[] getRemoteObject(OSURI osUri, long position, int len) throws ObjectStorageException {
- String rangeStr = String.format(RANGE_FORMAT, position, position + len - 1);
- try {
- GetObjectRequest req =
- GetObjectRequest.builder()
- .bucket(osUri.getBucket())
- .key(osUri.getKey())
- .range(rangeStr)
- .build();
- ResponseBytes<GetObjectResponse> resp = s3Client.getObjectAsBytes(req);
- return resp.asByteArray();
- } catch (S3Exception e) {
- throw new ObjectStorageException(e);
- }
- }
-
- @Override
- public void copyObject(OSURI srcUri, OSURI destUri) throws ObjectStorageException {
- try {
- CopyObjectRequest req =
- CopyObjectRequest.builder()
- .sourceBucket(srcUri.getBucket())
- .sourceKey(srcUri.getKey())
- .destinationBucket(destUri.getBucket())
- .destinationKey(destUri.getKey())
- .build();
- s3Client.copyObject(req);
- } catch (S3Exception e) {
- throw new ObjectStorageException(e);
- }
- }
-
- @Override
- public void deleteObjectsByPrefix(OSURI prefixUri) throws ObjectStorageException {
- try {
- ListObjectsV2Request listReq =
- ListObjectsV2Request.builder()
- .bucket(prefixUri.getBucket())
- .prefix(prefixUri.getKey())
- .build();
- ListObjectsV2Response listRes;
-
- do {
- listRes = s3Client.listObjectsV2(listReq);
- Delete del =
- Delete.builder()
- .objects(
- listRes.contents().stream()
- .map(s3Object -> ObjectIdentifier.builder().key(s3Object.key()).build())
- .collect(Collectors.toList()))
- .build();
- DeleteObjectsRequest deleteObjectsRequest =
- DeleteObjectsRequest.builder().delete(del).build();
- s3Client.deleteObjects(deleteObjectsRequest);
-
- listReq =
- ListObjectsV2Request.builder()
- .bucket(prefixUri.getBucket())
- .prefix(prefixUri.getKey())
- .continuationToken(listRes.nextContinuationToken())
- .build();
- } while (listRes.isTruncated());
- } catch (S3Exception e) {
- throw new ObjectStorageException(e);
- }
- }
-
- public void close() {
- s3Client.close();
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/io/test/TestObjectStorageConnector.java b/object-storage/src/main/java/org/apache/iotdb/os/io/test/TestObjectStorageConnector.java
deleted file mode 100644
index 125e5f9f79e..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/io/test/TestObjectStorageConnector.java
+++ /dev/null
@@ -1,142 +0,0 @@
-/*
- * 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.os.io.test;
-
-import org.apache.iotdb.os.conf.ObjectStorageDescriptor;
-import org.apache.iotdb.os.conf.provider.TestConfig;
-import org.apache.iotdb.os.exception.ObjectStorageException;
-import org.apache.iotdb.os.fileSystem.OSURI;
-import org.apache.iotdb.os.io.IMetaData;
-import org.apache.iotdb.os.io.ObjectStorageConnector;
-import org.apache.iotdb.os.io.aws.S3MetaData;
-import org.apache.iotdb.os.utils.ObjectStorageConstant;
-
-import org.apache.commons.io.FileUtils;
-
-import java.io.File;
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.ByteBuffer;
-import java.nio.channels.Channels;
-import java.nio.channels.FileChannel;
-import java.nio.file.StandardOpenOption;
-
-public class TestObjectStorageConnector implements ObjectStorageConnector {
- private final TestConfig testConfig =
- (TestConfig) ObjectStorageDescriptor.getInstance().getConfig().getProviderConfig();
-
- @Override
- public boolean doesObjectExist(OSURI osUri) throws ObjectStorageException {
- File file = new File(getDstFilePath(osUri));
- return file.exists();
- }
-
- @Override
- public IMetaData getMetaData(OSURI osUri) throws ObjectStorageException {
- File file = new File(getDstFilePath(osUri));
- return new S3MetaData(file.length(), System.currentTimeMillis());
- }
-
- @Override
- public boolean createNewEmptyObject(OSURI osUri) throws ObjectStorageException {
- File file = new File(getDstFilePath(osUri));
- if (!file.exists()) {
- try {
- return file.createNewFile();
- } catch (IOException e) {
- throw new ObjectStorageException(e);
- }
- }
- return false;
- }
-
- @Override
- public boolean delete(OSURI osUri) throws ObjectStorageException {
- File file = new File(getDstFilePath(osUri));
- return file.delete();
- }
-
- @Override
- public boolean renameTo(OSURI fromOSUri, OSURI toOSUri) throws ObjectStorageException {
- File file = new File(getDstFilePath(fromOSUri));
- return file.renameTo(new File(getDstFilePath(toOSUri)));
- }
-
- @Override
- public InputStream getInputStream(OSURI osUri) throws ObjectStorageException {
- File file = new File(getDstFilePath(osUri));
- try {
- return Channels.newInputStream(FileChannel.open(file.toPath(), StandardOpenOption.READ));
- } catch (IOException e) {
- throw new ObjectStorageException(e);
- }
- }
-
- @Override
- public OSURI[] list(OSURI osUri) throws ObjectStorageException {
- return null;
- }
-
- @Override
- public void putLocalFile(OSURI osUri, File localFile) throws ObjectStorageException {
- try {
- File targetFile = new File(getDstFilePath(osUri));
- if (!targetFile.getParentFile().exists() && !targetFile.getParentFile().mkdirs()) {
- throw new ObjectStorageException(
- String.format(
- "[TieredMigration] cannot mkdir for path %s",
- targetFile.getParentFile().getAbsolutePath()));
- }
- FileUtils.copyFile(localFile, targetFile);
- } catch (IOException e) {
- throw new ObjectStorageException(e);
- }
- }
-
- @Override
- public byte[] getRemoteObject(OSURI osUri, long position, int len) throws ObjectStorageException {
- File file = new File(getDstFilePath(osUri));
- ByteBuffer dst = ByteBuffer.allocate(len);
- try (FileChannel channel = FileChannel.open(file.toPath(), StandardOpenOption.READ)) {
- channel.read(dst, position);
- } catch (IOException e) {
- throw new ObjectStorageException(e);
- }
- return dst.array();
- }
-
- @Override
- public void deleteObjectsByPrefix(OSURI prefixUri) throws ObjectStorageException {
- File file = new File(getDstFilePath(prefixUri));
- try {
- FileUtils.deleteDirectory(file);
- } catch (IOException e) {
- throw new ObjectStorageException(e);
- }
- }
-
- private String getDstFilePath(OSURI osuri) {
- return testConfig.getTestDir()
- + File.separator
- + osuri.getKey().replace(ObjectStorageConstant.FILE_SEPARATOR, File.separator);
- }
-
- @Override
- public void copyObject(OSURI srcUri, OSURI destUri) throws ObjectStorageException {}
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/utils/ObjectStorageConstant.java b/object-storage/src/main/java/org/apache/iotdb/os/utils/ObjectStorageConstant.java
deleted file mode 100644
index 226b13bfd54..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/utils/ObjectStorageConstant.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * 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.os.utils;
-
-public class ObjectStorageConstant {
- public static final String FILE_SEPARATOR = "/";
- public static final String CACHE_FILE_SUFFIX = ".cache";
- public static final String TMP_CACHE_FILE_SUFFIX = ".cache.tmp";
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/utils/ObjectStorageType.java b/object-storage/src/main/java/org/apache/iotdb/os/utils/ObjectStorageType.java
deleted file mode 100644
index 1548c2bab1d..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/utils/ObjectStorageType.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * 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.os.utils;
-
-import org.apache.iotdb.os.conf.ObjectStorageDescriptor;
-import org.apache.iotdb.os.io.ObjectStorageConnector;
-import org.apache.iotdb.os.io.aws.S3ObjectStorageConnector;
-import org.apache.iotdb.os.io.test.TestObjectStorageConnector;
-
-public enum ObjectStorageType {
- TEST,
- AWS_S3;
-
- public static ObjectStorageConnector getConnector(ObjectStorageType type) {
- switch (type) {
- case AWS_S3:
- return new S3ObjectStorageConnector();
- case TEST:
- return new TestObjectStorageConnector();
- default:
- return null;
- }
- }
-
- public static ObjectStorageConnector getConnector() {
- return getConnector(ObjectStorageDescriptor.getInstance().getConfig().getOsType());
- }
-}
diff --git a/object-storage/src/test/java/org/apache/iotdb/os/cache/OSFileCacheTest.java b/object-storage/src/test/java/org/apache/iotdb/os/cache/OSFileCacheTest.java
deleted file mode 100644
index 3b3b2f0a1c8..00000000000
--- a/object-storage/src/test/java/org/apache/iotdb/os/cache/OSFileCacheTest.java
+++ /dev/null
@@ -1,137 +0,0 @@
-/*
- * 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.os.cache;
-
-import org.apache.iotdb.os.conf.ObjectStorageConfig;
-import org.apache.iotdb.os.conf.ObjectStorageDescriptor;
-import org.apache.iotdb.os.fileSystem.OSFile;
-import org.apache.iotdb.os.fileSystem.OSURI;
-import org.apache.iotdb.os.io.ObjectStorageConnector;
-import org.apache.iotdb.os.utils.ObjectStorageType;
-
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
-import org.mockito.Mock;
-import org.mockito.junit.MockitoJUnitRunner;
-import org.testcontainers.shaded.org.apache.commons.io.FileUtils;
-
-import java.io.File;
-import java.nio.ByteBuffer;
-import java.nio.channels.FileChannel;
-import java.nio.file.StandardOpenOption;
-
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.when;
-
-@RunWith(MockitoJUnitRunner.class)
-public class OSFileCacheTest {
- private static final File cacheDir = new File("target" + File.separator + "cache");
- private static final ObjectStorageConfig config =
- ObjectStorageDescriptor.getInstance().getConfig();
- @InjectMocks private OSFileCache cache = new OSFileCache(ObjectStorageType.TEST);
-
- @InjectMocks
- private OSFile testFile =
- new OSFile(new OSURI("test_bucket", "test_key"), ObjectStorageType.TEST);
-
- @Mock private ObjectStorageConnector connector;
- private static ObjectStorageType prevObjectStorageType;
- private int prevCachePageSize;
- private String[] prevCacheDirs;
-
- @BeforeClass
- public static void beforeClass() throws Exception {
- prevObjectStorageType = config.getOsType();
- config.setOsType(ObjectStorageType.TEST);
- }
-
- @AfterClass
- public static void afterClass() throws Exception {
- config.setOsType(prevObjectStorageType);
- }
-
- @Before
- public void setUp() throws Exception {
- prevCachePageSize = config.getCachePageSize();
- config.setCachePageSize(100);
- prevCacheDirs = config.getCacheDirs();
- config.setCacheDirs(new String[] {cacheDir.getPath()});
- cacheDir.mkdirs();
- CacheFileManager.getInstance().setCacheFileId(0);
- }
-
- @After
- public void tearDown() throws Exception {
- config.setCachePageSize(prevCachePageSize);
- config.setCacheDirs(prevCacheDirs);
- FileUtils.deleteDirectory(cacheDir);
- }
-
- @Test
- public void testGet() throws Exception {
- // pull 0-100, and 0-100 exists
- byte[] bytes0to100 = new byte[100];
- for (int i = 0; i < 100; ++i) {
- bytes0to100[i] = (byte) i;
- }
- when(connector.getRemoteObject(testFile.toOSURI(), 0, 100)).thenReturn(bytes0to100);
- OSFileCacheKey key0to100 = new OSFileCacheKey(testFile, 0);
- OSFileCacheValue value0to100 = cache.get(key0to100);
- assertTrue(value0to100.getCacheFile().exists());
- assertEquals(value0to100.getCacheFile().length(), value0to100.getLength());
- try (FileChannel channel =
- FileChannel.open(value0to100.getCacheFile().toPath(), StandardOpenOption.READ)) {
- byte[] readBytes = new byte[100];
- ByteBuffer buffer = ByteBuffer.wrap(readBytes);
- int bytesNum =
- channel.read(
- buffer, value0to100.getStartPositionInCacheFile() + value0to100.getMetaSize());
- assertEquals(100, bytesNum);
- assertEquals(100, value0to100.getDataSize());
- assertArrayEquals(bytes0to100, readBytes);
- }
- // pull 100-200, but only 100-150 exists
- byte[] bytes100to150 = new byte[150 - 100];
- for (int i = 0; i < 150 - 100; ++i) {
- bytes100to150[i] = (byte) i;
- }
- when(connector.getRemoteObject(testFile.toOSURI(), 100, 100)).thenReturn(bytes100to150);
- OSFileCacheKey key100to150 = new OSFileCacheKey(testFile, 100);
- OSFileCacheValue value100to150 = cache.get(key100to150);
- assertTrue(value100to150.getCacheFile().exists());
- try (FileChannel channel =
- FileChannel.open(value100to150.getCacheFile().toPath(), StandardOpenOption.READ)) {
- byte[] readBytes = new byte[50];
- ByteBuffer buffer = ByteBuffer.wrap(readBytes);
- int bytesNum =
- channel.read(
- buffer, value100to150.getStartPositionInCacheFile() + value100to150.getMetaSize());
- assertEquals(50, bytesNum);
- assertEquals(50, value100to150.getDataSize());
- assertArrayEquals(bytes100to150, readBytes);
- }
- }
-}
diff --git a/object-storage/src/test/java/org/apache/iotdb/os/cache/OSFileChannelTest.java b/object-storage/src/test/java/org/apache/iotdb/os/cache/OSFileChannelTest.java
deleted file mode 100644
index 614affeeede..00000000000
--- a/object-storage/src/test/java/org/apache/iotdb/os/cache/OSFileChannelTest.java
+++ /dev/null
@@ -1,219 +0,0 @@
-/*
- * 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.os.cache;
-
-import org.apache.iotdb.os.conf.ObjectStorageConfig;
-import org.apache.iotdb.os.conf.ObjectStorageDescriptor;
-import org.apache.iotdb.os.exception.ObjectStorageException;
-import org.apache.iotdb.os.fileSystem.OSFile;
-import org.apache.iotdb.os.fileSystem.OSURI;
-import org.apache.iotdb.os.io.ObjectStorageConnector;
-import org.apache.iotdb.os.io.aws.S3MetaData;
-import org.apache.iotdb.os.utils.ObjectStorageType;
-
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
-import org.mockito.Mock;
-import org.mockito.junit.MockitoJUnitRunner;
-import org.testcontainers.shaded.org.apache.commons.io.FileUtils;
-
-import java.io.File;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-
-import static org.junit.Assert.assertEquals;
-import static org.mockito.Mockito.when;
-
-@RunWith(MockitoJUnitRunner.class)
-public class OSFileChannelTest {
- private static final File cacheDir = new File("target" + File.separator + "cache");
- private static final ObjectStorageConfig config =
- ObjectStorageDescriptor.getInstance().getConfig();
- private static final int cachePageSize = 100;
- @InjectMocks private OSFileCache cache = new OSFileCache(ObjectStorageType.TEST);
-
- @InjectMocks
- private OSFile testFile =
- new OSFile(new OSURI("test_bucket", "test_key"), ObjectStorageType.TEST);
-
- @Mock private ObjectStorageConnector connector;
- private static ObjectStorageType prevObjectStorageType;
- private int prevCachePageSize;
- private String[] prevCacheDirs;
-
- @BeforeClass
- public static void beforeClass() throws Exception {
- prevObjectStorageType = config.getOsType();
- config.setOsType(ObjectStorageType.TEST);
- }
-
- @AfterClass
- public static void afterClass() throws Exception {
- config.setOsType(prevObjectStorageType);
- }
-
- @Before
- public void setUp() throws Exception {
- prevCachePageSize = config.getCachePageSize();
- config.setCachePageSize(cachePageSize);
- prevCacheDirs = config.getCacheDirs();
- config.setCacheDirs(new String[] {cacheDir.getPath()});
- cacheDir.mkdirs();
- CacheFileManager.getInstance().setCacheFileId(0);
- }
-
- @After
- public void tearDown() throws Exception {
- config.setCachePageSize(prevCachePageSize);
- config.setCacheDirs(prevCacheDirs);
- FileUtils.deleteDirectory(cacheDir);
- }
-
- private void prepareOSFile(int size) throws ObjectStorageException {
- when(connector.getMetaData(testFile.toOSURI()))
- .thenReturn(new S3MetaData(size, System.currentTimeMillis()));
- int startPos = 0;
- while (startPos < size) {
- byte[] bytes = new byte[(int) Math.min(cachePageSize, size - startPos)];
- for (int i = 0; i < bytes.length; ++i) {
- bytes[i] = (byte) (startPos + i);
- }
- when(connector.getRemoteObject(testFile.toOSURI(), startPos, cachePageSize))
- .thenReturn(bytes);
- startPos += cachePageSize;
- }
- }
-
- @Test
- public void testReadFileSmallerThanPageSize() throws Exception {
- int size = 50;
- prepareOSFile(size);
- try (OSFileChannel channel = new OSFileChannel(testFile, cache)) {
- ByteBuffer dst = ByteBuffer.allocate(cachePageSize);
- assertEquals(size, channel.read(dst));
- dst.flip();
- for (int i = 0; i < size; ++i) {
- assertEquals((byte) i, dst.get());
- }
- }
- }
-
- @Test
- public void testReadFileBiggerThanPageSize() throws Exception {
- int size = 550;
- prepareOSFile(size);
- try (OSFileChannel channel = new OSFileChannel(testFile, cache)) {
- ByteBuffer dst = ByteBuffer.allocate(size);
- assertEquals(size, channel.read(dst));
- dst.flip();
- for (int i = 0; i < size; ++i) {
- assertEquals((byte) i, dst.get());
- }
- }
- }
-
- @Test
- public void testByteBufferBiggerThanFileSize() throws Exception {
- int fileSize = 550;
- int byteBufferSize = 600;
- prepareOSFile(fileSize);
- try (OSFileChannel channel = new OSFileChannel(testFile, cache)) {
- ByteBuffer dst = ByteBuffer.allocate(byteBufferSize);
- assertEquals(fileSize, channel.read(dst));
- assertEquals(byteBufferSize, dst.limit());
- dst.flip();
- for (int i = 0; i < fileSize; ++i) {
- assertEquals((byte) i, dst.get());
- }
- }
- }
-
- @Test
- public void testReadByPosition() throws Exception {
- int size = 550;
- prepareOSFile(size);
- try (OSFileChannel channel = new OSFileChannel(testFile, cache)) {
- int position = 0;
- while (position < size) {
- ByteBuffer dst = ByteBuffer.allocate(50);
- assertEquals(50, channel.read(dst, position));
- dst.flip();
- for (int i = 0; i < 50; ++i) {
- assertEquals((byte) (position + i), dst.get());
- }
- position += 50;
- }
- }
- }
-
- @Test
- public void testConcurrentRead() throws Exception {
- int size = 550;
- prepareOSFile(size);
- try (OSFileChannel channel = new OSFileChannel(testFile, cache)) {
- // random read by
- int threadsNum = 3;
- ExecutorService executorService = Executors.newFixedThreadPool(threadsNum);
- List<Future<Void>> futures = new ArrayList<>();
- for (int i = 0; i < threadsNum * 3; ++i) {
- Callable<Void> readTask =
- () -> {
- int position = 0;
- while (position < size) {
- ByteBuffer dst = ByteBuffer.allocate(50);
- assertEquals(50, channel.read(dst, position));
- dst.flip();
- for (int j = 0; j < 50; ++j) {
- assertEquals((byte) (position + j), dst.get());
- }
- position += 50;
- }
- return null;
- };
- Future<Void> future = executorService.submit(readTask);
- futures.add(future);
- }
- // sequence read
- int position = 0;
- while (position < size) {
- ByteBuffer dst = ByteBuffer.allocate(50);
- assertEquals(50, channel.read(dst));
- dst.flip();
- for (int i = 0; i < 50; ++i) {
- assertEquals((byte) (position + i), dst.get());
- }
- position += 50;
- }
- // wait all random read tasks end
- for (Future<Void> future : futures) {
- future.get();
- }
- }
- }
-}
diff --git a/object-storage/src/test/java/org/apache/iotdb/os/cache/OSInputStreamTest.java b/object-storage/src/test/java/org/apache/iotdb/os/cache/OSInputStreamTest.java
deleted file mode 100644
index 3835923af38..00000000000
--- a/object-storage/src/test/java/org/apache/iotdb/os/cache/OSInputStreamTest.java
+++ /dev/null
@@ -1,181 +0,0 @@
-/*
- * 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.os.cache;
-
-import org.apache.iotdb.os.conf.ObjectStorageConfig;
-import org.apache.iotdb.os.conf.ObjectStorageDescriptor;
-import org.apache.iotdb.os.exception.ObjectStorageException;
-import org.apache.iotdb.os.fileSystem.OSFile;
-import org.apache.iotdb.os.fileSystem.OSURI;
-import org.apache.iotdb.os.io.ObjectStorageConnector;
-import org.apache.iotdb.os.io.aws.S3MetaData;
-import org.apache.iotdb.os.utils.ObjectStorageType;
-
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
-import org.mockito.Mock;
-import org.mockito.junit.MockitoJUnitRunner;
-import org.testcontainers.shaded.org.apache.commons.io.FileUtils;
-
-import java.io.File;
-import java.io.InputStream;
-
-import static org.junit.Assert.assertEquals;
-import static org.mockito.Mockito.when;
-
-@RunWith(MockitoJUnitRunner.class)
-public class OSInputStreamTest {
- private static final File cacheDir = new File("target" + File.separator + "cache");
- private static final ObjectStorageConfig config =
- ObjectStorageDescriptor.getInstance().getConfig();
- private static final int cachePageSize = 100;
- @InjectMocks private OSFileCache cache = new OSFileCache(ObjectStorageType.TEST);
-
- @InjectMocks
- private OSFile testFile =
- new OSFile(new OSURI("test_bucket", "test_key"), ObjectStorageType.TEST);
-
- @Mock private ObjectStorageConnector connector;
- private static ObjectStorageType prevObjectStorageType;
- private int prevCachePageSize;
- private String[] prevCacheDirs;
-
- @BeforeClass
- public static void beforeClass() throws Exception {
- prevObjectStorageType = config.getOsType();
- config.setOsType(ObjectStorageType.TEST);
- }
-
- @AfterClass
- public static void afterClass() throws Exception {
- config.setOsType(prevObjectStorageType);
- }
-
- @Before
- public void setUp() throws Exception {
- prevCachePageSize = config.getCachePageSize();
- config.setCachePageSize(cachePageSize);
- prevCacheDirs = config.getCacheDirs();
- config.setCacheDirs(new String[] {cacheDir.getPath()});
- cacheDir.mkdirs();
- CacheFileManager.getInstance().setCacheFileId(0);
- }
-
- @After
- public void tearDown() throws Exception {
- config.setCachePageSize(prevCachePageSize);
- config.setCacheDirs(prevCacheDirs);
- FileUtils.deleteDirectory(cacheDir);
- }
-
- private void prepareOSFile(int size) throws ObjectStorageException {
- when(connector.getMetaData(testFile.toOSURI()))
- .thenReturn(new S3MetaData(size, System.currentTimeMillis()));
- int startPos = 0;
- while (startPos < size) {
- byte[] bytes = new byte[(int) Math.min(cachePageSize, size - startPos)];
- for (int i = 0; i < bytes.length; ++i) {
- bytes[i] = (byte) (startPos + i);
- }
- when(connector.getRemoteObject(testFile.toOSURI(), startPos, cachePageSize))
- .thenReturn(bytes);
- startPos += cachePageSize;
- }
- }
-
- @Test
- public void testReadByte() throws Exception {
- int size = 50;
- prepareOSFile(size);
- try (OSFileChannel channel = new OSFileChannel(testFile, cache);
- InputStream in = OSFileChannel.newInputStream(channel)) {
- for (int i = 0; in.available() > 0; ++i) {
- assertEquals((byte) i, in.read());
- }
- assertEquals(0, in.available());
- }
- }
-
- @Test
- public void testReadByteArray() throws Exception {
- int size = 128;
- prepareOSFile(size);
- try (OSFileChannel channel = new OSFileChannel(testFile, cache);
- InputStream in = OSFileChannel.newInputStream(channel)) {
- int position = 0;
- while (position < size) {
- byte[] bytes = new byte[50];
- assertEquals(Math.min(50, size - position), in.read(bytes));
- for (int i = 0; i < Math.min(50, size - position); ++i) {
- assertEquals((byte) (position + i), bytes[i]);
- }
- position += 50;
- }
- assertEquals(0, in.available());
- }
- }
-
- @Test
- public void testReadByteArrayByLen() throws Exception {
- int size = 550;
- prepareOSFile(size);
- try (OSFileChannel channel = new OSFileChannel(testFile, cache);
- InputStream in = OSFileChannel.newInputStream(channel)) {
- byte[] bytes = new byte[size];
- int position = 0;
- while (position < size) {
- assertEquals(50, in.read(bytes, position, 50));
- for (int i = 0; i < 50; ++i) {
- assertEquals((byte) (position + i), bytes[position + i]);
- }
- position += 50;
- }
- assertEquals(0, in.available());
- }
- }
-
- @Test
- public void testSkip() throws Exception {
- int size = 550;
- prepareOSFile(size);
- try (OSFileChannel channel = new OSFileChannel(testFile, cache);
- InputStream in = OSFileChannel.newInputStream(channel)) {
- byte[] bytes = new byte[size];
- int position = 0;
- while (position < size) {
- if (position % 100 == 0) {
- in.skip(50);
- position += 50;
- continue;
- }
- assertEquals(50, in.read(bytes, position, 50));
- for (int i = 0; i < 50; ++i) {
- assertEquals((byte) (position + i), bytes[position + i]);
- }
- position += 50;
- }
- assertEquals(0, in.available());
- }
- }
-}
diff --git a/object-storage/src/test/resources/logback.xml b/object-storage/src/test/resources/logback.xml
deleted file mode 100644
index ac3fcb3a7f8..00000000000
--- a/object-storage/src/test/resources/logback.xml
+++ /dev/null
@@ -1,41 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-
- 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.
-
--->
-<configuration debug="false">
- <property name="LOG_PATH" value="target/logs"/>
- <!-- prevent logback from outputting its own status at the start of every log -->
- <statusListener class="ch.qos.logback.core.status.NopStatusListener"/>
- <appender class="ch.qos.logback.core.ConsoleAppender" name="stdout">
- <Target>System.out</Target>
- <encoder>
- <pattern>%-5p [%d] [%thread] %C:%L - %m %n</pattern>
- <charset>utf-8</charset>
- </encoder>
- <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
- <level>DEBUG</level>
- </filter>
- </appender>
- <!--<logger name="org.apache.iotdb.db.utils.OpenFileNumUtil" level="debug" />-->
- <!--<logger name="org.apache.iotdb.db.utils.OpenFileNumUtilTest" level="debug" />-->
- <root level="WARN">
- <appender-ref ref="stdout"/>
- </root>
-</configuration>
diff --git a/pom.xml b/pom.xml
index a8198fbd759..17c23f639fe 100644
--- a/pom.xml
+++ b/pom.xml
@@ -120,7 +120,6 @@
<module>iotdb-api/external-api</module>
<module>iotdb-client/isession</module>
<module>mlnode</module>
- <module>object-storage</module>
<module>iotdb-api/pipe-api</module>
<module>iotdb-api/subscription-api</module>
</modules>
diff --git a/server/pom.xml b/server/pom.xml
index 4538676180e..fbda65880f3 100644
--- a/server/pom.xml
+++ b/server/pom.xml
@@ -38,11 +38,6 @@
<tomcat-embed-core.version>10.1.0-M1</tomcat-embed-core.version>
</properties>
<dependencies>
- <dependency>
- <groupId>org.apache.iotdb</groupId>
- <artifactId>os-tsfile</artifactId>
- <version>${project.version}</version>
- </dependency>
<dependency>
<groupId>org.apache.iotdb</groupId>
<artifactId>service-rpc</artifactId>
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 56335e03c2e..ca066ec543a 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -38,9 +38,6 @@ import org.apache.iotdb.db.service.thrift.impl.ClientRPCServiceImpl;
import org.apache.iotdb.db.service.thrift.impl.NewInfluxDBServiceImpl;
import org.apache.iotdb.db.utils.datastructure.TVListSortAlgorithm;
import org.apache.iotdb.db.wal.utils.WALMode;
-import org.apache.iotdb.os.conf.ObjectStorageConfig;
-import org.apache.iotdb.os.conf.ObjectStorageDescriptor;
-import org.apache.iotdb.os.utils.ObjectStorageType;
import org.apache.iotdb.rpc.RpcTransportFactory;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
@@ -1137,17 +1134,6 @@ public class IoTDBConfig {
*/
private String RateLimiterType = "FixedIntervalRateLimiter";
- /** Threads for migration tasks */
- private int migrateThreadCount = 1;
-
- private double[] spaceMoveThresholds = {0.15};
-
- /** Enable hdfs or not */
- private boolean enableObjectStorage = true;
-
- /** Config for object storage */
- private ObjectStorageConfig osConfig = ObjectStorageDescriptor.getInstance().getConfig();
-
IoTDBConfig() {}
public int getMaxLogEntriesNumPerBatch() {
@@ -3941,94 +3927,6 @@ public class IoTDBConfig {
return sortTmpDir;
}
- public int getMigrateThreadCount() {
- return migrateThreadCount;
- }
-
- public void setMigrateThreadCount(int migrateThreadCount) {
- this.migrateThreadCount = migrateThreadCount;
- }
-
- public double[] getSpaceMoveThresholds() {
- return spaceMoveThresholds;
- }
-
- public void setSpaceMoveThresholds(double[] spaceMoveThresholds) {
- this.spaceMoveThresholds = spaceMoveThresholds;
- }
-
- public boolean isEnableObjectStorage() {
- return enableObjectStorage;
- }
-
- public void setEnableObjectStorage(boolean enableObjectStorage) {
- this.enableObjectStorage = enableObjectStorage;
- }
-
- public String getObjectStorageName() {
- return osConfig.getOsType().name();
- }
-
- public void setObjectStorageName(String objectStorageName) {
- osConfig.setOsType(ObjectStorageType.valueOf(objectStorageName));
- }
-
- public String getObjectStorageBucket() {
- return osConfig.getBucketName();
- }
-
- public void setObjectStorageBucket(String objectStorageBucket) {
- osConfig.setBucketName(objectStorageBucket);
- }
-
- public String getObjectStorageEndpoint() {
- return osConfig.getEndpoint();
- }
-
- public void setObjectStorageEndpoint(String objectStorageEndpoint) {
- osConfig.setEndpoint(objectStorageEndpoint);
- }
-
- public String getObjectStorageAccessKey() {
- return osConfig.getAccessKeyId();
- }
-
- public void setObjectStorageAccessKey(String objectStorageAccessKey) {
- osConfig.setAccessKeyId(objectStorageAccessKey);
- }
-
- public String getObjectStorageAccessSecret() {
- return osConfig.getAccessKeySecret();
- }
-
- public void setObjectStorageAccessSecret(String objectStorageAccessSecret) {
- osConfig.setAccessKeySecret(objectStorageAccessSecret);
- }
-
- public String[] getCacheDirs() {
- return osConfig.getCacheDirs();
- }
-
- public void setCacheDirs(String[] cacheDirs) {
- osConfig.setCacheDirs(cacheDirs);
- }
-
- public long getCacheMaxDiskUsage() {
- return osConfig.getCacheMaxDiskUsage();
- }
-
- public void setCacheMaxDiskUsage(long cacheMaxDiskUsage) {
- osConfig.setCacheMaxDiskUsage(cacheMaxDiskUsage);
- }
-
- public int getCachePageSize() {
- return osConfig.getCachePageSize();
- }
-
- public void setCachePageSize(int cachePageSize) {
- osConfig.setCachePageSize(cachePageSize);
- }
-
public String getClusterSchemaLimitLevel() {
return clusterSchemaLimitLevel;
}
@@ -4044,4 +3942,8 @@ public class IoTDBConfig {
public void setClusterSchemaLimitThreshold(long clusterSchemaLimitThreshold) {
this.clusterSchemaLimitThreshold = clusterSchemaLimitThreshold;
}
+
+ public String getObjectStorageBucket() {
+ throw new UnsupportedOperationException("object storage is not supported yet");
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 3c67feff6c7..74b59fff731 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -975,11 +975,6 @@ public class IoTDBDescriptor {
// At the same time, set TSFileConfig
List<FSType> fsTypes = new ArrayList<>();
fsTypes.add(FSType.LOCAL);
- if (Boolean.parseBoolean(
- properties.getProperty(
- "enable_object_storage", String.valueOf(conf.isEnableObjectStorage())))) {
- fsTypes.add(FSType.OBJECT_STORAGE);
- }
if (Boolean.parseBoolean(
properties.getProperty("enable_hdfs", String.valueOf(conf.isEnableHDFS())))) {
fsTypes.add(FSType.HDFS);
@@ -1083,12 +1078,6 @@ public class IoTDBDescriptor {
// author cache
loadAuthorCache(properties);
- // object storage
- loadMigrationProps(properties);
-
- // object storage
- loadObjectStorageProps(properties);
-
conf.setTimePartitionInterval(
DateTimeUtils.convertMilliTimeWithPrecision(
conf.getTimePartitionInterval(), conf.getTimestampPrecision()));
@@ -1157,63 +1146,6 @@ public class IoTDBDescriptor {
"author_cache_expire_time", String.valueOf(conf.getAuthorCacheExpireTime()))));
}
- private void loadMigrationProps(Properties properties) {
- conf.setMigrateThreadCount(
- Integer.parseInt(
- properties.getProperty(
- "migrate_thread_count", String.valueOf(conf.getMigrateThreadCount()))));
-
- String[] moveThresholdsStr = new String[conf.getSpaceMoveThresholds().length];
- for (int i = 0; i < moveThresholdsStr.length; ++i) {
- moveThresholdsStr[i] = String.valueOf(conf.getSpaceMoveThresholds()[i]);
- }
- moveThresholdsStr =
- properties
- .getProperty(
- "dn_default_space_move_thresholds",
- String.join(IoTDBConstant.TIER_SEPARATOR, moveThresholdsStr))
- .split(IoTDBConstant.TIER_SEPARATOR);
- double[] moveThresholds = new double[moveThresholdsStr.length];
- for (int i = 0; i < moveThresholds.length; ++i) {
- moveThresholds[i] = Double.parseDouble(moveThresholdsStr[i]);
- if (moveThresholds[i] < 0) {
- moveThresholds[i] = 0.15;
- }
- }
- conf.setSpaceMoveThresholds(moveThresholds);
- }
-
- private void loadObjectStorageProps(Properties properties) {
- conf.setEnableObjectStorage(
- Boolean.parseBoolean(
- properties.getProperty(
- "enable_object_storage", String.valueOf(conf.isEnableObjectStorage()))));
- conf.setObjectStorageName(
- properties.getProperty("object_storage_name", conf.getObjectStorageName()));
- conf.setObjectStorageBucket(
- properties.getProperty("object_storage_bucket", conf.getObjectStorageBucket()));
- conf.setObjectStorageEndpoint(
- properties.getProperty("object_storage_endpoint", conf.getObjectStorageEndpoint()));
- conf.setObjectStorageAccessKey(
- properties.getProperty("object_storage_access_key", conf.getObjectStorageAccessKey()));
- conf.setObjectStorageAccessSecret(
- properties.getProperty(
- "object_storage_access_secret", conf.getObjectStorageAccessSecret()));
- conf.setCacheDirs(
- properties
- .getProperty("remote_tsfile_cache_dirs", String.join(",", conf.getCacheDirs()))
- .split(","));
- conf.setCachePageSize(
- Integer.parseInt(
- properties.getProperty(
- "remote_tsfile_cache_page_size_in_kb", String.valueOf(conf.getCachePageSize()))));
- conf.setCacheMaxDiskUsage(
- Long.parseLong(
- properties.getProperty(
- "remote_tsfile_cache_max_disk_usage_in_mb",
- String.valueOf(conf.getCacheMaxDiskUsage()))));
- }
-
private void loadWALProps(Properties properties) {
conf.setWalMode(
WALMode.valueOf((properties.getProperty("wal_mode", conf.getWalMode().toString()))));
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/directories/TierManager.java b/server/src/main/java/org/apache/iotdb/db/conf/directories/TierManager.java
index db69bdb267d..9b8984e1f65 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/directories/TierManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/directories/TierManager.java
@@ -109,19 +109,6 @@ public class TierManager {
}
break;
case OBJECT_STORAGE:
- if (!config.isEnableObjectStorage()) {
- logger.error(
- "Cannot configure object storage directory when enable_object_storage is false, use default data dir instead.");
- tierDirs[i][j] =
- IoTDBConstant.DEFAULT_BASE_DIR + File.separator + IoTDBConstant.DATA_FOLDER_NAME;
- }
- if (i != tierDirs.length - 1) {
- logger.error("Object Storage can only exist on the last tier.");
- }
- // reset datanode id
- tierDirs[i][j] =
- FSUtils.getOSDefaultPath(config.getObjectStorageBucket(), config.getDataNodeId());
- break;
case HDFS:
default:
break;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/recover/CompactionRecoverManager.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/recover/CompactionRecoverManager.java
index 1ad685aac44..390ddf2b512 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/recover/CompactionRecoverManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/recover/CompactionRecoverManager.java
@@ -67,7 +67,6 @@ public class CompactionRecoverManager {
private void recoverCompaction(boolean isInnerSpace, boolean isLogSequence) {
List<String> dirs;
- // TODO(zhm) 当前远端存储介质不开启合并，无需恢复
if (isLogSequence) {
dirs = TierManager.getInstance().getAllLocalSequenceFileFolders();
} else {
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/recover/CompactionRecoverTask.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/recover/CompactionRecoverTask.java
index feeeddb2f13..b75940a7970 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/recover/CompactionRecoverTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/recover/CompactionRecoverTask.java
@@ -248,7 +248,6 @@ public class CompactionRecoverTask {
* the file is not found, it will return null.
*/
private File getFileFromDataDirs(String filePath) {
- // TODO(zhm) 当前远端存储介质不开启合并，无需恢复
String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getLocalDataDirs();
for (String dataDir : dataDirs) {
File f = new File(dataDir, filePath);
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/migration/LocalMigrationTask.java b/server/src/main/java/org/apache/iotdb/db/engine/migration/LocalMigrationTask.java
deleted file mode 100644
index 3e2ed543471..00000000000
--- a/server/src/main/java/org/apache/iotdb/db/engine/migration/LocalMigrationTask.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * 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.db.engine.migration;
-
-import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.TsFileResourceStatus;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.util.Arrays;
-
-public class LocalMigrationTask extends MigrationTask {
- private static final Logger logger = LoggerFactory.getLogger(LocalMigrationTask.class);
-
- protected LocalMigrationTask(MigrationCause cause, TsFileResource tsFile, String targetDir)
- throws IOException {
- super(cause, tsFile, targetDir);
- }
-
- @Override
- public void migrate() {
- // dest tsfile may exist if the last same migration task hasn't completed when the system
- // shutdown.
- filesShouldDelete.addAll(Arrays.asList(destTsFile, destResourceFile, destModsFile));
- cleanup();
-
- // copy TsFile and resource file
- tsFileResource.readLock();
- try {
- destTsFile.getParentFile().mkdirs();
- migrateFile(srcFile, destTsFile);
- migrateFile(srcResourceFile, destResourceFile);
- } catch (Exception e) {
- if (!tsFileResource.isDeleted()) {
- logger.error("Fail to copy TsFile from local {} to local {}", srcFile, srcResourceFile, e);
- }
- cleanup();
- return;
- } finally {
- tsFileResource.readUnlock();
- }
-
- // close mods file and replace TsFile path
- tsFileResource.writeLock();
- try {
- tsFileResource.resetModFile();
- // migrate MOD file only when it exists
- if (srcModsFile.exists()) {
- migrateFile(srcModsFile, destModsFile);
- }
- tsFileResource.setFile(destTsFile);
- tsFileResource.increaseTierLevel();
- tsFileResource.setStatus(TsFileResourceStatus.NORMAL);
- } catch (Exception e) {
- if (!tsFileResource.isDeleted()) {
- logger.error(
- "Fail to copy mods file from local {} to local {}", srcModsFile, destModsFile, e);
- }
- cleanup();
- return;
- } finally {
- tsFileResource.writeUnlock();
- }
-
- // clear src files
- filesShouldDelete.clear();
- filesShouldDelete.addAll(Arrays.asList(srcFile, srcResourceFile, srcModsFile));
- cleanup();
- }
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/migration/MigrationCause.java b/server/src/main/java/org/apache/iotdb/db/engine/migration/MigrationCause.java
deleted file mode 100644
index dc1d270ba61..00000000000
--- a/server/src/main/java/org/apache/iotdb/db/engine/migration/MigrationCause.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * 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.db.engine.migration;
-
-public enum MigrationCause {
- TTL,
- DISK_SPACE
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/migration/MigrationTask.java b/server/src/main/java/org/apache/iotdb/db/engine/migration/MigrationTask.java
deleted file mode 100644
index b97ef7acf35..00000000000
--- a/server/src/main/java/org/apache/iotdb/db/engine/migration/MigrationTask.java
+++ /dev/null
@@ -1,109 +0,0 @@
-/*
- * 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.db.engine.migration;
-
-import org.apache.iotdb.db.engine.modification.ModificationFile;
-import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.TsFileResourceStatus;
-import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
-import org.apache.iotdb.tsfile.fileSystem.fsFactory.FSFactory;
-import org.apache.iotdb.tsfile.utils.FSUtils;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-public abstract class MigrationTask implements Runnable {
- protected static final FSFactory fsFactory = FSFactoryProducer.getFSFactory();
-
- protected final MigrationCause cause;
- protected final TsFileResource tsFileResource;
- protected final String targetDir;
- protected final File srcFile;
- protected final File destTsFile;
- protected final File srcResourceFile;
- protected final File destResourceFile;
- protected final File srcModsFile;
- protected final File destModsFile;
-
- protected final List<File> filesShouldDelete = new ArrayList<>();
-
- protected MigrationTask(MigrationCause cause, TsFileResource tsFileResource, String targetDir)
- throws IOException {
- this.cause = cause;
- this.tsFileResource = tsFileResource;
- this.targetDir = targetDir;
- this.srcFile = tsFileResource.getTsFile();
- this.destTsFile = fsFactory.getFile(targetDir, getDestTsFilePath(srcFile));
- this.srcResourceFile =
- fsFactory.getFile(
- srcFile.getParentFile(), srcFile.getName() + TsFileResource.RESOURCE_SUFFIX);
- this.destResourceFile =
- fsFactory.getFile(targetDir, getDestTsFilePath(srcFile) + TsFileResource.RESOURCE_SUFFIX);
- this.srcModsFile =
- fsFactory.getFile(
- srcFile.getParentFile(), srcFile.getName() + ModificationFile.FILE_SUFFIX);
- this.destModsFile =
- fsFactory.getFile(targetDir, getDestTsFilePath(srcFile) + ModificationFile.FILE_SUFFIX);
- }
-
- private String getDestTsFilePath(File src) throws IOException {
- return FSUtils.getLocalTsFileShortPath(src, FSUtils.PATH_FROM_DATABASE_LEVEL);
- }
-
- public static MigrationTask newTask(
- MigrationCause cause, TsFileResource sourceTsFile, String targetDir) throws IOException {
- if (FSUtils.isLocal(targetDir)) {
- return new LocalMigrationTask(cause, sourceTsFile, targetDir);
- } else {
- return new RemoteMigrationTask(cause, sourceTsFile, targetDir);
- }
- }
-
- @Override
- public void run() {
- try {
- migrate();
- } finally {
- // try to set the final status to NORMAL to avoid migrate failure
- // TODO: this setting may occur side effects
- tsFileResource.setStatus(TsFileResourceStatus.NORMAL);
- MigrationTaskManager.getInstance().decreaseMigrationTasksNum();
- }
- }
-
- public abstract void migrate();
-
- protected void migrateFile(File src, File dest) throws IOException {
- fsFactory.copyFile(src, dest);
- filesShouldDelete.add(dest);
- }
-
- protected void cleanup() {
- filesShouldDelete.forEach(this::deleteIfExist);
- filesShouldDelete.clear();
- }
-
- protected void deleteIfExist(File file) {
- if (file.exists()) {
- file.delete();
- }
- }
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/migration/MigrationTaskManager.java b/server/src/main/java/org/apache/iotdb/db/engine/migration/MigrationTaskManager.java
deleted file mode 100644
index 2f9dbee2a64..00000000000
--- a/server/src/main/java/org/apache/iotdb/db/engine/migration/MigrationTaskManager.java
+++ /dev/null
@@ -1,211 +0,0 @@
-/*
- * 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.db.engine.migration;
-
-import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
-import org.apache.iotdb.commons.concurrent.ThreadName;
-import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
-import org.apache.iotdb.commons.conf.CommonConfig;
-import org.apache.iotdb.commons.conf.CommonDescriptor;
-import org.apache.iotdb.commons.exception.StartupException;
-import org.apache.iotdb.commons.service.IService;
-import org.apache.iotdb.commons.service.ServiceType;
-import org.apache.iotdb.db.conf.IoTDBConfig;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.TierManager;
-import org.apache.iotdb.db.engine.StorageEngine;
-import org.apache.iotdb.db.engine.storagegroup.DataRegion;
-import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.TsFileResourceStatus;
-import org.apache.iotdb.db.utils.DateTimeUtils;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
-public class MigrationTaskManager implements IService {
- private static final Logger logger = LoggerFactory.getLogger(MigrationTaskManager.class);
- private static final IoTDBConfig iotdbConfig = IoTDBDescriptor.getInstance().getConfig();
- private static final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig();
- private static final TierManager tierManager = TierManager.getInstance();
- private static final long CHECK_INTERVAL_IN_SECONDS = 10;
- private static final double MOVE_THRESHOLD_SAFE_LIMIT = 0.1;
- private static final int MIGRATION_TASK_LIMIT = 20;
- /** max concurrent migration tasks */
- private final AtomicInteger migrationTasksNum = new AtomicInteger(0);
- /** single thread to schedule */
- private ScheduledExecutorService scheduler;
- /** single thread to sync syncingBuffer to disk */
- private ExecutorService workers;
-
- @Override
- public void start() throws StartupException {
- if (iotdbConfig.getTierDataDirs().length == 1) {
- logger.info("tiered storage status: disable");
- return;
- }
- scheduler =
- IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
- ThreadName.MIGRATION_SCHEDULER.getName());
- workers =
- IoTDBThreadPoolFactory.newFixedThreadPool(
- iotdbConfig.getMigrateThreadCount(), ThreadName.MIGRATION.getName());
- ScheduledExecutorUtil.safelyScheduleAtFixedRate(
- scheduler,
- () -> new MigrationScheduleTask().run(),
- CHECK_INTERVAL_IN_SECONDS,
- CHECK_INTERVAL_IN_SECONDS,
- TimeUnit.SECONDS);
- }
-
- private class MigrationScheduleTask implements Runnable {
- private final long[] tierDiskTotalSpace = tierManager.getTierDiskTotalSpace();
- private final long[] tierDiskUsableSpace = tierManager.getTierDiskUsableSpace();
- private final Set<Integer> needMigrationTiers = new HashSet<>();
-
- public MigrationScheduleTask() {
- for (int i = 0; i < tierManager.getTiersNum(); i++) {
- double usage = tierDiskUsableSpace[i] * 1.0 / tierDiskTotalSpace[i];
- if (usage <= iotdbConfig.getSpaceMoveThresholds()[i]) {
- needMigrationTiers.add(i);
- }
- }
- }
-
- @Override
- public void run() {
- schedule();
- }
-
- private void schedule() {
- // submit migration tasks
- for (DataRegion dataRegion : StorageEngine.getInstance().getAllDataRegions()) {
- List<TsFileResource> tsfiles = dataRegion.getSequenceFileList();
- tsfiles.addAll(dataRegion.getUnSequenceFileList());
- tsfiles.sort(this::compareMigrationPriority);
- for (TsFileResource tsfile : tsfiles) {
- try {
- int currentTier = tsfile.getTierLevel();
- int nextTier = currentTier + 1;
- // only migrate closed TsFiles not in the last tier
- if (tsfile.getStatus() != TsFileResourceStatus.NORMAL
- || nextTier == iotdbConfig.getTierDataDirs().length) {
- continue;
- }
- // check tier ttl and disk space
- long tierTTL =
- DateTimeUtils.convertMilliTimeWithPrecision(
- System.currentTimeMillis() - commonConfig.getTierTTLInMs()[currentTier],
- iotdbConfig.getTimestampPrecision());
- if (!tsfile.stillLives(tierTTL)) {
- trySubmitMigrationTask(
- currentTier,
- MigrationCause.TTL,
- tsfile,
- tierManager.getNextFolderForTsFile(nextTier, tsfile.isSeq()));
- } else if (needMigrationTiers.contains(currentTier)) {
- trySubmitMigrationTask(
- currentTier,
- MigrationCause.DISK_SPACE,
- tsfile,
- tierManager.getNextFolderForTsFile(nextTier, tsfile.isSeq()));
- }
- } catch (Exception e) {
- logger.error(
- "An error occurred when check and try to migrate TsFileResource {}", tsfile, e);
- }
- }
- }
- }
-
- private void trySubmitMigrationTask(
- int tierLevel, MigrationCause cause, TsFileResource sourceTsFile, String targetDir)
- throws IOException {
- if (migrationTasksNum.get() >= MIGRATION_TASK_LIMIT
- || !sourceTsFile.setStatus(TsFileResourceStatus.MIGRATING)) {
- return;
- }
- migrationTasksNum.incrementAndGet();
- MigrationTask task = MigrationTask.newTask(cause, sourceTsFile, targetDir);
- workers.submit(task);
- tierDiskUsableSpace[tierLevel] -= sourceTsFile.getTsFileSize();
- if (needMigrationTiers.contains(tierLevel)) {
- double usage = tierDiskUsableSpace[tierLevel] * 1.0 / tierDiskTotalSpace[tierLevel];
- if (usage > iotdbConfig.getSpaceMoveThresholds()[tierLevel] + MOVE_THRESHOLD_SAFE_LIMIT) {
- needMigrationTiers.remove(tierLevel);
- }
- }
- }
-
- private int compareMigrationPriority(TsFileResource f1, TsFileResource f2) {
- // old time partitions first
- int res = Long.compare(f1.getTimePartition(), f2.getTimePartition());
- // sequence files in one partition
- if (res == 0) {
- if (f1.isSeq() && !f2.isSeq()) {
- res = -1;
- } else if (!f1.isSeq() && f2.isSeq()) {
- res = 1;
- }
- }
- // old version files in one partition
- if (res == 0) {
- res = Long.compare(f1.getVersion(), f2.getVersion());
- }
- return res;
- }
- }
-
- void decreaseMigrationTasksNum() {
- migrationTasksNum.decrementAndGet();
- }
-
- @Override
- public void stop() {
- if (scheduler != null) {
- scheduler.shutdownNow();
- }
- if (workers != null) {
- workers.shutdownNow();
- }
- }
-
- @Override
- public ServiceType getID() {
- return ServiceType.MIGRATION_SERVICE;
- }
-
- public static MigrationTaskManager getInstance() {
- return InstanceHolder.INSTANCE;
- }
-
- private static class InstanceHolder {
- private InstanceHolder() {}
-
- private static final MigrationTaskManager INSTANCE = new MigrationTaskManager();
- }
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/migration/RemoteMigrationTask.java b/server/src/main/java/org/apache/iotdb/db/engine/migration/RemoteMigrationTask.java
deleted file mode 100644
index 8daffc92776..00000000000
--- a/server/src/main/java/org/apache/iotdb/db/engine/migration/RemoteMigrationTask.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * 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.db.engine.migration;
-
-import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.TsFileResourceStatus;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.util.Arrays;
-
-public class RemoteMigrationTask extends MigrationTask {
- private static final Logger logger = LoggerFactory.getLogger(RemoteMigrationTask.class);
-
- protected RemoteMigrationTask(MigrationCause cause, TsFileResource tsFile, String targetDir)
- throws IOException {
- super(cause, tsFile, targetDir);
- }
-
- @Override
- public void migrate() {
- // dest tsfile may exist if the last same migration task hasn't completed when the system
- // shutdown.
- filesShouldDelete.addAll(Arrays.asList(destTsFile, destResourceFile));
- cleanup();
-
- // copy TsFile and resource file
- tsFileResource.readLock();
- try {
- migrateFile(srcFile, destTsFile);
- migrateFile(srcResourceFile, destResourceFile);
- } catch (Exception e) {
- if (!tsFileResource.isDeleted()) {
- logger.error("Fail to copy TsFile from local {} to remote {}", srcFile, srcResourceFile, e);
- }
- cleanup();
- return;
- } finally {
- tsFileResource.readUnlock();
- }
-
- // clear src files
- tsFileResource.writeLock();
- try {
- filesShouldDelete.clear();
- filesShouldDelete.add(srcFile);
- cleanup();
- tsFileResource.increaseTierLevel();
- tsFileResource.setStatus(TsFileResourceStatus.NORMAL_ON_REMOTE);
- } catch (Exception e) {
- if (!tsFileResource.isDeleted()) {
- logger.error("Fail to delete local TsFile {}", srcFile, e);
- }
- } finally {
- tsFileResource.writeUnlock();
- }
- }
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/service/DataNode.java b/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
index e173013f900..2a9d5778054 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
@@ -65,7 +65,6 @@ import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.cache.CacheHitRatioMonitor;
import org.apache.iotdb.db.engine.compaction.schedule.CompactionTaskManager;
import org.apache.iotdb.db.engine.flush.FlushManager;
-import org.apache.iotdb.db.engine.migration.MigrationTaskManager;
import org.apache.iotdb.db.metadata.schemaregion.SchemaEngine;
import org.apache.iotdb.db.metadata.template.ClusterTemplateManager;
import org.apache.iotdb.db.mpp.execution.exchange.MPPDataExchangeService;
@@ -84,9 +83,7 @@ import org.apache.iotdb.db.wal.WALManager;
import org.apache.iotdb.db.wal.utils.WALMode;
import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
import org.apache.iotdb.metrics.utils.InternalReporterType;
-import org.apache.iotdb.os.HybridFileInputFactoryDecorator;
import org.apache.iotdb.rpc.TSStatusCode;
-import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
import org.apache.iotdb.udf.api.exception.UDFManagementException;
import org.apache.thrift.TException;
@@ -171,7 +168,8 @@ public class DataNode implements DataNodeMBean {
// Send restart request of this DataNode
sendRestartRequestToConfigNode();
}
-
+ // TierManager need DataNodeId to do some operations so the reset method need to be invoked after DataNode adding
+ TierManager.getInstance().resetFolders();
// Active DataNode
active();
@@ -393,7 +391,6 @@ public class DataNode implements DataNodeMBean {
/* Store runtime configurations when register success */
int dataNodeID = dataNodeRegisterResp.getDataNodeId();
config.setDataNodeId(dataNodeID);
- configOSStorage(dataNodeID);
IoTDBStartCheck.getInstance()
.serializeClusterNameAndDataNodeId(config.getClusterName(), dataNodeID);
@@ -408,12 +405,6 @@ public class DataNode implements DataNodeMBean {
}
}
- private void configOSStorage(int dataNodeID) {
- TierManager.getInstance().resetFolders();
- FSFactoryProducer.setFileInputFactory(new HybridFileInputFactoryDecorator(dataNodeID));
- // recover OS cache
- }
-
private void sendRestartRequestToConfigNode() throws StartupException {
logger.info("Sending restart request to ConfigNode-leader...");
@@ -459,7 +450,6 @@ public class DataNode implements DataNodeMBean {
/* Store runtime configurations when restart request is accepted */
storeRuntimeConfigurations(
dataNodeRestartResp.getConfigNodeList(), dataNodeRestartResp.getRuntimeConfiguration());
- configOSStorage(config.getDataNodeId());
logger.info("Restart request to cluster: {} is accepted.", config.getClusterName());
} else {
/* Throw exception when restart is rejected */
@@ -553,8 +543,6 @@ public class DataNode implements DataNodeMBean {
registerManager.register(CompactionTaskManager.getInstance());
- registerManager.register(MigrationTaskManager.getInstance());
-
registerManager.register(PipeAgent.runtime());
}
diff --git a/server/src/main/java/org/apache/iotdb/db/service/metrics/SystemMetrics.java b/server/src/main/java/org/apache/iotdb/db/service/metrics/SystemMetrics.java
index cbf545adaf1..b2bdc7e2dbd 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/metrics/SystemMetrics.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/metrics/SystemMetrics.java
@@ -187,7 +187,6 @@ public class SystemMetrics implements IMetricSet {
}
private void collectSystemDiskInfo(AbstractMetricService metricService) {
- // TODO(zhm) 远端存储介质不支持 file store
String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getLocalDataDirs();
for (String dataDir : dataDirs) {
Path path = Paths.get(dataDir);