From 9dac339da5b324bd201a48ed13e7136e43b0b9ea Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Wed, 8 Jul 2026 14:18:36 +0800 Subject: [PATCH 1/7] Fix load write permission check --- .../iotdb/db/it/IoTDBLoadTsFileAuthIT.java | 167 ++++++++++++++++++ .../plan/analyze/load/LoadTsFileAnalyzer.java | 2 + .../TreeSchemaAutoCreatorAndVerifier.java | 72 ++++++-- 3 files changed, 227 insertions(+), 14 deletions(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileAuthIT.java 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 000000000000..2de4d160f55d --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileAuthIT.java @@ -0,0 +1,167 @@ +/* + * 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.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 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 PASSWORD = "test123123456"; + + 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() + .setLoadTsFileAnalyzeSchemaMemorySizeInBytes(10 * 1024L); + + 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); + } + + private static void prepareSchemaAndTsFile(final File tsFile) 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(), MEASUREMENT.getType())); + } + + 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) { + } + } +} 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 89b2f8f2c523..825323e4f81f 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 @@ -528,6 +528,8 @@ private void doAnalyzeSingleTreeFile( 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 a5e8f2d1c3c9..6d1704c0cfd0 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 void autoCreateAndVerify( // 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 compressionEncodingPair = reader.readTimeseriesCompressionTypeAndEncoding(timeseriesMetadata); schemaCache.addTimeSeries( @@ -191,6 +178,63 @@ public void autoCreateAndVerify( } } + public void checkWritePermission( + Map> device2TimeseriesMetadataList) throws AuthException { + for (final Map.Entry> 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. From 1acf7a4c58d4011d53101b30b26fb3870e9d0017 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Wed, 8 Jul 2026 15:55:29 +0800 Subject: [PATCH 2/7] fix async --- .../db/it/IoTDBLoadTsFileActiveRetryIT.java | 169 ++++++++++++++++++ .../thrift/IoTDBDataNodeReceiver.java | 3 +- .../plan/analyze/load/LoadTsFileAnalyzer.java | 3 +- .../load/active/ActiveLoadPathHelper.java | 47 ++++- .../load/active/ActiveLoadTsFileLoader.java | 9 + .../db/storageengine/load/util/LoadUtil.java | 28 ++- .../load/active/ActiveLoadPathHelperTest.java | 130 ++++++++++++++ 7 files changed, 379 insertions(+), 10 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelperTest.java diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileActiveRetryIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileActiveRetryIT.java index 3827615fe24d..17e2671e44ae 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileActiveRetryIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileActiveRetryIT.java @@ -19,6 +19,7 @@ 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; @@ -43,6 +44,10 @@ import java.util.Collections; import java.util.concurrent.TimeUnit; +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 IoTDBLoadTsFileActiveRetryIT { @@ -50,6 +55,11 @@ public class IoTDBLoadTsFileActiveRetryIT { private static final String DATABASE = "root.sg.test_0"; private static final String DEVICE = DATABASE + ".d_0"; private static final String MEASUREMENT = "sensor_00"; + private static final String ACTIVE_LOAD_USER = "active_load_user"; + private static final String ACTIVE_LOAD_NO_WRITE_USER = "active_load_no_write_user"; + private static final String ACTIVE_LOAD_WRITE_USER = "active_load_write_user"; + private static final String DELETED_ACTIVE_LOAD_USER = "deleted_active_load_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 final MeasurementSchema TSFILE_SCHEMA = @@ -129,6 +139,129 @@ public void testActiveLoadTemporaryUnavailableShouldKeepFileForRetry() throws Ex } } + @Test + public void testAsyncLoadShouldStoreUserInfoInActiveDir() throws Exception { + final DataNodeWrapper dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapper(0); + final File retryTsFile = new File(tmpDir, "3-0-0-0.tsfile"); + generateTsFile(retryTsFile); + + try (final Connection connection = + EnvFactory.getEnv().getConnectionWithSpecifiedDataNode(dataNodeWrapper); + final Statement statement = connection.createStatement()) { + statement.execute("create database " + DATABASE); + statement.execute( + String.format( + "create timeseries %s.%s %s", DEVICE, MEASUREMENT, TSDataType.INT64.name())); + createUser(ACTIVE_LOAD_USER, PASSWORD); + grantUserSeriesPrivilege(ACTIVE_LOAD_USER, PrivilegeType.WRITE_DATA, DATABASE + ".**"); + + executeNonQuery( + String.format( + "load \"%s\" with ('database-level'='3', 'async'='true', 'on-success'='none', " + + "'convert-on-type-mismatch'='true')", + retryTsFile.getAbsolutePath()), + ACTIVE_LOAD_USER, + PASSWORD); + + final File activeDir = getActiveLoadDir(dataNodeWrapper); + final File activeTsFile = waitForFile(activeDir, retryTsFile.getName(), 30_000L); + + Assert.assertNotNull( + "Async load should copy tsfile into active load directory", activeTsFile); + Assert.assertTrue( + "Async load should store user info in active load directory", + activeTsFile.getAbsolutePath().contains("user-")); + Assert.assertFalse( + "Async load should not expose raw username in active load directory", + activeTsFile.getAbsolutePath().contains(ACTIVE_LOAD_USER)); + } + } + + @Test + public void testActiveLoadShouldCheckWriteDataPermissionWithStoredUser() throws Exception { + final DataNodeWrapper dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapper(0); + final File noWriteTsFile = new File(tmpDir, "4-0-0-0.tsfile"); + final File writeTsFile = new File(tmpDir, "5-0-0-0.tsfile"); + generateTsFile(noWriteTsFile); + generateTsFile(writeTsFile); + + try (final Connection connection = + EnvFactory.getEnv().getConnectionWithSpecifiedDataNode(dataNodeWrapper); + final Statement statement = connection.createStatement()) { + statement.execute("create database " + DATABASE); + statement.execute( + String.format( + "create timeseries %s.%s %s", DEVICE, MEASUREMENT, TSDataType.INT32.name())); + createUser(ACTIVE_LOAD_NO_WRITE_USER, PASSWORD); + createUser(ACTIVE_LOAD_WRITE_USER, PASSWORD); + grantUserSeriesPrivilege(ACTIVE_LOAD_WRITE_USER, PrivilegeType.WRITE_DATA, DATABASE + ".**"); + + executeNonQuery( + String.format( + "load \"%s\" with ('database-level'='3', 'async'='true', 'on-success'='none', " + + "'verify'='false')", + noWriteTsFile.getAbsolutePath()), + ACTIVE_LOAD_NO_WRITE_USER, + PASSWORD); + executeNonQuery( + String.format( + "load \"%s\" with ('database-level'='3', 'async'='true', 'on-success'='none', " + + "'verify'='false')", + writeTsFile.getAbsolutePath()), + ACTIVE_LOAD_WRITE_USER, + PASSWORD); + } + + Assert.assertNotNull( + "Active load without WRITE_DATA should be moved to fail dir", + waitForFile( + getActiveLoadFailDir(dataNodeWrapper), + noWriteTsFile.getName(), + TimeUnit.SECONDS.toMillis(60))); + assertCountEventually(10, TimeUnit.SECONDS.toMillis(60)); + } + + @Test + public void testActiveLoadShouldMoveFileToFailDirWhenUserDoesNotExist() throws Exception { + final DataNodeWrapper dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapper(0); + final File sourceTsFile = new File(tmpDir, "6-0-0-0.tsfile"); + generateTsFile(sourceTsFile); + + try (final Connection connection = + EnvFactory.getEnv().getConnectionWithSpecifiedDataNode(dataNodeWrapper); + final Statement statement = connection.createStatement()) { + statement.execute("create database " + DATABASE); + statement.execute( + String.format( + "create timeseries %s.%s %s", DEVICE, MEASUREMENT, TSDataType.INT64.name())); + createUser(DELETED_ACTIVE_LOAD_USER, PASSWORD); + grantUserSeriesPrivilege( + DELETED_ACTIVE_LOAD_USER, PrivilegeType.WRITE_DATA, DATABASE + ".**"); + + executeNonQuery( + String.format( + "load \"%s\" with ('database-level'='3', 'async'='true', 'on-success'='none', " + + "'convert-on-type-mismatch'='true')", + sourceTsFile.getAbsolutePath()), + DELETED_ACTIVE_LOAD_USER, + PASSWORD); + + Assert.assertNotNull( + "Async load should copy tsfile into active load directory", + waitForFileUnderPathContaining( + getActiveLoadDir(dataNodeWrapper), sourceTsFile.getName(), "user-", 30_000L)); + + statement.execute("drop user " + DELETED_ACTIVE_LOAD_USER); + } + + Assert.assertNotNull( + "Active load should move tsfile to fail dir when user does not exist", + waitForFile( + getActiveLoadFailDir(dataNodeWrapper), + sourceTsFile.getName(), + TimeUnit.SECONDS.toMillis(60))); + } + private void generateTsFile(final File tsFile) throws Exception { try (final TsFileGenerator generator = new TsFileGenerator(tsFile)) { generator.registerTimeseries(DEVICE, Collections.singletonList(TSFILE_SCHEMA)); @@ -171,6 +304,20 @@ private File waitForFile(final File root, final String fileName, final long time return null; } + private File waitForFileUnderPathContaining( + final File root, final String fileName, final String pathSegment, final long timeoutMs) + throws InterruptedException { + final long deadline = System.currentTimeMillis() + timeoutMs; + while (System.currentTimeMillis() < deadline) { + final File file = findFile(root, fileName); + if (file != null && file.getAbsolutePath().contains(pathSegment)) { + return file; + } + Thread.sleep(500L); + } + return null; + } + private boolean containsFile(final File root, final String fileName) { return findFile(root, fileName) != null; } @@ -190,6 +337,28 @@ private void assertFileKeptForRetry( } } + 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 java.sql.ResultSet resultSet = + statement.executeQuery("select count(" + MEASUREMENT + ") 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 File findFile(final File root, final String fileName) { if (root == null || !root.exists()) { return null; 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 46c00a095c84..2697eb0bf7e3 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 @@ -653,7 +653,8 @@ static Map buildLoadTsFileAttributesForAsync( 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 825323e4f81f..f459c899f043 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 @@ -295,7 +295,8 @@ private boolean doAsyncLoad(final IAnalysis analysis) { isConvertOnTypeMismatch, isVerifySchema, tabletConversionThresholdBytes, - isGeneratedByPipe); + isGeneratedByPipe, + Objects.nonNull(context) ? context.getUsername() : null); if (LoadUtil.loadTsFileAsyncToActiveDir(tsFiles, activeLoadAttributes, isDeleteAfterLoad)) { analysis.setFinishQueryAfterAnalyze(true); 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 7b971c15a10b..50a5b8c7c916 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 @@ -31,6 +31,7 @@ import java.net.URLEncoder; import java.nio.charset.StandardCharsets; import java.util.Arrays; +import java.util.Base64; import java.util.Collections; import java.util.HashMap; import java.util.LinkedHashMap; @@ -43,10 +44,13 @@ public final class ActiveLoadPathHelper { private static final String SEGMENT_SEPARATOR = "-"; + public static final String USER_KEY = "user"; + private static final String USER_VALUE_MASK_PREFIX = "b64:"; private static final List KEY_ORDER = Collections.unmodifiableList( Arrays.asList( + USER_KEY, LoadTsFileConfigurator.DATABASE_NAME_KEY, LoadTsFileConfigurator.DATABASE_LEVEL_KEY, LoadTsFileConfigurator.CONVERT_ON_TYPE_MISMATCH_KEY, @@ -65,8 +69,13 @@ public static Map buildAttributes( final Boolean convertOnTypeMismatch, final Boolean verify, final Long tabletConversionThresholdBytes, - final Boolean pipeGenerated) { + final Boolean pipeGenerated, + final String userName) { final Map 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 +217,17 @@ public static boolean containsDatabaseName(final Map attributes) } 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 + + Base64.getUrlEncoder() + .withoutPadding() + .encodeToString(value.getBytes(StandardCharsets.UTF_8)); } private static String encodeValue(final String value) { @@ -227,7 +246,11 @@ private static Optional extractAndValidateAttributeValue( } 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 +278,11 @@ private static void validateAttributeValue(final String key, final String value) case LoadTsFileConfigurator.VERIFY_KEY: LoadTsFileConfigurator.validateVerifyParam(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); } @@ -279,4 +307,17 @@ private static String decodeValue(final String value) { 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; + } + try { + return new String( + Base64.getUrlDecoder().decode(value.substring(USER_VALUE_MASK_PREFIX.length())), + StandardCharsets.UTF_8); + } catch (final IllegalArgumentException e) { + return value; + } + } } 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 5e6cade3087c..737958754bf0 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 @@ private TSStatus loadTsFile( : new File(entry.getPendingDir()); final Map attributes = ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir); ActiveLoadPathHelper.applyAttributesToStatement(attributes, statement, isVerify); + final String userName = + attributes.getOrDefault(ActiveLoadPathHelper.USER_KEY, AuthorityChecker.SUPER_USER); + final Optional userId = AuthorityChecker.getUserId(userName); + if (!userId.isPresent()) { + return new TSStatus(TSStatusCode.USER_NOT_EXIST.getStatusCode()) + .setMessage("The user in the 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 cafb1b0e7002..88c71fb1cef2 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.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 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 @@ private static boolean loadTsFilesToActiveDir( LOGGER.warn(StorageEngineMessages.LOAD_ACTIVE_LISTENING_DIR_NOT_SET); return false; } - final Map attributes = - Objects.nonNull(loadAttributes) ? loadAttributes : Collections.emptyMap(); + final Map attributes = appendCurrentUserIfAbsent(loadAttributes); final File targetDir = ActiveLoadPathHelper.resolveTargetDir(targetFilePath, attributes); loadTsFileAsyncToTargetDir( @@ -138,6 +140,23 @@ private static boolean loadTsFilesToActiveDir( return true; } + private static Map appendCurrentUserIfAbsent( + final Map loadAttributes) { + final Map 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 loadAttributes, final List files, @@ -161,8 +180,7 @@ public static boolean loadFilesToActiveDir( LOGGER.warn(StorageEngineMessages.LOAD_ACTIVE_LISTENING_DIR_NOT_SET); return false; } - final Map attributes = - Objects.nonNull(loadAttributes) ? loadAttributes : Collections.emptyMap(); + final Map 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 000000000000..c3a17f10dfb6 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelperTest.java @@ -0,0 +1,130 @@ +/* + * 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-")); + Assert.assertFalse(targetDir.getAbsolutePath().contains(userName)); + + final Map 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 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 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 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()); + } +} From c11a422ab7f3470521107bb14d1bc8f262efaa3f Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Wed, 8 Jul 2026 16:00:45 +0800 Subject: [PATCH 3/7] Optimize load auth tests --- .../db/it/IoTDBLoadTsFileActiveRetryIT.java | 124 ------------ .../iotdb/db/it/IoTDBLoadTsFileAuthIT.java | 186 +++++++++++++++++- 2 files changed, 183 insertions(+), 127 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileActiveRetryIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileActiveRetryIT.java index 17e2671e44ae..834af2bcfaa5 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileActiveRetryIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileActiveRetryIT.java @@ -56,9 +56,6 @@ public class IoTDBLoadTsFileActiveRetryIT { private static final String DEVICE = DATABASE + ".d_0"; private static final String MEASUREMENT = "sensor_00"; private static final String ACTIVE_LOAD_USER = "active_load_user"; - private static final String ACTIVE_LOAD_NO_WRITE_USER = "active_load_no_write_user"; - private static final String ACTIVE_LOAD_WRITE_USER = "active_load_write_user"; - private static final String DELETED_ACTIVE_LOAD_USER = "deleted_active_load_user"; private static final String PASSWORD = "test123123456"; private static final long UNALLOCATABLE_TABLET_CONVERSION_BATCH_MEMORY_SIZE_IN_BYTES = Long.MAX_VALUE / 4; @@ -177,91 +174,6 @@ public void testAsyncLoadShouldStoreUserInfoInActiveDir() throws Exception { } } - @Test - public void testActiveLoadShouldCheckWriteDataPermissionWithStoredUser() throws Exception { - final DataNodeWrapper dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapper(0); - final File noWriteTsFile = new File(tmpDir, "4-0-0-0.tsfile"); - final File writeTsFile = new File(tmpDir, "5-0-0-0.tsfile"); - generateTsFile(noWriteTsFile); - generateTsFile(writeTsFile); - - try (final Connection connection = - EnvFactory.getEnv().getConnectionWithSpecifiedDataNode(dataNodeWrapper); - final Statement statement = connection.createStatement()) { - statement.execute("create database " + DATABASE); - statement.execute( - String.format( - "create timeseries %s.%s %s", DEVICE, MEASUREMENT, TSDataType.INT32.name())); - createUser(ACTIVE_LOAD_NO_WRITE_USER, PASSWORD); - createUser(ACTIVE_LOAD_WRITE_USER, PASSWORD); - grantUserSeriesPrivilege(ACTIVE_LOAD_WRITE_USER, PrivilegeType.WRITE_DATA, DATABASE + ".**"); - - executeNonQuery( - String.format( - "load \"%s\" with ('database-level'='3', 'async'='true', 'on-success'='none', " - + "'verify'='false')", - noWriteTsFile.getAbsolutePath()), - ACTIVE_LOAD_NO_WRITE_USER, - PASSWORD); - executeNonQuery( - String.format( - "load \"%s\" with ('database-level'='3', 'async'='true', 'on-success'='none', " - + "'verify'='false')", - writeTsFile.getAbsolutePath()), - ACTIVE_LOAD_WRITE_USER, - PASSWORD); - } - - Assert.assertNotNull( - "Active load without WRITE_DATA should be moved to fail dir", - waitForFile( - getActiveLoadFailDir(dataNodeWrapper), - noWriteTsFile.getName(), - TimeUnit.SECONDS.toMillis(60))); - assertCountEventually(10, TimeUnit.SECONDS.toMillis(60)); - } - - @Test - public void testActiveLoadShouldMoveFileToFailDirWhenUserDoesNotExist() throws Exception { - final DataNodeWrapper dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapper(0); - final File sourceTsFile = new File(tmpDir, "6-0-0-0.tsfile"); - generateTsFile(sourceTsFile); - - try (final Connection connection = - EnvFactory.getEnv().getConnectionWithSpecifiedDataNode(dataNodeWrapper); - final Statement statement = connection.createStatement()) { - statement.execute("create database " + DATABASE); - statement.execute( - String.format( - "create timeseries %s.%s %s", DEVICE, MEASUREMENT, TSDataType.INT64.name())); - createUser(DELETED_ACTIVE_LOAD_USER, PASSWORD); - grantUserSeriesPrivilege( - DELETED_ACTIVE_LOAD_USER, PrivilegeType.WRITE_DATA, DATABASE + ".**"); - - executeNonQuery( - String.format( - "load \"%s\" with ('database-level'='3', 'async'='true', 'on-success'='none', " - + "'convert-on-type-mismatch'='true')", - sourceTsFile.getAbsolutePath()), - DELETED_ACTIVE_LOAD_USER, - PASSWORD); - - Assert.assertNotNull( - "Async load should copy tsfile into active load directory", - waitForFileUnderPathContaining( - getActiveLoadDir(dataNodeWrapper), sourceTsFile.getName(), "user-", 30_000L)); - - statement.execute("drop user " + DELETED_ACTIVE_LOAD_USER); - } - - Assert.assertNotNull( - "Active load should move tsfile to fail dir when user does not exist", - waitForFile( - getActiveLoadFailDir(dataNodeWrapper), - sourceTsFile.getName(), - TimeUnit.SECONDS.toMillis(60))); - } - private void generateTsFile(final File tsFile) throws Exception { try (final TsFileGenerator generator = new TsFileGenerator(tsFile)) { generator.registerTimeseries(DEVICE, Collections.singletonList(TSFILE_SCHEMA)); @@ -304,20 +216,6 @@ private File waitForFile(final File root, final String fileName, final long time return null; } - private File waitForFileUnderPathContaining( - final File root, final String fileName, final String pathSegment, final long timeoutMs) - throws InterruptedException { - final long deadline = System.currentTimeMillis() + timeoutMs; - while (System.currentTimeMillis() < deadline) { - final File file = findFile(root, fileName); - if (file != null && file.getAbsolutePath().contains(pathSegment)) { - return file; - } - Thread.sleep(500L); - } - return null; - } - private boolean containsFile(final File root, final String fileName) { return findFile(root, fileName) != null; } @@ -337,28 +235,6 @@ private void assertFileKeptForRetry( } } - 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 java.sql.ResultSet resultSet = - statement.executeQuery("select count(" + MEASUREMENT + ") 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 File findFile(final File root, final String fileName) { if (root == null || !root.exists()) { return null; 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 index 2de4d160f55d..53bcb4d72381 100644 --- 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 @@ -21,6 +21,7 @@ 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; @@ -46,6 +47,7 @@ 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; @@ -63,7 +65,12 @@ public class IoTDBLoadTsFileAuthIT { 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 DELETED_ASYNC_LOAD_USER = "deleted_async_load_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; @@ -76,7 +83,11 @@ public static void setUp() throws Exception { EnvFactory.getEnv() .getConfig() .getDataNodeConfig() - .setLoadTsFileAnalyzeSchemaMemorySizeInBytes(10 * 1024L); + .setMaxAllocateMemoryRatioForLoad(1.0) + .setLoadTsFileAnalyzeSchemaMemorySizeInBytes(10 * 1024L) + .setLoadTsFileTabletConversionBatchMemorySizeInBytes( + UNALLOCATABLE_TABLET_CONVERSION_BATCH_MEMORY_SIZE_IN_BYTES) + .setLoadActiveListeningCheckIntervalSeconds(1); EnvFactory.getEnv().initClusterEnvironment(); } @@ -141,16 +152,92 @@ public void testLoadWithoutSchemaCheckRejectsUserWithOtherPathWriteDataPermissio PASSWORD); } + @Test + public void testAsyncLoadShouldCheckWriteDataPermissionWithStoredUser() throws Exception { + final DataNodeWrapper dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapper(0); + 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); + + Assert.assertNotNull( + "Async load without WRITE_DATA should be moved to fail dir", + waitForFile( + getActiveLoadFailDir(dataNodeWrapper), + noWriteTsFile.getName(), + TimeUnit.SECONDS.toMillis(60))); + assertCountEventually(10, TimeUnit.SECONDS.toMillis(60)); + } + + @Test + public void testAsyncLoadShouldMoveFileToFailDirWhenUserDoesNotExist() throws Exception { + final DataNodeWrapper dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapper(0); + final File tsFile = new File(tmpDir, "6-0-0-0.tsfile"); + prepareSchema(TSDataType.INT64); + generateTsFile(tsFile); + createUser(DELETED_ASYNC_LOAD_USER, PASSWORD); + grantUserSeriesPrivilege(DELETED_ASYNC_LOAD_USER, PrivilegeType.WRITE_DATA, DATABASE + ".**"); + + executeNonQuery( + String.format( + "load \"%s\" with ('database-level'='2', 'async'='true', 'on-success'='none', " + + "'convert-on-type-mismatch'='true')", + tsFile.getAbsolutePath()), + DELETED_ASYNC_LOAD_USER, + PASSWORD); + + Assert.assertNotNull( + "Async load should copy tsfile into active load directory", + waitForFileUnderPathContaining( + getActiveLoadDir(dataNodeWrapper), tsFile.getName(), "user-", 30_000L)); + + try (final Connection connection = EnvFactory.getEnv().getConnection(); + final Statement statement = connection.createStatement()) { + statement.execute("drop user " + DELETED_ASYNC_LOAD_USER); + } + + Assert.assertNotNull( + "Async load should move tsfile to fail dir when user does not exist", + waitForFile( + getActiveLoadFailDir(dataNodeWrapper), + tsFile.getName(), + 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(), MEASUREMENT.getType())); + "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); @@ -164,4 +251,97 @@ private static void deleteDatabase() throws Exception { } catch (final IoTDBSQLException ignored) { } } + + private File getActiveLoadDir(final DataNodeWrapper dataNodeWrapper) { + return new File( + dataNodeWrapper.getNodePath() + + File.separator + + "ext" + + File.separator + + "load" + + File.separator + + "pending"); + } + + private File getActiveLoadFailDir(final DataNodeWrapper dataNodeWrapper) { + return new File( + dataNodeWrapper.getNodePath() + + File.separator + + "ext" + + File.separator + + "load" + + File.separator + + "failed"); + } + + private File waitForFile(final File root, final String fileName, final long timeoutMs) + throws InterruptedException { + final long deadline = System.currentTimeMillis() + timeoutMs; + while (System.currentTimeMillis() < deadline) { + final File file = findFile(root, fileName); + if (file != null) { + return file; + } + Thread.sleep(500L); + } + return null; + } + + private File waitForFileUnderPathContaining( + final File root, final String fileName, final String pathSegment, final long timeoutMs) + throws InterruptedException { + final long deadline = System.currentTimeMillis() + timeoutMs; + while (System.currentTimeMillis() < deadline) { + final File file = findFile(root, fileName); + if (file != null && file.getAbsolutePath().contains(pathSegment)) { + return file; + } + Thread.sleep(500L); + } + return null; + } + + 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 File findFile(final File root, final String fileName) { + if (root == null || !root.exists()) { + return null; + } + if (root.isFile()) { + return root.getName().equals(fileName) ? root : null; + } + + final File[] children = root.listFiles(); + if (children == null) { + return null; + } + for (final File child : children) { + final File file = findFile(child, fileName); + if (file != null) { + return file; + } + } + return null; + } } From 4d3160a1c3e79ee843b9a409ff21066f0dbce808 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 9 Jul 2026 11:13:27 +0800 Subject: [PATCH 4/7] Optimize load user path encoding --- .../load/active/ActiveLoadPathHelper.java | 13 ++++++++----- .../load/active/ActiveLoadPathHelperTest.java | 18 +++++++++++++++++- 2 files changed, 25 insertions(+), 6 deletions(-) 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 50a5b8c7c916..c5b5f8c531a5 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 @@ -45,7 +45,8 @@ public final class ActiveLoadPathHelper { private static final String SEGMENT_SEPARATOR = "-"; public static final String USER_KEY = "user"; - private static final String USER_VALUE_MASK_PREFIX = "b64:"; + // 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 List KEY_ORDER = Collections.unmodifiableList( @@ -312,12 +313,14 @@ private static String unmaskValueIfNecessary(final String key, final String valu 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( - Base64.getUrlDecoder().decode(value.substring(USER_VALUE_MASK_PREFIX.length())), - StandardCharsets.UTF_8); + return new String(Base64.getUrlDecoder().decode(encodedUserName), StandardCharsets.UTF_8); } catch (final IllegalArgumentException e) { - return value; + return fallback; } } } 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 c3a17f10dfb6..77936a653c7b 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 @@ -42,8 +42,9 @@ public void testUserAttributeShouldBeMaskedInPathAndDecodedWhenParsing() throws 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-")); + Assert.assertTrue(targetDir.getAbsolutePath().contains("user-v1-")); Assert.assertFalse(targetDir.getAbsolutePath().contains(userName)); + Assert.assertFalse(targetDir.getAbsolutePath().contains("b64%3A")); final Map attributes = ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir); @@ -67,6 +68,21 @@ public void testRawUserAttributeShouldBeIgnored() throws Exception { } } + @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 attributes = + ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir); + Assert.assertFalse(attributes.containsKey(ActiveLoadPathHelper.USER_KEY)); + } finally { + deleteRecursively(pendingDir); + } + } + @Test public void testUnknownRawAttributeDirectoryShouldBeIgnoredForDowngradeCompatibility() throws Exception { From 8ae23f237bcddbd945a500d8c75f7a929506e07e Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 9 Jul 2026 11:43:59 +0800 Subject: [PATCH 5/7] fix async --- .../iotdb/db/it/IoTDBLoadTsFileAuthIT.java | 35 ------------------- 1 file changed, 35 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 index 53bcb4d72381..8aa08f4e0dfa 100644 --- 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 @@ -187,41 +187,6 @@ public void testAsyncLoadShouldCheckWriteDataPermissionWithStoredUser() throws E assertCountEventually(10, TimeUnit.SECONDS.toMillis(60)); } - @Test - public void testAsyncLoadShouldMoveFileToFailDirWhenUserDoesNotExist() throws Exception { - final DataNodeWrapper dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapper(0); - final File tsFile = new File(tmpDir, "6-0-0-0.tsfile"); - prepareSchema(TSDataType.INT64); - generateTsFile(tsFile); - createUser(DELETED_ASYNC_LOAD_USER, PASSWORD); - grantUserSeriesPrivilege(DELETED_ASYNC_LOAD_USER, PrivilegeType.WRITE_DATA, DATABASE + ".**"); - - executeNonQuery( - String.format( - "load \"%s\" with ('database-level'='2', 'async'='true', 'on-success'='none', " - + "'convert-on-type-mismatch'='true')", - tsFile.getAbsolutePath()), - DELETED_ASYNC_LOAD_USER, - PASSWORD); - - Assert.assertNotNull( - "Async load should copy tsfile into active load directory", - waitForFileUnderPathContaining( - getActiveLoadDir(dataNodeWrapper), tsFile.getName(), "user-", 30_000L)); - - try (final Connection connection = EnvFactory.getEnv().getConnection(); - final Statement statement = connection.createStatement()) { - statement.execute("drop user " + DELETED_ASYNC_LOAD_USER); - } - - Assert.assertNotNull( - "Async load should move tsfile to fail dir when user does not exist", - waitForFile( - getActiveLoadFailDir(dataNodeWrapper), - tsFile.getName(), - TimeUnit.SECONDS.toMillis(60))); - } - private static void prepareSchemaAndTsFile(final File tsFile) throws Exception { prepareSchema(MEASUREMENT.getType()); generateTsFile(tsFile); From 6632900b6e2126bffd38b68e383a8bf1a38df511 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 9 Jul 2026 15:35:31 +0800 Subject: [PATCH 6/7] Optimize load auth review fixes --- .../db/it/IoTDBLoadTsFileActiveRetryIT.java | 45 ------------------- .../iotdb/db/it/IoTDBLoadTsFileAuthIT.java | 26 ----------- .../iotdb/db/i18n/StorageEngineMessages.java | 3 ++ .../iotdb/db/i18n/StorageEngineMessages.java | 3 ++ .../load/active/ActiveLoadPathHelper.java | 12 ++--- .../load/active/ActiveLoadTsFileLoader.java | 2 +- 6 files changed, 13 insertions(+), 78 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileActiveRetryIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileActiveRetryIT.java index 834af2bcfaa5..3827615fe24d 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileActiveRetryIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileActiveRetryIT.java @@ -19,7 +19,6 @@ 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; @@ -44,10 +43,6 @@ import java.util.Collections; import java.util.concurrent.TimeUnit; -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 IoTDBLoadTsFileActiveRetryIT { @@ -55,8 +50,6 @@ public class IoTDBLoadTsFileActiveRetryIT { private static final String DATABASE = "root.sg.test_0"; private static final String DEVICE = DATABASE + ".d_0"; private static final String MEASUREMENT = "sensor_00"; - private static final String ACTIVE_LOAD_USER = "active_load_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 final MeasurementSchema TSFILE_SCHEMA = @@ -136,44 +129,6 @@ public void testActiveLoadTemporaryUnavailableShouldKeepFileForRetry() throws Ex } } - @Test - public void testAsyncLoadShouldStoreUserInfoInActiveDir() throws Exception { - final DataNodeWrapper dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapper(0); - final File retryTsFile = new File(tmpDir, "3-0-0-0.tsfile"); - generateTsFile(retryTsFile); - - try (final Connection connection = - EnvFactory.getEnv().getConnectionWithSpecifiedDataNode(dataNodeWrapper); - final Statement statement = connection.createStatement()) { - statement.execute("create database " + DATABASE); - statement.execute( - String.format( - "create timeseries %s.%s %s", DEVICE, MEASUREMENT, TSDataType.INT64.name())); - createUser(ACTIVE_LOAD_USER, PASSWORD); - grantUserSeriesPrivilege(ACTIVE_LOAD_USER, PrivilegeType.WRITE_DATA, DATABASE + ".**"); - - executeNonQuery( - String.format( - "load \"%s\" with ('database-level'='3', 'async'='true', 'on-success'='none', " - + "'convert-on-type-mismatch'='true')", - retryTsFile.getAbsolutePath()), - ACTIVE_LOAD_USER, - PASSWORD); - - final File activeDir = getActiveLoadDir(dataNodeWrapper); - final File activeTsFile = waitForFile(activeDir, retryTsFile.getName(), 30_000L); - - Assert.assertNotNull( - "Async load should copy tsfile into active load directory", activeTsFile); - Assert.assertTrue( - "Async load should store user info in active load directory", - activeTsFile.getAbsolutePath().contains("user-")); - Assert.assertFalse( - "Async load should not expose raw username in active load directory", - activeTsFile.getAbsolutePath().contains(ACTIVE_LOAD_USER)); - } - } - private void generateTsFile(final File tsFile) throws Exception { try (final TsFileGenerator generator = new TsFileGenerator(tsFile)) { generator.registerTimeseries(DEVICE, Collections.singletonList(TSFILE_SCHEMA)); 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 index 8aa08f4e0dfa..ebbd5eb38d8f 100644 --- 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 @@ -67,7 +67,6 @@ public class IoTDBLoadTsFileAuthIT { 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 DELETED_ASYNC_LOAD_USER = "deleted_async_load_user"; private static final String PASSWORD = "test123123456"; private static final long UNALLOCATABLE_TABLET_CONVERSION_BATCH_MEMORY_SIZE_IN_BYTES = Long.MAX_VALUE / 4; @@ -217,17 +216,6 @@ private static void deleteDatabase() throws Exception { } } - private File getActiveLoadDir(final DataNodeWrapper dataNodeWrapper) { - return new File( - dataNodeWrapper.getNodePath() - + File.separator - + "ext" - + File.separator - + "load" - + File.separator - + "pending"); - } - private File getActiveLoadFailDir(final DataNodeWrapper dataNodeWrapper) { return new File( dataNodeWrapper.getNodePath() @@ -252,20 +240,6 @@ private File waitForFile(final File root, final String fileName, final long time return null; } - private File waitForFileUnderPathContaining( - final File root, final String fileName, final String pathSegment, final long timeoutMs) - throws InterruptedException { - final long deadline = System.currentTimeMillis() + timeoutMs; - while (System.currentTimeMillis() < deadline) { - final File file = findFile(root, fileName); - if (file != null && file.getAbsolutePath().contains(pathSegment)) { - return file; - } - Thread.sleep(500L); - } - return null; - } - private void assertCountEventually(final long expected, final long timeoutMs) throws Exception { final long deadline = System.currentTimeMillis() + timeoutMs; AssertionError lastError = null; 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 f8b6e50ed537..512879bfd3cb 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 @@ private 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 bffda3d52879..96d09c1c1414 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 @@ private 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/storageengine/load/active/ActiveLoadPathHelper.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelper.java index c5b5f8c531a5..991d368d3953 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,13 +25,14 @@ 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; import java.net.URLEncoder; import java.nio.charset.StandardCharsets; import java.util.Arrays; -import java.util.Base64; import java.util.Collections; import java.util.HashMap; import java.util.LinkedHashMap; @@ -47,6 +48,7 @@ public final class ActiveLoadPathHelper { 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 KEY_ORDER = Collections.unmodifiableList( @@ -226,9 +228,7 @@ private static String maskValueIfNecessary(final String key, final String value) return value; } return USER_VALUE_MASK_PREFIX - + Base64.getUrlEncoder() - .withoutPadding() - .encodeToString(value.getBytes(StandardCharsets.UTF_8)); + + USER_VALUE_ENCODING.encode(value.getBytes(StandardCharsets.UTF_8)); } private static String encodeValue(final String value) { @@ -281,7 +281,7 @@ private static void validateAttributeValue(final String key, final String value) break; case USER_KEY: if (value == null || value.isEmpty()) { - throw new SemanticException("User name must not be empty"); + throw new SemanticException(StorageEngineMessages.USER_NAME_MUST_NOT_BE_EMPTY); } break; default: @@ -318,7 +318,7 @@ private static String unmaskValueIfNecessary(final String key, final String valu private static String decodeUserName(final String encodedUserName, final String fallback) { try { - return new String(Base64.getUrlDecoder().decode(encodedUserName), StandardCharsets.UTF_8); + 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 737958754bf0..ff5222e57544 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 @@ -266,7 +266,7 @@ private TSStatus loadTsFile( final Optional userId = AuthorityChecker.getUserId(userName); if (!userId.isPresent()) { return new TSStatus(TSStatusCode.USER_NOT_EXIST.getStatusCode()) - .setMessage("The user in the active load path does not exist"); + .setMessage(StorageEngineMessages.USER_IN_ACTIVE_LOAD_PATH_DOES_NOT_EXIST); } session.setUserId(userId.get()); session.setUsername(userName); From 5650409a1300d1578ddc14a71d7c689153b2dd95 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 9 Jul 2026 17:59:17 +0800 Subject: [PATCH 7/7] fix async --- .../iotdb/db/it/IoTDBLoadTsFileAuthIT.java | 43 +++++++++---------- 1 file changed, 21 insertions(+), 22 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 index ebbd5eb38d8f..d842941e0cc9 100644 --- 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 @@ -153,7 +153,6 @@ public void testLoadWithoutSchemaCheckRejectsUserWithOtherPathWriteDataPermissio @Test public void testAsyncLoadShouldCheckWriteDataPermissionWithStoredUser() throws Exception { - final DataNodeWrapper dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapper(0); final File noWriteTsFile = new File(tmpDir, "4-0-0-0.tsfile"); final File writeTsFile = new File(tmpDir, "5-0-0-0.tsfile"); prepareSchemaAndTsFile(noWriteTsFile); @@ -177,12 +176,7 @@ public void testAsyncLoadShouldCheckWriteDataPermissionWithStoredUser() throws E ASYNC_WRITE_USER, PASSWORD); - Assert.assertNotNull( - "Async load without WRITE_DATA should be moved to fail dir", - waitForFile( - getActiveLoadFailDir(dataNodeWrapper), - noWriteTsFile.getName(), - TimeUnit.SECONDS.toMillis(60))); + waitUntilAllActiveLoadPendingDirsAreEmpty(TimeUnit.SECONDS.toMillis(60)); assertCountEventually(10, TimeUnit.SECONDS.toMillis(60)); } @@ -216,7 +210,7 @@ private static void deleteDatabase() throws Exception { } } - private File getActiveLoadFailDir(final DataNodeWrapper dataNodeWrapper) { + private File getActiveLoadPendingDir(final DataNodeWrapper dataNodeWrapper) { return new File( dataNodeWrapper.getNodePath() + File.separator @@ -224,20 +218,26 @@ private File getActiveLoadFailDir(final DataNodeWrapper dataNodeWrapper) { + File.separator + "load" + File.separator - + "failed"); + + "pending"); } - private File waitForFile(final File root, final String fileName, final long timeoutMs) + private void waitUntilAllActiveLoadPendingDirsAreEmpty(final long timeoutMs) throws InterruptedException { final long deadline = System.currentTimeMillis() + timeoutMs; while (System.currentTimeMillis() < deadline) { - final File file = findFile(root, fileName); - if (file != null) { - return file; + boolean hasTsFile = false; + for (final DataNodeWrapper dataNodeWrapper : EnvFactory.getEnv().getDataNodeWrapperList()) { + if (containsTsFile(getActiveLoadPendingDir(dataNodeWrapper))) { + hasTsFile = true; + break; + } + } + if (!hasTsFile) { + return; } Thread.sleep(500L); } - return null; + Assert.fail("Timed out waiting for active load pending dirs to become empty"); } private void assertCountEventually(final long expected, final long timeoutMs) throws Exception { @@ -263,24 +263,23 @@ private void assertCountEventually(final long expected, final long timeoutMs) th Assert.fail("Timed out waiting for count " + expected); } - private File findFile(final File root, final String fileName) { + private boolean containsTsFile(final File root) { if (root == null || !root.exists()) { - return null; + return false; } if (root.isFile()) { - return root.getName().equals(fileName) ? root : null; + return root.getName().endsWith(".tsfile"); } final File[] children = root.listFiles(); if (children == null) { - return null; + return false; } for (final File child : children) { - final File file = findFile(child, fileName); - if (file != null) { - return file; + if (containsTsFile(child)) { + return true; } } - return null; + return false; } }