This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 5a58b8ba427 [to dev/1.3] Optimize load logic (#18153) (#18459)
5a58b8ba427 is described below
commit 5a58b8ba4272d618dd44f8876ef21af378f5f657
Author: Zhenyu Luo <[email protected]>
AuthorDate: Thu Aug 13 13:58:16 2026 +0800
[to dev/1.3] Optimize load logic (#18153) (#18459)
* 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
* fix(load): adapt load auth IT to branch tsfile/IT APIs
---
.../apache/iotdb/db/it/IoTDBLoadTsFileAuthIT.java | 283 +++++++++++++++++++++
.../protocol/thrift/IoTDBDataNodeReceiver.java | 3 +-
.../plan/analyze/load/LoadTsFileAnalyzer.java | 104 ++++++--
.../load/active/ActiveLoadPathHelper.java | 55 +++-
.../load/active/ActiveLoadTsFileLoader.java | 6 +-
.../storageengine/load/active/ActiveLoadUtil.java | 28 +-
.../load/active/ActiveLoadPathHelperTest.java | 103 +++++++-
7 files changed, 540 insertions(+), 42 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..09773e4533f
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileAuthIT.java
@@ -0,0 +1,283 @@
+/*
+ * 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.commons.io.FileUtils;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+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 MeasurementSchema 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().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.getMeasurementId(), 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.getMeasurementId() + ") 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/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 92ff5ca2de5..b7da3775787 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
@@ -571,7 +571,8 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
validateTsFile || shouldConvertDataTypeOnTypeMismatch ||
shouldWaitForSchemaBeforeLoad,
!shouldWaitForSchemaBeforeLoad,
null,
- shouldMarkAsPipeRequest);
+ shouldMarkAsPipeRequest,
+ AuthorityChecker.SUPER_USER);
}
private TSStatus loadTsFileSync(
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 227064b286a..a4a23478612 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
@@ -271,7 +271,8 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
isVerifySchema,
isAutoCreateSchemaRequested,
tabletConversionThresholdBytes,
- isGeneratedByPipe);
+ isGeneratedByPipe,
+ context.getSession().getUserName());
if (ActiveLoadUtil.loadTsFileAsyncToActiveDir(
tsFiles, activeLoadAttributes, isDeleteAfterLoad)) {
analysis.setFinishQueryAfterAnalyze(true);
@@ -465,6 +466,8 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
if (isAutoCreateSchemaOrVerifySchemaEnabled) {
schemaAutoCreatorAndVerifier.autoCreateAndVerify(reader,
device2TimeseriesMetadata);
+ } else {
+
schemaAutoCreatorAndVerifier.checkWritePermission(device2TimeseriesMetadata);
}
// TODO: how to get the correct write point count when
// !isAutoCreateSchemaOrVerifySchemaEnabled
@@ -673,33 +676,7 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
// not a timeseries, skip
} else {
// check WRITE_DATA permission of timeseries
- long startTime = System.nanoTime();
- try {
- String userName = context.getSession().getUserName();
- if (!AuthorityChecker.SUPER_USER.equals(userName)) {
- TSStatus status;
- try {
- List<PartialPath> paths =
- Collections.singletonList(
- new PartialPath(device,
timeseriesMetadata.getMeasurementId()));
- status =
- AuthorityChecker.getTSStatus(
- AuthorityChecker.checkFullPathListPermission(
- userName, paths,
PrivilegeType.WRITE_DATA.ordinal()),
- paths,
- PrivilegeType.WRITE_DATA);
- } catch (IllegalPathException e) {
- throw new RuntimeException(e);
- }
- 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(
@@ -723,6 +700,77 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
}
}
+ 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(
+ "Failed to check if device {} 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) {
+ // In aligned devices, there may be empty measurements which will
cause
+ // IllegalPathException.
+ if (!timeseriesMetadata.getMeasurementId().isEmpty()) {
+ LOGGER.warn(
+ "Failed to check if device {}, timeSeries {} is deleted by
mods. Will see it as not deleted.",
+ 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 {
+ String userName = context.getSession().getUserName();
+ if (!AuthorityChecker.SUPER_USER.equals(userName)) {
+ TSStatus status;
+ try {
+ List<PartialPath> paths =
+ Collections.singletonList(new PartialPath(device,
measurementId));
+ status =
+ AuthorityChecker.getTSStatus(
+ AuthorityChecker.checkFullPathListPermission(
+ userName, paths, PrivilegeType.WRITE_DATA.ordinal()),
+ paths,
+ PrivilegeType.WRITE_DATA);
+ } catch (IllegalPathException e) {
+ throw new RuntimeException(e);
+ }
+ 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 c9e33ffb5c2..7c131a2e912 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
@@ -24,6 +24,8 @@ import org.apache.iotdb.db.exception.sql.SemanticException;
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;
@@ -42,10 +44,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,
@@ -64,7 +71,8 @@ public final class ActiveLoadPathHelper {
final Boolean verify,
final Boolean autoCreateSchema,
final Long tabletConversionThresholdBytes,
- final Boolean pipeGenerated) {
+ final Boolean pipeGenerated,
+ final String userName) {
return buildAttributes(
null,
databaseLevel,
@@ -72,7 +80,8 @@ public final class ActiveLoadPathHelper {
verify,
autoCreateSchema,
tabletConversionThresholdBytes,
- pipeGenerated);
+ pipeGenerated,
+ userName);
}
public static Map<String, String> buildAttributes(
@@ -82,8 +91,12 @@ public final class ActiveLoadPathHelper {
final Boolean verify,
final Boolean autoCreateSchema,
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);
@@ -223,7 +236,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) {
@@ -242,7 +263,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);
@@ -273,6 +298,11 @@ public final class ActiveLoadPathHelper {
case LoadTsFileConfigurator.AUTO_CREATE_SCHEMA_KEY:
LoadTsFileConfigurator.validateAutoCreateSchemaParam(value);
break;
+ case USER_KEY:
+ if (value == null || value.isEmpty()) {
+ throw new SemanticException("User name must not be empty.");
+ }
+ break;
default:
LoadTsFileConfigurator.validateParameters(key, value);
}
@@ -298,4 +328,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 fddd8223903..7d661a18109 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
@@ -32,7 +32,6 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.protocol.session.IClientSession;
import org.apache.iotdb.db.protocol.session.InternalClientSession;
import org.apache.iotdb.db.protocol.session.SessionManager;
-import org.apache.iotdb.db.queryengine.common.SessionInfo;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
@@ -264,6 +263,9 @@ 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);
+ session.setUsername(userName);
return executeStatement(
entry.isGeneratedByPipe() ? new PipeEnrichedStatement(statement) :
statement, session);
@@ -276,7 +278,7 @@ public class ActiveLoadTsFileLoader {
.executeForTreeModel(
statement,
SessionManager.getInstance().requestQueryId(),
- new SessionInfo(0, AuthorityChecker.SUPER_USER,
ZoneId.systemDefault(), ""),
+ SESSION_MANAGER.getSessionInfo(session),
"",
ClusterPartitionFetcher.getInstance(),
ClusterSchemaFetcher.getInstance(),
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadUtil.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadUtil.java
index a27bc6882df..93f8f04b481 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadUtil.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadUtil.java
@@ -21,8 +21,11 @@ package org.apache.iotdb.db.storageengine.load.active;
import org.apache.iotdb.commons.utils.FileUtils;
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.exception.DiskSpaceInsufficientException;
+import org.apache.iotdb.db.protocol.session.IClientSession;
+import org.apache.iotdb.db.protocol.session.SessionManager;
import org.apache.iotdb.db.storageengine.load.disk.ILoadDiskSelector;
import org.apache.iotdb.db.storageengine.rescon.disk.FolderManager;
import
org.apache.iotdb.db.storageengine.rescon.disk.strategy.DirectoryStrategyType;
@@ -37,8 +40,8 @@ import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.Comparator;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -92,8 +95,7 @@ public class ActiveLoadUtil {
LOGGER.warn("Load active listening dir is 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);
transferFilesToActiveDir(
@@ -106,6 +108,23 @@ public class ActiveLoadUtil {
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,
@@ -129,8 +148,7 @@ public class ActiveLoadUtil {
LOGGER.warn("Load active listening dir is 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);
final List<File> sourceFiles = new ArrayList<>(files.size());
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
index 1a75dcce142..b0bab25eb90 100644
---
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
@@ -36,7 +36,7 @@ public class ActiveLoadPathHelperTest {
final File pendingDir =
Files.createTempDirectory("active-load-schema").toFile();
try {
final Map<String, String> attributes =
- ActiveLoadPathHelper.buildAttributes(null, null, null, true, false,
null, true);
+ ActiveLoadPathHelper.buildAttributes(null, null, null, true, false,
null, true, null);
final File targetDir = ActiveLoadPathHelper.resolveTargetDir(pendingDir,
attributes);
Assert.assertTrue(targetDir.mkdirs());
final File tsFile = new File(targetDir, "1-0-0-0.tsfile");
@@ -58,6 +58,102 @@ 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, 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;
@@ -70,4 +166,9 @@ public class ActiveLoadPathHelperTest {
}
Assert.assertTrue(file.delete());
}
+
+ private static void createFile(final File file) throws Exception {
+ Assert.assertTrue(file.getParentFile().mkdirs());
+ Assert.assertTrue(file.createNewFile());
+ }
}
