This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new e8061edf668 Optimize load logic (#18153)
e8061edf668 is described below
commit e8061edf668cc88cf7f951edce97f5ca907d79b9
Author: Zhenyu Luo <[email protected]>
AuthorDate: Fri Jul 10 10:33:34 2026 +0800
Optimize load logic (#18153)
* Fix load write permission check
* fix async
* Optimize load auth tests
* Optimize load user path encoding
* fix async
* Optimize load auth review fixes
* fix async
---
.../apache/iotdb/db/it/IoTDBLoadTsFileAuthIT.java | 285 +++++++++++++++++++++
.../iotdb/db/i18n/StorageEngineMessages.java | 3 +
.../iotdb/db/i18n/StorageEngineMessages.java | 3 +
.../protocol/thrift/IoTDBDataNodeReceiver.java | 3 +-
.../plan/analyze/load/LoadTsFileAnalyzer.java | 5 +-
.../load/TreeSchemaAutoCreatorAndVerifier.java | 72 +++++-
.../load/active/ActiveLoadPathHelper.java | 50 +++-
.../load/active/ActiveLoadTsFileLoader.java | 9 +
.../iotdb/db/storageengine/load/util/LoadUtil.java | 28 +-
.../load/active/ActiveLoadPathHelperTest.java | 146 +++++++++++
10 files changed, 580 insertions(+), 24 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileAuthIT.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileAuthIT.java
new file mode 100644
index 00000000000..d842941e0cc
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileAuthIT.java
@@ -0,0 +1,285 @@
+/*
+ * 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.it;
+
+import org.apache.iotdb.commons.auth.entity.PrivilegeType;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.it.utils.TsFileGenerator;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+import org.apache.iotdb.jdbc.IoTDBSQLException;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.external.commons.io.FileUtils;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.iotdb.db.it.utils.TestUtils.assertNonQueryTestFail;
+import static org.apache.iotdb.db.it.utils.TestUtils.createUser;
+import static org.apache.iotdb.db.it.utils.TestUtils.executeNonQuery;
+import static org.apache.iotdb.db.it.utils.TestUtils.grantUserSeriesPrivilege;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({LocalStandaloneIT.class, ClusterIT.class})
+public class IoTDBLoadTsFileAuthIT {
+ private static final long PARTITION_INTERVAL = 10 * 1000L;
+ private static final String DATABASE = "root.load_auth";
+ private static final String DEVICE = DATABASE + ".d1";
+ private static final IMeasurementSchema MEASUREMENT =
+ new MeasurementSchema("s1", TSDataType.INT32, TSEncoding.RLE);
+ private static final String NO_WRITE_USER = "load_no_write_user";
+ private static final String WRITE_USER = "load_write_user";
+ private static final String OTHER_PATH_WRITE_USER =
"load_other_path_write_user";
+ private static final String ASYNC_NO_WRITE_USER = "async_load_no_write_user";
+ private static final String ASYNC_WRITE_USER = "async_load_write_user";
+ private static final String PASSWORD = "test123123456";
+ private static final long
UNALLOCATABLE_TABLET_CONVERSION_BATCH_MEMORY_SIZE_IN_BYTES =
+ Long.MAX_VALUE / 4;
+
+ private static File tmpDir;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ tmpDir = new File(Files.createTempDirectory("load-auth").toUri());
+
EnvFactory.getEnv().getConfig().getCommonConfig().setTimePartitionInterval(PARTITION_INTERVAL);
+
EnvFactory.getEnv().getConfig().getCommonConfig().setEnforceStrongPassword(false);
+
EnvFactory.getEnv().getConfig().getCommonConfig().setAutoCreateSchemaEnabled(false);
+ EnvFactory.getEnv()
+ .getConfig()
+ .getDataNodeConfig()
+ .setMaxAllocateMemoryRatioForLoad(1.0)
+ .setLoadTsFileAnalyzeSchemaMemorySizeInBytes(10 * 1024L)
+ .setLoadTsFileTabletConversionBatchMemorySizeInBytes(
+ UNALLOCATABLE_TABLET_CONVERSION_BATCH_MEMORY_SIZE_IN_BYTES)
+ .setLoadActiveListeningCheckIntervalSeconds(1);
+
+ EnvFactory.getEnv().initClusterEnvironment();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ deleteDatabase();
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ FileUtils.deleteDirectory(tmpDir);
+ }
+
+ @After
+ public void cleanData() throws Exception {
+ deleteDatabase();
+ }
+
+ @Test
+ public void testLoadWithoutSchemaCheckStillChecksWriteDataPermission()
throws Exception {
+ final File tsFile = new File(tmpDir, "1-0-0-0.tsfile");
+ prepareSchemaAndTsFile(tsFile);
+ createUser(NO_WRITE_USER, PASSWORD);
+
+ assertNonQueryTestFail(
+ String.format("load \"%s\" with ('database-level'='2',
'verify'='false')", tsFile),
+ "No permissions for this operation, please add privilege WRITE_DATA",
+ NO_WRITE_USER,
+ PASSWORD);
+ }
+
+ @Test
+ public void testLoadWithoutSchemaCheckAllowsUserWithWriteDataPermission()
throws Exception {
+ final File tsFile = new File(tmpDir, "2-0-0-0.tsfile");
+ prepareSchemaAndTsFile(tsFile);
+ createUser(WRITE_USER, PASSWORD);
+ grantUserSeriesPrivilege(WRITE_USER, PrivilegeType.WRITE_DATA, DATABASE +
".**");
+
+ executeNonQuery(
+ String.format("load \"%s\" with ('database-level'='2',
'verify'='false')", tsFile),
+ WRITE_USER,
+ PASSWORD);
+
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement();
+ final ResultSet resultSet = statement.executeQuery("select count(s1)
from " + DEVICE)) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals(10, resultSet.getLong(1));
+ }
+ }
+
+ @Test
+ public void
testLoadWithoutSchemaCheckRejectsUserWithOtherPathWriteDataPermission()
+ throws Exception {
+ final File tsFile = new File(tmpDir, "3-0-0-0.tsfile");
+ prepareSchemaAndTsFile(tsFile);
+ createUser(OTHER_PATH_WRITE_USER, PASSWORD);
+ grantUserSeriesPrivilege(OTHER_PATH_WRITE_USER, PrivilegeType.WRITE_DATA,
"root.other.**");
+
+ assertNonQueryTestFail(
+ String.format("load \"%s\" with ('database-level'='2',
'verify'='false')", tsFile),
+ "No permissions for this operation, please add privilege WRITE_DATA",
+ OTHER_PATH_WRITE_USER,
+ PASSWORD);
+ }
+
+ @Test
+ public void testAsyncLoadShouldCheckWriteDataPermissionWithStoredUser()
throws Exception {
+ final File noWriteTsFile = new File(tmpDir, "4-0-0-0.tsfile");
+ final File writeTsFile = new File(tmpDir, "5-0-0-0.tsfile");
+ prepareSchemaAndTsFile(noWriteTsFile);
+ generateTsFile(writeTsFile);
+ createUser(ASYNC_NO_WRITE_USER, PASSWORD);
+ createUser(ASYNC_WRITE_USER, PASSWORD);
+ grantUserSeriesPrivilege(ASYNC_WRITE_USER, PrivilegeType.WRITE_DATA,
DATABASE + ".**");
+
+ executeNonQuery(
+ String.format(
+ "load \"%s\" with ('database-level'='2', 'async'='true',
'on-success'='none', "
+ + "'verify'='false')",
+ noWriteTsFile.getAbsolutePath()),
+ ASYNC_NO_WRITE_USER,
+ PASSWORD);
+ executeNonQuery(
+ String.format(
+ "load \"%s\" with ('database-level'='2', 'async'='true',
'on-success'='none', "
+ + "'verify'='false')",
+ writeTsFile.getAbsolutePath()),
+ ASYNC_WRITE_USER,
+ PASSWORD);
+
+ waitUntilAllActiveLoadPendingDirsAreEmpty(TimeUnit.SECONDS.toMillis(60));
+ assertCountEventually(10, TimeUnit.SECONDS.toMillis(60));
+ }
+
+ private static void prepareSchemaAndTsFile(final File tsFile) throws
Exception {
+ prepareSchema(MEASUREMENT.getType());
+ generateTsFile(tsFile);
+ }
+
+ private static void prepareSchema(final TSDataType dataType) throws
Exception {
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement()) {
+ statement.execute("create database " + DATABASE);
+ statement.execute(
+ String.format(
+ "create timeseries %s.%s %s", DEVICE,
MEASUREMENT.getMeasurementName(), dataType));
+ }
+ }
+
+ private static void generateTsFile(final File tsFile) throws Exception {
+ try (final TsFileGenerator generator = new TsFileGenerator(tsFile)) {
+ generator.registerTimeseries(DEVICE,
Collections.singletonList(MEASUREMENT));
+ generator.generateData(DEVICE, 10, PARTITION_INTERVAL / 10, false);
+ }
+ }
+
+ private static void deleteDatabase() throws Exception {
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement()) {
+ statement.execute("delete database " + DATABASE);
+ } catch (final IoTDBSQLException ignored) {
+ }
+ }
+
+ private File getActiveLoadPendingDir(final DataNodeWrapper dataNodeWrapper) {
+ return new File(
+ dataNodeWrapper.getNodePath()
+ + File.separator
+ + "ext"
+ + File.separator
+ + "load"
+ + File.separator
+ + "pending");
+ }
+
+ private void waitUntilAllActiveLoadPendingDirsAreEmpty(final long timeoutMs)
+ throws InterruptedException {
+ final long deadline = System.currentTimeMillis() + timeoutMs;
+ while (System.currentTimeMillis() < deadline) {
+ boolean hasTsFile = false;
+ for (final DataNodeWrapper dataNodeWrapper :
EnvFactory.getEnv().getDataNodeWrapperList()) {
+ if (containsTsFile(getActiveLoadPendingDir(dataNodeWrapper))) {
+ hasTsFile = true;
+ break;
+ }
+ }
+ if (!hasTsFile) {
+ return;
+ }
+ Thread.sleep(500L);
+ }
+ Assert.fail("Timed out waiting for active load pending dirs to become
empty");
+ }
+
+ private void assertCountEventually(final long expected, final long
timeoutMs) throws Exception {
+ final long deadline = System.currentTimeMillis() + timeoutMs;
+ AssertionError lastError = null;
+ while (System.currentTimeMillis() < deadline) {
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement();
+ final ResultSet resultSet =
+ statement.executeQuery(
+ "select count(" + MEASUREMENT.getMeasurementName() + ") from
" + DEVICE)) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals(expected, resultSet.getLong(1));
+ return;
+ } catch (final AssertionError e) {
+ lastError = e;
+ }
+ Thread.sleep(500L);
+ }
+ if (lastError != null) {
+ throw lastError;
+ }
+ Assert.fail("Timed out waiting for count " + expected);
+ }
+
+ private boolean containsTsFile(final File root) {
+ if (root == null || !root.exists()) {
+ return false;
+ }
+ if (root.isFile()) {
+ return root.getName().endsWith(".tsfile");
+ }
+
+ final File[] children = root.listFiles();
+ if (children == null) {
+ return false;
+ }
+ for (final File child : children) {
+ if (containsTsFile(child)) {
+ return true;
+ }
+ }
+ return false;
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java
index f8b6e50ed53..512879bfd3c 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java
@@ -538,11 +538,14 @@ public final class StorageEngineMessages {
public static final String FAILED_COUNT_ACTIVE_DIRS_FILE_NUMBER = "Failed to
count active listening dirs file number.";
public static final String ACTIVE_LOAD_METRIC_COLLECTOR_REGISTERED = "Active
load metric collector periodical jobs registered";
public static final String DATABASE_NAME_MUST_NOT_BE_EMPTY = "Database name
must not be empty.";
+ public static final String USER_NAME_MUST_NOT_BE_EMPTY = "User name must not
be empty.";
public static final String ERROR_EXECUTING_ACTIVE_LOAD_JOB = "Error occurred
when executing active load periodical job.";
public static final String ACTIVE_LOAD_EXECUTOR_STARTED = "Active load
periodical jobs executor is started successfully.";
public static final String ACTIVE_LOAD_EXECUTOR_STOPPED = "Active load
periodical jobs executor is stopped successfully.";
public static final String ACTIVE_LOAD_TEMPORARILY_UNAVAILABLE =
"Rejecting auto load tsfile {} (isGeneratedByPipe = {}) due to temporary
unavailability, will retry later. Status: {}";
+ public static final String USER_IN_ACTIVE_LOAD_PATH_DOES_NOT_EXIST =
+ "The user in the active load path does not exist";
public static final String ERROR_MOVING_FILE_TO_FAIL_DIR = "Error occurred
during moving file {} to fail directory.";
public static final String FAILED_COUNT_FILES_IN_FAIL_DIR = "Failed to count
failed files in fail directory.";
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java
index bffda3d5287..96d09c1c141 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java
@@ -538,11 +538,14 @@ public final class StorageEngineMessages {
public static final String FAILED_COUNT_ACTIVE_DIRS_FILE_NUMBER = "统计 Active
监听目录文件数量失败。";
public static final String ACTIVE_LOAD_METRIC_COLLECTOR_REGISTERED = "Active
加载指标收集定期任务已注册";
public static final String DATABASE_NAME_MUST_NOT_BE_EMPTY = "数据库名称不能为空。";
+ public static final String USER_NAME_MUST_NOT_BE_EMPTY = "用户名不能为空。";
public static final String ERROR_EXECUTING_ACTIVE_LOAD_JOB = "执行 Active
加载定期任务时发生错误。";
public static final String ACTIVE_LOAD_EXECUTOR_STARTED = "Active
加载定期任务执行器已成功启动。";
public static final String ACTIVE_LOAD_EXECUTOR_STOPPED = "Active
加载定期任务执行器已成功停止。";
public static final String ACTIVE_LOAD_TEMPORARILY_UNAVAILABLE =
"拒绝自动加载 TsFile {} (isGeneratedByPipe = {})，原因是系统暂时不可用，将稍后重试。状态: {}";
+ public static final String USER_IN_ACTIVE_LOAD_PATH_DOES_NOT_EXIST =
+ "Active 加载路径中的用户不存在";
public static final String ERROR_MOVING_FILE_TO_FAIL_DIR = "将文件 {}
移动到失败目录时发生错误。";
public static final String FAILED_COUNT_FILES_IN_FAIL_DIR =
"统计失败目录中的失败文件数量失败。";
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
index 3048ad13084..81879355b62 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
@@ -656,7 +656,8 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
shouldConvertDataTypeOnTypeMismatch,
validateTsFile || shouldConvertDataTypeOnTypeMismatch,
null,
- shouldMarkAsPipeRequest);
+ shouldMarkAsPipeRequest,
+ AuthorityChecker.SUPER_USER);
}
private TSStatus loadTsFileSync(final String dataBaseName, final String
fileAbsolutePath)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
index 158f38875e1..48d39165840 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
@@ -299,7 +299,8 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
isConvertOnTypeMismatch,
isVerifySchema,
tabletConversionThresholdBytes,
- isGeneratedByPipe);
+ isGeneratedByPipe,
+ Objects.nonNull(context) ? context.getUsername() : null);
if (LoadUtil.loadTsFileAsyncToActiveDir(tsFiles, activeLoadAttributes,
isDeleteAfterLoad)) {
analysis.setFinishQueryAfterAnalyze(true);
@@ -551,6 +552,8 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
if (isAutoCreateSchemaOrVerifySchemaEnabled) {
getOrCreateTreeSchemaVerifier().autoCreateAndVerify(reader,
device2TimeseriesMetadata);
+ } else {
+
getOrCreateTreeSchemaVerifier().checkWritePermission(device2TimeseriesMetadata);
}
// TODO: how to get the correct write point count when
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java
index a5e8f2d1c3c..6d1704c0cfd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java
@@ -154,20 +154,7 @@ public class TreeSchemaAutoCreatorAndVerifier {
// not a timeseries, skip
} else {
// check WRITE_DATA permission of timeseries
- long startTime = System.nanoTime();
- try {
- UserEntity userEntity =
loadTsFileAnalyzer.context.getSession().getUserEntity();
- TSStatus status =
- AuthorityChecker.getAccessControl()
- .checkFullPathWriteDataPermission(
- userEntity, device,
timeseriesMetadata.getMeasurementId());
- if (status.getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- throw new AuthException(
- TSStatusCode.representOf(status.getCode()),
status.getMessage());
- }
- } finally {
-
PerformanceOverviewMetrics.getInstance().recordAuthCost(System.nanoTime() -
startTime);
- }
+ checkWritePermission(device, timeseriesMetadata.getMeasurementId());
final Pair<CompressionType, TSEncoding> compressionEncodingPair =
reader.readTimeseriesCompressionTypeAndEncoding(timeseriesMetadata);
schemaCache.addTimeSeries(
@@ -191,6 +178,63 @@ public class TreeSchemaAutoCreatorAndVerifier {
}
}
+ public void checkWritePermission(
+ Map<IDeviceID, List<TimeseriesMetadata>> device2TimeseriesMetadataList)
throws AuthException {
+ for (final Map.Entry<IDeviceID, List<TimeseriesMetadata>> entry :
+ device2TimeseriesMetadataList.entrySet()) {
+ final IDeviceID device = entry.getKey();
+
+ try {
+ if (schemaCache.isDeviceDeletedByMods(device)) {
+ continue;
+ }
+ } catch (IllegalPathException e) {
+ LOGGER.warn(
+ DataNodeQueryMessages
+
.FAILED_TO_CHECK_IF_DEVICE_ARG_IS_DELETED_BY_MODS_WILL_SEE_IT_AS_NOT_DELETED,
+ device,
+ e);
+ }
+
+ for (final TimeseriesMetadata timeseriesMetadata : entry.getValue()) {
+ try {
+ if (schemaCache.isTimeSeriesDeletedByMods(device,
timeseriesMetadata)) {
+ continue;
+ }
+ } catch (IllegalPathException e) {
+ if (!timeseriesMetadata.getMeasurementId().isEmpty()) {
+ LOGGER.warn(
+ DataNodeQueryMessages
+
.FAILED_TO_CHECK_IF_DEVICE_ARG_TIMESERIES_ARG_IS_DELETED_BY_MODS_WILL_SEE_IT_AS_NOT,
+ device,
+ timeseriesMetadata.getMeasurementId(),
+ e);
+ }
+ }
+
+ if (!TSDataType.VECTOR.equals(timeseriesMetadata.getTsDataType())) {
+ checkWritePermission(device, timeseriesMetadata.getMeasurementId());
+ }
+ }
+ }
+ }
+
+ private void checkWritePermission(final IDeviceID device, final String
measurementId)
+ throws AuthException {
+ final long startTime = System.nanoTime();
+ try {
+ UserEntity userEntity =
loadTsFileAnalyzer.context.getSession().getUserEntity();
+ TSStatus status =
+ AuthorityChecker.getAccessControl()
+ .checkFullPathWriteDataPermission(userEntity, device,
measurementId);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new AuthException(TSStatusCode.representOf(status.getCode()),
status.getMessage());
+ }
+ } finally {
+
PerformanceOverviewMetrics.getInstance().recordAuthCost(System.nanoTime() -
startTime);
+ }
+ }
+
/**
* This can only be invoked after all timeseries in the current tsfile have
been processed.
* Otherwise, the isAligned status may be wrong.
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelper.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelper.java
index 7b971c15a10..991d368d395 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelper.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelper.java
@@ -25,6 +25,8 @@ import org.apache.iotdb.db.i18n.StorageEngineMessages;
import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
import org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator;
+import com.google.common.io.BaseEncoding;
+
import java.io.File;
import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
@@ -43,10 +45,15 @@ import java.util.Optional;
public final class ActiveLoadPathHelper {
private static final String SEGMENT_SEPARATOR = "-";
+ public static final String USER_KEY = "user";
+ // Keep a version in the user path segment so future encryption algorithms
can be added safely.
+ private static final String USER_VALUE_MASK_PREFIX = "v1-";
+ private static final BaseEncoding USER_VALUE_ENCODING =
BaseEncoding.base32().omitPadding();
private static final List<String> KEY_ORDER =
Collections.unmodifiableList(
Arrays.asList(
+ USER_KEY,
LoadTsFileConfigurator.DATABASE_NAME_KEY,
LoadTsFileConfigurator.DATABASE_LEVEL_KEY,
LoadTsFileConfigurator.CONVERT_ON_TYPE_MISMATCH_KEY,
@@ -65,8 +72,13 @@ public final class ActiveLoadPathHelper {
final Boolean convertOnTypeMismatch,
final Boolean verify,
final Long tabletConversionThresholdBytes,
- final Boolean pipeGenerated) {
+ final Boolean pipeGenerated,
+ final String userName) {
final Map<String, String> attributes = new LinkedHashMap<>();
+ if (Objects.nonNull(userName) && !userName.isEmpty()) {
+ attributes.put(USER_KEY, userName);
+ }
+
if (Objects.nonNull(databaseName) && !databaseName.isEmpty()) {
attributes.put(LoadTsFileConfigurator.DATABASE_NAME_KEY, databaseName);
}
@@ -208,7 +220,15 @@ public final class ActiveLoadPathHelper {
}
private static String formatSegment(final String key, final String value) {
- return key + SEGMENT_SEPARATOR + encodeValue(value);
+ return key + SEGMENT_SEPARATOR + encodeValue(maskValueIfNecessary(key,
value));
+ }
+
+ private static String maskValueIfNecessary(final String key, final String
value) {
+ if (!USER_KEY.equals(key)) {
+ return value;
+ }
+ return USER_VALUE_MASK_PREFIX
+ + USER_VALUE_ENCODING.encode(value.getBytes(StandardCharsets.UTF_8));
}
private static String encodeValue(final String value) {
@@ -227,7 +247,11 @@ public final class ActiveLoadPathHelper {
}
final String encodedValue = dirName.substring(prefixLength);
- final String decodedValue = decodeValue(encodedValue);
+ final String rawDecodedValue = decodeValue(encodedValue);
+ if (USER_KEY.equals(key) &&
!rawDecodedValue.startsWith(USER_VALUE_MASK_PREFIX)) {
+ return Optional.empty();
+ }
+ final String decodedValue = unmaskValueIfNecessary(key, rawDecodedValue);
try {
validateAttributeValue(key, decodedValue);
return Optional.of(decodedValue);
@@ -255,6 +279,11 @@ public final class ActiveLoadPathHelper {
case LoadTsFileConfigurator.VERIFY_KEY:
LoadTsFileConfigurator.validateVerifyParam(value);
break;
+ case USER_KEY:
+ if (value == null || value.isEmpty()) {
+ throw new
SemanticException(StorageEngineMessages.USER_NAME_MUST_NOT_BE_EMPTY);
+ }
+ break;
default:
LoadTsFileConfigurator.validateParameters(key, value);
}
@@ -279,4 +308,19 @@ public final class ActiveLoadPathHelper {
return value;
}
}
+
+ private static String unmaskValueIfNecessary(final String key, final String
value) {
+ if (!USER_KEY.equals(key) || !value.startsWith(USER_VALUE_MASK_PREFIX)) {
+ return value;
+ }
+ return decodeUserName(value.substring(USER_VALUE_MASK_PREFIX.length()),
value);
+ }
+
+ private static String decodeUserName(final String encodedUserName, final
String fallback) {
+ try {
+ return new String(USER_VALUE_ENCODING.decode(encodedUserName),
StandardCharsets.UTF_8);
+ } catch (final IllegalArgumentException e) {
+ return fallback;
+ }
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java
index 5e6cade3087..ff5222e5754 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java
@@ -261,6 +261,15 @@ public class ActiveLoadTsFileLoader {
: new File(entry.getPendingDir());
final Map<String, String> attributes =
ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir);
ActiveLoadPathHelper.applyAttributesToStatement(attributes, statement,
isVerify);
+ final String userName =
+ attributes.getOrDefault(ActiveLoadPathHelper.USER_KEY,
AuthorityChecker.SUPER_USER);
+ final Optional<Long> userId = AuthorityChecker.getUserId(userName);
+ if (!userId.isPresent()) {
+ return new TSStatus(TSStatusCode.USER_NOT_EXIST.getStatusCode())
+
.setMessage(StorageEngineMessages.USER_IN_ACTIVE_LOAD_PATH_DOES_NOT_EXIST);
+ }
+ session.setUserId(userId.get());
+ session.setUsername(userName);
final File parentFile;
if (statement.getDatabase() == null && entry.isTableModel()) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java
index cafb1b0e700..88c71fb1cef 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java
@@ -23,8 +23,11 @@ import org.apache.iotdb.commons.disk.FolderManager;
import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType;
import org.apache.iotdb.commons.exception.DiskSpaceInsufficientException;
import org.apache.iotdb.commons.utils.RetryUtils;
+import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.i18n.StorageEngineMessages;
+import org.apache.iotdb.db.protocol.session.IClientSession;
+import org.apache.iotdb.db.protocol.session.SessionManager;
import
org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
import
org.apache.iotdb.db.storageengine.dataregion.modification.v1.ModificationFileV1;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
@@ -37,7 +40,7 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.util.Arrays;
-import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -124,8 +127,7 @@ public class LoadUtil {
LOGGER.warn(StorageEngineMessages.LOAD_ACTIVE_LISTENING_DIR_NOT_SET);
return false;
}
- final Map<String, String> attributes =
- Objects.nonNull(loadAttributes) ? loadAttributes :
Collections.emptyMap();
+ final Map<String, String> attributes =
appendCurrentUserIfAbsent(loadAttributes);
final File targetDir =
ActiveLoadPathHelper.resolveTargetDir(targetFilePath, attributes);
loadTsFileAsyncToTargetDir(
@@ -138,6 +140,23 @@ public class LoadUtil {
return true;
}
+ private static Map<String, String> appendCurrentUserIfAbsent(
+ final Map<String, String> loadAttributes) {
+ final Map<String, String> attributes =
+ Objects.nonNull(loadAttributes)
+ ? new LinkedHashMap<>(loadAttributes)
+ : new LinkedHashMap<>();
+ if (!attributes.containsKey(ActiveLoadPathHelper.USER_KEY)) {
+ final IClientSession session =
SessionManager.getInstance().getCurrSession();
+ attributes.put(
+ ActiveLoadPathHelper.USER_KEY,
+ session == null || session.getUsername() == null
+ ? AuthorityChecker.SUPER_USER
+ : session.getUsername());
+ }
+ return attributes;
+ }
+
public static boolean loadFilesToActiveDir(
final Map<String, String> loadAttributes,
final List<String> files,
@@ -161,8 +180,7 @@ public class LoadUtil {
LOGGER.warn(StorageEngineMessages.LOAD_ACTIVE_LISTENING_DIR_NOT_SET);
return false;
}
- final Map<String, String> attributes =
- Objects.nonNull(loadAttributes) ? loadAttributes :
Collections.emptyMap();
+ final Map<String, String> attributes =
appendCurrentUserIfAbsent(loadAttributes);
final File targetDir =
ActiveLoadPathHelper.resolveTargetDir(targetFilePath, attributes);
for (final String file : files) {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelperTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelperTest.java
new file mode 100644
index 00000000000..77936a653c7
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelperTest.java
@@ -0,0 +1,146 @@
+/*
+ * 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.storageengine.load.active;
+
+import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
+import org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.util.Map;
+
+public class ActiveLoadPathHelperTest {
+
+ @Test
+ public void testUserAttributeShouldBeMaskedInPathAndDecodedWhenParsing()
throws Exception {
+ final String userName = "active_load_user";
+ final File pendingDir =
Files.createTempDirectory("active-load-path").toFile();
+ try {
+ final File targetDir =
+ ActiveLoadPathHelper.resolveTargetDir(
+ pendingDir,
+ ActiveLoadPathHelper.buildAttributes(null, null, null, null,
null, null, userName));
+ final File tsFile = new File(targetDir, "1-0-0-0.tsfile");
+
+ Assert.assertTrue(targetDir.getAbsolutePath().contains("user-v1-"));
+ Assert.assertFalse(targetDir.getAbsolutePath().contains(userName));
+ Assert.assertFalse(targetDir.getAbsolutePath().contains("b64%3A"));
+
+ final Map<String, String> attributes =
+ ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir);
+ Assert.assertEquals(userName,
attributes.get(ActiveLoadPathHelper.USER_KEY));
+ } finally {
+ deleteRecursively(pendingDir);
+ }
+ }
+
+ @Test
+ public void testRawUserAttributeShouldBeIgnored() throws Exception {
+ final File pendingDir =
Files.createTempDirectory("active-load-path").toFile();
+ try {
+ final File tsFile = new File(new File(pendingDir,
"user-active_load_user"), "1-0-0-0.tsfile");
+
+ final Map<String, String> attributes =
+ ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir);
+
Assert.assertFalse(attributes.containsKey(ActiveLoadPathHelper.USER_KEY));
+ } finally {
+ deleteRecursively(pendingDir);
+ }
+ }
+
+ @Test
+ public void testNonV1UserAttributeShouldBeIgnored() throws Exception {
+ final File pendingDir =
Files.createTempDirectory("active-load-path").toFile();
+ try {
+ final File tsFile =
+ new File(new File(pendingDir, "user-v2-active_load_user"),
"1-0-0-0.tsfile");
+
+ final Map<String, String> attributes =
+ ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir);
+
Assert.assertFalse(attributes.containsKey(ActiveLoadPathHelper.USER_KEY));
+ } finally {
+ deleteRecursively(pendingDir);
+ }
+ }
+
+ @Test
+ public void
testUnknownRawAttributeDirectoryShouldBeIgnoredForDowngradeCompatibility()
+ throws Exception {
+ final File pendingDir =
Files.createTempDirectory("active-load-path").toFile();
+ try {
+ final File tsFile =
+ new File(new File(pendingDir, "future-load-param-future-value"),
"1-0-0-0.tsfile");
+ createFile(tsFile);
+
+ final Map<String, String> attributes =
+ ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir);
+ Assert.assertFalse(attributes.containsKey("future-load-param"));
+
+ final LoadTsFileStatement statement =
+ LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath());
+ ActiveLoadPathHelper.applyAttributesToStatement(attributes, statement,
true);
+ Assert.assertTrue(statement.isVerifySchema());
+ } finally {
+ deleteRecursively(pendingDir);
+ }
+ }
+
+ @Test
+ public void
testKnownPrefixWithInvalidFutureLikeValueShouldBeIgnoredForDowngradeCompatibility()
+ throws Exception {
+ final File pendingDir =
Files.createTempDirectory("active-load-path").toFile();
+ try {
+ final File tsFile = new File(new File(pendingDir,
"verify-future-value"), "1-0-0-0.tsfile");
+ createFile(tsFile);
+
+ final Map<String, String> attributes =
+ ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir);
+
Assert.assertFalse(attributes.containsKey(LoadTsFileConfigurator.VERIFY_KEY));
+
+ final LoadTsFileStatement statement =
+ LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath());
+ ActiveLoadPathHelper.applyAttributesToStatement(attributes, statement,
true);
+ Assert.assertTrue(statement.isVerifySchema());
+ } finally {
+ deleteRecursively(pendingDir);
+ }
+ }
+
+ private static void deleteRecursively(final File file) {
+ if (file == null || !file.exists()) {
+ return;
+ }
+ final File[] children = file.listFiles();
+ if (children != null) {
+ for (final File child : children) {
+ deleteRecursively(child);
+ }
+ }
+ Assert.assertTrue(file.delete());
+ }
+
+ private static void createFile(final File file) throws Exception {
+ Assert.assertTrue(file.getParentFile().mkdirs());
+ Assert.assertTrue(file.createNewFile());
+ }
+}
