diff --git a/.github/workflows/daily-it.yml b/.github/workflows/daily-it.yml index 8d9b74862e4f5..4aa36f5e102ce 100644 --- a/.github/workflows/daily-it.yml +++ b/.github/workflows/daily-it.yml @@ -1,6 +1,9 @@ name: Daily IT on: + pull_request: + branches: + - master schedule: # Run at UTC 19:00 every day (CST 03:00 AM) - cron: "0 19 * * *" diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java index 80d00c3843b3d..814de53ba68e4 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java @@ -37,6 +37,7 @@ import java.util.function.Consumer; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -66,6 +67,12 @@ public void testAssertAlwaysFail() { () -> TestUtils.assertAlwaysFail(EnvFactory.getEnv(), "show databases", 1)); } + @Test + public void testHalfTimeoutElapsedUsesSeconds() { + assertFalse(TestUtils.isHalfTimeoutElapsed(299_999, 600)); + assertTrue(TestUtils.isHalfTimeoutElapsed(300_001, 600)); + } + @Test public void testAssertDataAlwaysFailsOnMismatch() { final BaseEnv env = EnvFactory.getEnv(); diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java index e559fe9fb1897..79a2f1c984769 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java @@ -85,6 +85,10 @@ public class TestUtils { public static final String TIME_PRECISION_IN_NS = "ns"; + static boolean isHalfTimeoutElapsed(final long elapsedTimeInMillis, final long timeoutSeconds) { + return elapsedTimeInMillis > TimeUnit.SECONDS.toMillis(timeoutSeconds) / 2; + } + /** * Creates a statement on the write node without comparing results across DataNodes. * @@ -1702,7 +1706,9 @@ public static void assertDataEventuallyOnEnv( // For IoTV2 batch mode, the pipe receiver may need to flush because the replica // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases - if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) { + if (!flushed[0] + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, timeoutSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -1783,7 +1789,9 @@ public static void assertDataSizeEventuallyOnEnv( // For IoTV2 batch mode, the pipe receiver may need to flush because the replica // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases - if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) { + if (!flushed[0] + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, timeoutSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -1822,7 +1830,9 @@ public static void assertDataEventuallyOnEnv( // For IoTV2 batch mode, the pipe receiver may need to flush because the replica // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases - if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) { + if (!flushed[0] + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, timeoutSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -1904,7 +1914,9 @@ public static void assertDataEventuallyOnEnv( // For IoTV2 batch mode, the pipe receiver may need to flush because the replica // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases - if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) { + if (!flushed[0] + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, timeoutSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -2024,7 +2036,9 @@ public static void assertDataEventuallyOnEnv( // For IoTV2 batch mode, the pipe receiver may need to flush because the replica // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases - if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) { + if (!flushed[0] + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, timeoutSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -2095,8 +2109,8 @@ public static void assertDataAlwaysOnEnv( // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases if (!flushed[0] - && System.currentTimeMillis() - startTime - > TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) { + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, consistentSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -2137,8 +2151,8 @@ public static void assertDataAlwaysOnEnv( // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases if (!flushed[0] - && System.currentTimeMillis() - startTime - > TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) { + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, consistentSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -2192,8 +2206,8 @@ public static void assertDataAlwaysOnEnv( // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases if (!flushed[0] - && System.currentTimeMillis() - startTime - > TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) { + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, consistentSeconds)) { flushed[0] = true; statement.execute("flush"); } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java index 95376af638b1a..9162c792c5adc 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java @@ -176,16 +176,22 @@ private void initSink(final Map sinkAttributes) throws Exception */ @Override protected boolean executeOnce() throws Exception { - if (isClosed.get()) { + if (isClosed.get() || shouldStopSubmittingSelf.get()) { return false; } final Event event = lastEvent != null ? lastEvent : source.supply(); - // Record the last event for retry when exception occurs + // Keep the event for retry or restart if supply() returns after the task is stopped. setLastEvent(event); + // close() may have released the last event before supply() returned. + if (isClosed.get()) { + clearReferenceCountAndReleaseLastEvent(event); + return false; + } + try { - if (event == null) { + if (event == null || shouldStopSubmittingSelf.get()) { return false; } diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtaskTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtaskTest.java new file mode 100644 index 0000000000000..2ca2aca5302c4 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtaskTest.java @@ -0,0 +1,150 @@ +/* + * 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.confignode.manager.pipe.agent.task; + +import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; +import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; +import org.apache.iotdb.commons.pipe.agent.task.subtask.PipeAbstractSinkSubtask; +import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant; +import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.apache.iotdb.pipe.api.PipeConnector; +import org.apache.iotdb.pipe.api.PipeExtractor; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.lang.reflect.Field; +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyZeroInteractions; +import static org.mockito.Mockito.when; + +public class PipeConfigNodeSubtaskTest { + + private PipeConfigNodeSubtask subtask; + private PipeExtractor source; + private PipeConnector sink; + private boolean subtaskClosed; + + @Before + public void setUp() throws Exception { + source = mock(PipeExtractor.class); + sink = mock(PipeConnector.class); + subtask = + new PipeConfigNodeSubtask( + "PipeConfigNodeSubtaskTest", + System.currentTimeMillis(), + Collections.emptyMap(), + Collections.emptyMap(), + Collections.singletonMap( + PipeSinkConstant.CONNECTOR_KEY, + BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName()), + new PipeTaskMeta(MinimumProgressIndex.INSTANCE, Integer.MIN_VALUE)); + final Field sourceField = PipeConfigNodeSubtask.class.getDeclaredField("source"); + sourceField.setAccessible(true); + sourceField.set(subtask, source); + final Field sinkField = PipeAbstractSinkSubtask.class.getDeclaredField("outputPipeSink"); + sinkField.setAccessible(true); + sinkField.set(subtask, sink); + } + + @After + public void tearDown() { + if (!subtaskClosed) { + subtask.close(); + } + } + + @Test + public void testStoppedSubtaskDoesNotPollSource() throws Exception { + assertFalse(subtask.executeOnce()); + verifyZeroInteractions(source, sink); + } + + @Test + public void testStopDuringSupplyRetainsEventUntilRestart() throws Exception { + assertStopDuringSupply(false); + } + + @Test + public void testDropDuringSupplyReleasesEvent() throws Exception { + assertStopDuringSupply(true); + } + + private void assertStopDuringSupply(final boolean drop) throws Exception { + final EnrichedEvent event = mock(EnrichedEvent.class); + final CountDownLatch supplyStarted = new CountDownLatch(1); + final CountDownLatch supplyReleased = new CountDownLatch(1); + final ExecutorService worker = Executors.newSingleThreadExecutor(); + when(source.supply()) + .thenAnswer( + invocation -> { + supplyStarted.countDown(); + assertTrue(supplyReleased.await(10, TimeUnit.SECONDS)); + return event; + }); + subtask.allowSubmittingSelf(); + try { + final Future execution = worker.submit(() -> subtask.executeOnce()); + assertTrue(supplyStarted.await(10, TimeUnit.SECONDS)); + assertTrue(subtask.disallowSubmittingSelf()); + if (drop) { + subtask.close(); + subtaskClosed = true; + } + supplyReleased.countDown(); + + assertFalse(execution.get(10, TimeUnit.SECONDS)); + assertFalse(subtask.executeOnce()); + verify(sink, never()).transfer(event); + verify(event, never()).decreaseReferenceCount(anyString(), anyBoolean()); + + if (drop) { + verify(event).clearReferenceCount(anyString()); + } else { + verifyZeroInteractions(sink); + subtask.allowSubmittingSelf(); + assertTrue(subtask.executeOnce()); + verify(source).supply(); + verify(sink).transfer(event); + verify(event).decreaseReferenceCount(anyString(), eq(true)); + } + } finally { + supplyReleased.countDown(); + worker.shutdownNow(); + assertTrue(worker.awaitTermination(10, TimeUnit.SECONDS)); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java index 9690eed5d758a..9fc46f47e695d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java @@ -556,6 +556,9 @@ public boolean internallyDecreaseResourceReferenceCount(final String holderMessa holderMessage), e); } finally { + // Retries must repeat the privilege check, so release its metadata only with the event. + tableNames = null; + treeSchemaMap = null; if (Objects.nonNull(pipeName)) { try { PipeDataNodeSinglePipeMetrics.getInstance() @@ -932,10 +935,6 @@ else if (Objects.nonNull(treeSchemaMap)) { e.getMessage()); LOGGER.warn(errorMsg, e); throw new PipeException(errorMsg, e); - } finally { - // GC useless - tableNames = null; - treeSchemaMap = null; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java index 2af8eccc17388..f91969ae32ad3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java @@ -1370,6 +1370,15 @@ public static void checkAndEnrichSourceUser( .FAILED_TO_S_PIPE_S_IN_IOTDB_SOURCE_PASSWORD_MUST_BE_SET_WHEN_THE_USERNAME_IS_SPECIFIED, isAlter ? DataNodeQueryMessages.ALTER : DataNodeQueryMessages.CREATE, pipeName)); + } else { + updatePipeUserIdFromUsername( + replacedSourceAttributes, + sourceParameters, + PipeSourceConstant.SOURCE_IOTDB_USER_ID, + PipeSourceConstant.EXTRACTOR_IOTDB_USER_KEY, + PipeSourceConstant.SOURCE_IOTDB_USER_KEY, + PipeSourceConstant.EXTRACTOR_IOTDB_USERNAME_KEY, + PipeSourceConstant.SOURCE_IOTDB_USERNAME_KEY); } } @@ -1442,6 +1451,15 @@ public static void checkAndEnrichSinkUser( .FAILED_TO_S_PIPE_S_IN_WRITE_BACK_SINK_PASSWORD_MUST_BE_SET_WHEN_THE_USERNAME_IS, isAlter ? DataNodeQueryMessages.ALTER : DataNodeQueryMessages.CREATE, pipeName)); + } else { + updatePipeUserIdFromUsername( + connectorAttributes, + connectorParameters, + PipeSinkConstant.SINK_IOTDB_USER_ID, + PipeSinkConstant.CONNECTOR_IOTDB_USER_KEY, + PipeSinkConstant.SINK_IOTDB_USER_KEY, + PipeSinkConstant.CONNECTOR_IOTDB_USERNAME_KEY, + PipeSinkConstant.SINK_IOTDB_USERNAME_KEY); } } @@ -1510,6 +1528,14 @@ public static void markSourceAuthenticationAsExplicitIfNecessary( PipeSourceConstant.SOURCE_IOTDB_PASSWORD_KEY)) { sourceAttributes.put( SystemConstant.SOURCE_AUTHENTICATION_INJECTED_KEY, Boolean.FALSE.toString()); + updatePipeUserIdFromUsername( + sourceAttributes, + sourceParameters, + PipeSourceConstant.SOURCE_IOTDB_USER_ID, + PipeSourceConstant.EXTRACTOR_IOTDB_USER_KEY, + PipeSourceConstant.SOURCE_IOTDB_USER_KEY, + PipeSourceConstant.EXTRACTOR_IOTDB_USERNAME_KEY, + PipeSourceConstant.SOURCE_IOTDB_USERNAME_KEY); } } @@ -1524,6 +1550,26 @@ public static void markSinkAuthenticationAsExplicitIfNecessary( PipeSinkConstant.CONNECTOR_IOTDB_PASSWORD_KEY, PipeSinkConstant.SINK_IOTDB_PASSWORD_KEY)) { sinkAttributes.put(SystemConstant.SINK_AUTHENTICATION_INJECTED_KEY, Boolean.FALSE.toString()); + updatePipeUserIdFromUsername( + sinkAttributes, + sinkParameters, + PipeSinkConstant.SINK_IOTDB_USER_ID, + PipeSinkConstant.CONNECTOR_IOTDB_USER_KEY, + PipeSinkConstant.SINK_IOTDB_USER_KEY, + PipeSinkConstant.CONNECTOR_IOTDB_USERNAME_KEY, + PipeSinkConstant.SINK_IOTDB_USERNAME_KEY); + } + } + + private static void updatePipeUserIdFromUsername( + final Map attributes, + final PipeParameters parameters, + final String userIdAttributeKey, + final String... usernameAttributeKeys) { + final String username = parameters.getStringByKeys(usernameAttributeKeys); + if (username != null) { + attributes.put( + userIdAttributeKey, String.valueOf(AuthorityChecker.getUserId(username).orElse(-1L))); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTaskRetention.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTaskRetention.java index 828df844ed7a0..6197bbccc9895 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTaskRetention.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTaskRetention.java @@ -62,14 +62,7 @@ final class LoadTaskRetention { /** Checks whether the staged directory must be retained for follower WAL synchronization. */ boolean mustRetain(final long searchIndex) { - // IoTConsensusV2 replicates through pipe, so no replica reads these bytes back out of the WAL - // and there is no index to wait for: the staged directory can go as soon as the task ends. - if (LoadConsensusCompat.usesIoTConsensusV2()) { - return false; - } - final long safelyDeletedSearchIndex = currentSafelyDeletedSearchIndex(); - return safelyDeletedSearchIndex != ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX - && safelyDeletedSearchIndex < searchIndex; + return currentSafelyDeletedSearchIndex() < searchIndex; } /** Takes ownership of a staged directory, postponing deletion until WAL watermarks allow it. */ @@ -122,14 +115,14 @@ void release() { } final long safelyDeletedSearchIndex = currentSafelyDeletedSearchIndex(); - final long boundary = - safelyDeletedSearchIndex == ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX - ? Long.MAX_VALUE - : safelyDeletedSearchIndex; + // The default V1 watermark means no replica progress has been confirmed yet. + if (safelyDeletedSearchIndex == ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX) { + return; + } for (final Map.Entry entry : loadId2RetainedTaskDir.entrySet()) { final RetainedTaskDir retained = entry.getValue(); - if (retained.reachedSearchIndex() <= boundary + if (retained.reachedSearchIndex() <= safelyDeletedSearchIndex && loadId2RetainedTaskDir.remove(entry.getKey(), retained)) { LOGGER.info( StorageEngineMessages @@ -165,10 +158,15 @@ private void ensureWalListenerRegistered() { } private long currentSafelyDeletedSearchIndex() { + // V2 replicates through pipe. Other protocols without a V1 WAL (for example Ratis) do not + // expand staged references from this node's WAL either, so neither has a watermark to wait for. + if (LoadConsensusCompat.usesIoTConsensusV2()) { + return Long.MAX_VALUE; + } return dataRegion .getWALNode() .map(IWALNode::getSafelyDeletedSearchIndex) - .orElse(ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX); + .orElse(Long.MAX_VALUE); } // ------------------------------------------------------------------------- diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileCleaner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileCleaner.java index ee440de7790f7..93a537c1e4aca 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileCleaner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileCleaner.java @@ -350,14 +350,22 @@ private static boolean isComplete(final File taskDir) { } private static boolean replicasReached(final DataRegion dataRegion, final long searchIndex) { - final long safelyDeletedSearchIndex = - dataRegion - .getWALNode() - .map(wal -> wal.getSafelyDeletedSearchIndex()) - .orElse(ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX); - - return safelyDeletedSearchIndex == ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX - || safelyDeletedSearchIndex >= searchIndex; + // IoTConsensusV2 and consensus implementations without a WAL replicate the staged payload + // directly, so no follower can read it back from this node's WAL. IoTConsensus V1, however, + // must report a real watermark before its staged references are safe to delete. + if (LoadConsensusCompat.usesIoTConsensusV2()) { + return true; + } + return dataRegion + .getWALNode() + .map( + wal -> { + final long safelyDeletedSearchIndex = wal.getSafelyDeletedSearchIndex(); + return safelyDeletedSearchIndex + != ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX + && safelyDeletedSearchIndex >= searchIndex; + }) + .orElse(true); } // ------------------------------------------------------------------------- diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java index 2c81646c53de3..d3d9efb1345a5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java @@ -24,12 +24,17 @@ import org.apache.iotdb.commons.consensus.index.ProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; import org.apache.iotdb.commons.file.SystemFileFactory; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.commons.utils.PathUtils; import org.apache.iotdb.commons.utils.RetryUtils; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.load.LoadFileException; import org.apache.iotdb.db.i18n.StorageEngineMessages; +import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusOp; +import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; import org.apache.iotdb.db.storageengine.dataregion.DataRegion; import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; @@ -502,6 +507,18 @@ void prepare( fileLength)); } } + // Rebuild from the staged devices so recovered writers also retain the table schemas that + // Pipe needs to enumerate table data in the sealed file. + if (PathUtils.isTableModelDatabase(dataRegion.getDatabaseName())) { + for (final String tableName : writer.getTableNames()) { + final TsTable table = + DataNodeTableCache.getInstance() + .getTable(dataRegion.getDatabaseName(), tableName, false); + if (table != null) { + writer.registerTableSchema(TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute(table)); + } + } + } writer.close(); final TsFileResource tsFileResource = timePartition2Resource.get(entry.getKey()); @@ -516,7 +533,7 @@ void prepare( /** Formally imports sealed staged TsFiles into the target storage engine data regions. */ void loadAll(final boolean isGeneratedByPipe, final boolean deleteStagedSource) - throws LoadFileException { + throws IOException, LoadFileException { for (final Map.Entry entry : timePartition2Resource.entrySet()) { final DataRegion targetDataRegion = dataRegion; @@ -531,6 +548,14 @@ void loadAll(final boolean isGeneratedByPipe, final boolean deleteStagedSource) uuidDirName(tsFileResource.getTsFile()))); } + if (LoadConsensusCompat.usesIoTConsensusV2()) { + // V2 only replicates files carrying this node's progress. The coordinator's index may + // belong to another node and precede writes between PREPARE and COMMIT. Assign progress + // at import, and persist it so recovery keeps the file eligible for replication. + PipeDataNodeAgent.runtime().assignProgressIndexForTsFileLoad(tsFileResource); + tsFileResource.serialize(); + } + targetDataRegion.loadNewTsFile( tsFileResource, deleteStagedSource, isGeneratedByPipe, false, Optional.empty()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/tsfile/write/TsFilePrecalculatedChunkWriter.java b/iotdb-core/datanode/src/main/java/org/apache/tsfile/write/TsFilePrecalculatedChunkWriter.java index a6c0d996c9817..b48b7d14cb17d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/tsfile/write/TsFilePrecalculatedChunkWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/tsfile/write/TsFilePrecalculatedChunkWriter.java @@ -31,6 +31,7 @@ import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.MeasurementMetadataIndexEntry; import org.apache.tsfile.file.metadata.MetadataIndexNode; +import org.apache.tsfile.file.metadata.TableSchema; import org.apache.tsfile.file.metadata.TimeseriesMetadata; import org.apache.tsfile.file.metadata.TsFileMetadata; import org.apache.tsfile.file.metadata.enums.MetadataIndexNodeType; @@ -53,6 +54,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; import java.util.TreeMap; import java.util.stream.Collectors; @@ -80,6 +82,7 @@ public class TsFilePrecalculatedChunkWriter implements AutoCloseable { private final File file; private final Map>> device2MetadataMap = new TreeMap<>(); + private final Map tableSchemaMap = new TreeMap<>(); private long maxPhysicalFileSize; private boolean sealed; @@ -385,6 +388,7 @@ public void close() throws IOException { } tsFileMetadata.setTableMetadataIndexNodeMap(tableIndexNodeMap); + tsFileMetadata.setTableSchemaMap(tableSchemaMap); tsFileMetadata.setMetaOffset(metaOffset); tsFileMetadata.setBloomFilter(bloomFilter); @@ -446,6 +450,16 @@ public Map> getChunkMetadataListMap() { return result; } + public Set getTableNames() { + return device2MetadataMap.keySet().stream() + .map(IDeviceID::getTableName) + .collect(Collectors.toSet()); + } + + public void registerTableSchema(final TableSchema tableSchema) { + tableSchemaMap.put(tableSchema.getTableName(), tableSchema); + } + private String getFileForLog() { return file == null ? "" : file.getAbsolutePath(); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java index 5d5be1dbadee9..10f642527dcd9 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java @@ -80,7 +80,8 @@ public void testAuthCheck() throws Exception { TsFileNameGenerator.generateNewTsFilePath( TestConstant.BASE_OUTPUT_PATH + IoTDBConstant.SEQUENCE_FOLDER_NAME, 1, 1, 1, 1)); try { - AuthorityChecker.setAccessControl(new TestAccessControl()); + final TestAccessControl accessControl = Mockito.spy(new TestAccessControl()); + AuthorityChecker.setAccessControl(accessControl); final TsFileResource resource = new TsFileResource(file); try (CompactionTestFileWriter writer = new CompactionTestFileWriter(resource)) { @@ -126,6 +127,18 @@ public void testAuthCheck() throws Exception { Long.MIN_VALUE, Long.MAX_VALUE); Assert.assertThrows(AccessDeniedException.class, tableEvent::throwIfNoPrivilege); + // A denied event must remain retryable until SELECT is granted. + Assert.assertThrows(AccessDeniedException.class, tableEvent::throwIfNoPrivilege); + Mockito.doReturn(true) + .when(accessControl) + .checkCanSelectFromTable4Pipe(Mockito.anyString(), Mockito.any(), Mockito.any()); + tableEvent.throwIfNoPrivilege(); + // Processing can fail after a successful check, so the next attempt must check again. + tableEvent.throwIfNoPrivilege(); + Mockito.doReturn(false) + .when(accessControl) + .checkCanSelectFromTable4Pipe(Mockito.anyString(), Mockito.any(), Mockito.any()); + Assert.assertThrows(AccessDeniedException.class, tableEvent::throwIfNoPrivilege); tableEvent.close(); final PipeTsFileInsertionEvent treeEvent = @@ -157,6 +170,8 @@ public void testAuthCheck() throws Exception { treeEvent.setTreeSchemaMap(Collections.singletonMap(deviceID, new String[] {"s0", "s1"})); Assert.assertThrows(AccessDeniedException.class, treeEvent::throwIfNoPrivilege); + // Retrying a realtime event must not fall back to the historical coarse filter. + Assert.assertThrows(AccessDeniedException.class, treeEvent::throwIfNoPrivilege); treeEvent.close(); } finally { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/PipeAuthenticationIdentityTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/PipeAuthenticationIdentityTest.java new file mode 100644 index 0000000000000..741dd0171ba26 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/PipeAuthenticationIdentityTest.java @@ -0,0 +1,131 @@ +/* + * 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.queryengine.plan.execution.config; + +import org.apache.iotdb.commons.audit.UserEntity; +import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant; +import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant; +import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; +import org.apache.iotdb.db.auth.AuthorityChecker; +import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.powermock.core.classloader.annotations.PowerMockIgnore; +import org.powermock.core.classloader.annotations.PrepareForTest; +import org.powermock.modules.junit4.PowerMockRunner; + +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; + +import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.when; +import static org.powermock.api.mockito.PowerMockito.mockStatic; + +@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"}) +@RunWith(PowerMockRunner.class) +@PrepareForTest(AuthorityChecker.class) +public class PipeAuthenticationIdentityTest { + + @Test + public void testModifiedCredentialsReplaceInjectedSuperUserId() { + mockStatic(AuthorityChecker.class); + when(AuthorityChecker.getUserId("thulab")).thenReturn(Optional.of(10001L)); + + final Map sourceAttributes = new HashMap<>(); + TableConfigTaskVisitor.checkAndEnrichSourceUser( + "pipe", sourceAttributes, new UserEntity(0L, "root", "host"), false); + + final Map sourceChanges = new HashMap<>(); + sourceChanges.put("username", "thulab"); + sourceChanges.put("password", "password"); + TableConfigTaskVisitor.markSourceAuthenticationAsExplicitIfNecessary(sourceChanges); + new PipeParameters(sourceAttributes) + .addOrReplaceEquivalentAttributes(new PipeParameters(sourceChanges)); + + final Map sinkAttributes = new HashMap<>(); + sinkAttributes.put(PipeSinkConstant.SINK_KEY, "write-back-sink"); + TableConfigTaskVisitor.checkAndEnrichSinkUser( + "pipe", sinkAttributes, new UserEntity(0L, "root", "host"), false); + + final Map sinkChanges = new HashMap<>(); + sinkChanges.put(PipeSinkConstant.CONNECTOR_IOTDB_USER_KEY, "thulab"); + sinkChanges.put(PipeSinkConstant.CONNECTOR_IOTDB_PASSWORD_KEY, "password"); + TableConfigTaskVisitor.markSinkAuthenticationAsExplicitIfNecessary(sinkChanges); + new PipeParameters(sinkAttributes) + .addOrReplaceEquivalentAttributes(new PipeParameters(sinkChanges)); + + assertEquals("10001", sourceAttributes.get(PipeSourceConstant.SOURCE_IOTDB_USER_ID)); + assertEquals( + Boolean.FALSE.toString(), + sourceAttributes.get(SystemConstant.SOURCE_AUTHENTICATION_INJECTED_KEY)); + assertEquals("10001", sinkAttributes.get(PipeSinkConstant.SINK_IOTDB_USER_ID)); + assertEquals( + Boolean.FALSE.toString(), + sinkAttributes.get(SystemConstant.SINK_AUTHENTICATION_INJECTED_KEY)); + } + + @Test + public void testExplicitCredentialsUseTheirOwnUserIdOnCreateAndReplace() { + mockStatic(AuthorityChecker.class); + when(AuthorityChecker.getUserId("thulab")).thenReturn(Optional.of(10001L)); + + for (final boolean isAlter : new boolean[] {false, true}) { + final Map sourceAttributes = new HashMap<>(); + sourceAttributes.put(PipeSourceConstant.EXTRACTOR_IOTDB_USER_KEY, "thulab"); + sourceAttributes.put(PipeSourceConstant.EXTRACTOR_IOTDB_PASSWORD_KEY, "password"); + TableConfigTaskVisitor.checkAndEnrichSourceUser( + "pipe", sourceAttributes, new UserEntity(0L, "root", "host"), isAlter); + + final Map sinkAttributes = new HashMap<>(); + sinkAttributes.put(PipeSinkConstant.SINK_KEY, "write-back-sink"); + sinkAttributes.put(PipeSinkConstant.SINK_IOTDB_USERNAME_KEY, "thulab"); + sinkAttributes.put(PipeSinkConstant.SINK_IOTDB_PASSWORD_KEY, "password"); + TableConfigTaskVisitor.checkAndEnrichSinkUser( + "pipe", sinkAttributes, new UserEntity(0L, "root", "host"), isAlter); + + assertEquals("10001", sourceAttributes.get(PipeSourceConstant.SOURCE_IOTDB_USER_ID)); + assertEquals("10001", sinkAttributes.get(PipeSinkConstant.SINK_IOTDB_USER_ID)); + } + } + + @Test + public void testPasswordOnlyModificationPreservesExistingUserId() { + final Map sourceAttributes = new HashMap<>(); + sourceAttributes.put(PipeSourceConstant.SOURCE_IOTDB_USER_ID, "10001"); + final Map sourceChanges = new HashMap<>(); + sourceChanges.put("password", "new-password"); + TableConfigTaskVisitor.markSourceAuthenticationAsExplicitIfNecessary(sourceChanges); + new PipeParameters(sourceAttributes) + .addOrReplaceEquivalentAttributes(new PipeParameters(sourceChanges)); + + final Map sinkAttributes = new HashMap<>(); + sinkAttributes.put(PipeSinkConstant.SINK_IOTDB_USER_ID, "10001"); + final Map sinkChanges = new HashMap<>(); + sinkChanges.put("password", "new-password"); + TableConfigTaskVisitor.markSinkAuthenticationAsExplicitIfNecessary(sinkChanges); + new PipeParameters(sinkAttributes) + .addOrReplaceEquivalentAttributes(new PipeParameters(sinkChanges)); + + assertEquals("10001", sourceAttributes.get(PipeSourceConstant.SOURCE_IOTDB_USER_ID)); + assertEquals("10001", sinkAttributes.get(PipeSinkConstant.SINK_IOTDB_USER_ID)); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileCleanerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileCleanerTest.java index ca25a96ad4bbe..01482f2dfe045 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileCleanerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileCleanerTest.java @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.conf.ConfigurationFileUtils; import org.apache.iotdb.commons.conf.TrimProperties; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.consensus.iot.log.ConsensusReqReader; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusNode; @@ -186,6 +187,47 @@ public void testAnIdleTaskIsReclaimedOnlyOnceItsPiecesAreReplicated() throws Exc } } + @Test + public void testAFinishedTaskWaitsForTheFirstWatermark() throws Exception { + final String loadId = "unreported-watermark-task"; + final File taskDir = new File(tempDir, loadId); + Assert.assertTrue(taskDir.mkdirs()); + stageOneChunk(taskDir, 5L); + final LoadTsFileCleaner cleaner = LoadTsFileCleaner.getInstance(); + cleaner.register(loadId, dataRegion, taskDir, 7L, LoadTsFileConsensusOp.ABORT); + + walWatermark.set(ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX); + cleaner.sweepRegisteredTasks(); + Assert.assertTrue("an unreported watermark does not confirm replication", taskDir.exists()); + + walWatermark.set(5L); + cleaner.sweepRegisteredTasks(); + Assert.assertTrue("the terminal command is not replicated yet", taskDir.exists()); + + walWatermark.set(7L); + cleaner.sweepRegisteredTasks(); + Assert.assertFalse("the terminal command is replicated", taskDir.exists()); + } + + @Test + public void testIoTConsensusV2DoesNotWaitForAWALWatermark() throws Exception { + final String loadId = "v2-finished-task"; + final File taskDir = new File(tempDir, loadId); + Assert.assertTrue(taskDir.mkdirs()); + stageOneChunk(taskDir, 5L); + final LoadTsFileCleaner cleaner = LoadTsFileCleaner.getInstance(); + cleaner.register(loadId, dataRegion, taskDir, 7L, LoadTsFileConsensusOp.ABORT); + walWatermark.set(ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX); + + LoadConsensusCompat.setIoTConsensusV2ForTest(Boolean.TRUE); + try { + cleaner.sweepRegisteredTasks(); + Assert.assertFalse("V2 does not expand staged references from the WAL", taskDir.exists()); + } finally { + LoadConsensusCompat.setIoTConsensusV2ForTest(null); + } + } + @Test public void testAFinishedCommittedTaskIsReleasedOnceItsStagedFilesAreComplete() throws Exception { final String loadId = "finished-task"; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java index c98b3afa19583..a3d2473316760 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java @@ -20,15 +20,22 @@ package org.apache.iotdb.db.storageengine.load; import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot; +import org.apache.iotdb.commons.consensus.index.ProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.consensus.iot.log.ConsensusReqReader; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.load.LoadFileException; +import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; +import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; +import org.apache.iotdb.db.pipe.processor.iotconsensusv2.IoTConsensusV2Processor; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusOp; import org.apache.iotdb.db.queryengine.plan.scheduler.load.ChunkOffsetCalculator; import org.apache.iotdb.db.storageengine.dataregion.DataRegion; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus; import org.apache.iotdb.db.storageengine.dataregion.wal.node.IWALNode; import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode; @@ -634,10 +641,8 @@ private static NonAlignedChunkData[] twoChunksOfTheSamePartition() { } @Test - public void testAbortedConsensusTaskReleasesStagedFilesWhenNoWatermarkIsReported() - throws Exception { - // Without a consensus watermark (single replica) nothing can read the staged bytes back, so the - // directory of a finished task is deleted right away. + public void testAbortedConsensusTaskReleasesStagedFilesWithoutAWAL() throws Exception { + // Protocols without a V1 WAL do not expand staged references from the WAL. final String loadId = "aborted-load"; final LoadTsFileManager manager = new LoadTsFileManager(dataRegion); final List refs = @@ -660,8 +665,8 @@ public void testFinishedConsensusTaskKeepsStagedFilesUntilTheWatermarkAdvances() try (final WALNode walNode = new WALNode("load-retention-test", new File(tempDir, "wal").getAbsolutePath())) { Mockito.when(dataRegion.getWALNode()).thenReturn(Optional.of(walNode)); - // The watermark lags behind the command, so a follower may still need the staged bytes. - walNode.setSafelyDeletedSearchIndex(5L); + // Until V1 reports its first watermark, a follower may still need the staged bytes. + walNode.setSafelyDeletedSearchIndex(ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX); final LoadTsFileManager manager = new LoadTsFileManager(dataRegion); final String loadId = "retained-load"; @@ -682,6 +687,10 @@ public void testFinishedConsensusTaskKeepsStagedFilesUntilTheWatermarkAdvances() assertTrue(progress.isFile()); assertNotNull(LoadTsFileProgress.readTerminal(progress)); + // A reported watermark that still lags behind the command must keep the staged bytes too. + walNode.setSafelyDeletedSearchIndex(5L); + assertTrue(staged.isFile()); + // Once every follower got past the command, the directory is released. walNode.setSafelyDeletedSearchIndex(10L); assertFalse(staged.exists()); @@ -708,6 +717,27 @@ public void testStagedDirectoryOfFinishedTaskIsDeletedOnRecovery() throws Except assertFalse(manager.prepare(prepareNode("finished-load"), Collections.emptyMap())); } + @Test + public void testRecoveredFinishedTaskWaitsForTheFirstWatermark() throws Exception { + useWatermarkWALNode(ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX); + final File taskDir = new File(tempDir, "root.load_manager_test-0/recovered-finished-load"); + assertTrue(taskDir.mkdirs()); + final File stagedFile = new File(taskDir, "123.tsfile"); + Files.write(stagedFile.toPath(), new byte[] {1, 2, 3}); + new LoadTsFileProgress(stagedFile).recordTerminal(LoadTsFileConsensusOp.ABORT, 7L); + + final LoadTsFileManager manager = new LoadTsFileManager(dataRegion); + try { + assertTrue("a restart must keep bytes until replication is confirmed", stagedFile.isFile()); + assertFalse(manager.prepare(prepareNode(taskDir.getName()), Collections.emptyMap())); + + advanceWatermark(7L); + assertFalse(taskDir.exists()); + } finally { + manager.stop(); + } + } + @Test public void testIoTConsensusV2WaitsForNoIndexBeforeReleasing() { LoadConsensusCompat.setIoTConsensusV2ForTest(Boolean.TRUE); @@ -774,11 +804,85 @@ public void testStagedDirectoryWithoutTerminalRecordIsKeptOnRecovery() throws Ex assertFalse(manager.prepare(prepareNode("in-progress-load"), Collections.emptyMap())); } + @Test + public void testIoTConsensusV2CommitUsesReceivingNodeProgress() throws Exception { + assertCommittedProgress(true, config.getDataNodeId() + 1); + } + + @Test + public void testIoTConsensusV2CommitAdvancesPreparedProgress() throws Exception { + assertCommittedProgress(true, config.getDataNodeId()); + } + + @Test + public void testOtherConsensusCommitPreservesCoordinatorProgress() throws Exception { + assertCommittedProgress(false, config.getDataNodeId() + 1); + } + + private void assertCommittedProgress(final boolean useIoTConsensusV2, final int coordinatorNodeId) + throws Exception { + LoadConsensusCompat.setIoTConsensusV2ForTest(useIoTConsensusV2); + final String loadId = "load-progress"; + final LoadTsFileManager manager = new LoadTsFileManager(dataRegion); + try { + emulateStagedFileImport(); + final LoadTsFileConsensusNode piece = stagedPiece(loadId, laidOutChunk()); + manager.writePiece(piece); + final File staged = new File(tempDir, piece.getPieceRefs().get(0).getRelativePath()); + final RecoverProgressIndex preparedLocalProgress = + PipeDataNodeAgent.runtime().getNextProgressIndexForTsFileLoad(); + final ProgressIndex coordinatorProgress = + new RecoverProgressIndex( + coordinatorNodeId, + preparedLocalProgress.getDataNodeId2LocalIndex().get(config.getDataNodeId())); + assertTrue( + manager.prepare( + prepareNode(loadId), + Collections.singletonMap(new TTimePartitionSlot(0L), coordinatorProgress))); + + // A local write after PREPARE must precede the progress assigned when the file is imported. + final ProgressIndex interveningWrite = + PipeDataNodeAgent.runtime().getNextProgressIndexForTsFileLoad(); + final LoadTsFileConsensusNode commit = commitNode(loadId); + assertTrue(manager.loadAll(commit, Collections.emptyMap())); + + final TsFileResource imported = + new TsFileResource(new File(new File(tempDir, "data"), staged.getName())); + imported.deserialize(); + imported.setStatus(TsFileResourceStatus.NORMAL); + final ProgressIndex persistedProgress = imported.getMaxProgressIndexAfterClose(); + final PipeTsFileInsertionEvent event = Mockito.mock(PipeTsFileInsertionEvent.class); + Mockito.when(event.forceGetProgressIndex()).thenReturn(persistedProgress); + assertEquals( + "A V2 LOAD must remain eligible for consensus replication after resource recovery", + useIoTConsensusV2, + IoTConsensusV2Processor.isShouldReplicate(event)); + if (useIoTConsensusV2) { + assertTrue(persistedProgress.isAfter(interveningWrite)); + } else { + assertEquals(coordinatorProgress, persistedProgress); + } + } finally { + manager.stop(); + LoadConsensusCompat.setIoTConsensusV2ForTest(null); + } + } + @Test public void testCommitKeepsTheStagedBytesWhileAReplicaMayStillNeedThem() throws Exception { // The watermark lags behind the COMMIT command, so a replica that catches up from this node's // WAL still has to expand the pieces of the task from the staging directory. - useWatermarkWALNode(5L); + assertCommitKeepsTheStagedBytes(5L); + } + + @Test + public void testCommitKeepsTheStagedBytesBeforeTheFirstWatermark() throws Exception { + assertCommitKeepsTheStagedBytes(ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX); + } + + private void assertCommitKeepsTheStagedBytes(final long safelyDeletedSearchIndex) + throws Exception { + useWatermarkWALNode(safelyDeletedSearchIndex); emulateStagedFileImport(); final String loadId = "committed-load"; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileTableSchemaTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileTableSchemaTest.java new file mode 100644 index 0000000000000..306f3050c4882 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileTableSchemaTest.java @@ -0,0 +1,181 @@ +/* + * 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; + +import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TagColumnSchema; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; +import org.apache.iotdb.db.pipe.event.common.tsfile.parser.table.TsFileInsertionEventTableParser; +import org.apache.iotdb.db.queryengine.plan.scheduler.load.ChunkOffsetCalculator; +import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; +import org.apache.iotdb.db.storageengine.dataregion.DataRegion; +import org.apache.iotdb.db.storageengine.load.splitter.ChunkData; +import org.apache.iotdb.db.storageengine.load.splitter.TsFileData; +import org.apache.iotdb.db.storageengine.load.splitter.TsFileSplitter; +import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent; + +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.file.metadata.enums.CompressionType; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.write.TsFileWriter; +import org.apache.tsfile.write.record.Tablet; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.mockito.Mockito; + +import java.io.File; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +import static org.junit.Assert.assertEquals; + +public class LoadTsFileTableSchemaTest { + + private static final String DATABASE = "load_table_schema_test"; + private static final String TABLE = "measurements"; + private static final int ROW_COUNT = 20; + + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private String[] originalLoadDirs; + private DataRegion dataRegion; + private TsTable table; + + @Before + public void setUp() throws Exception { + originalLoadDirs = IoTDBDescriptor.getInstance().getConfig().getLoadTsFileDirs(); + IoTDBDescriptor.getInstance() + .getConfig() + .setLoadTsFileDirs(new String[] {temporaryFolder.getRoot().getAbsolutePath()}); + dataRegion = Mockito.mock(DataRegion.class); + Mockito.when(dataRegion.getDatabaseName()).thenReturn(DATABASE); + Mockito.when(dataRegion.getDataRegionIdString()).thenReturn("0"); + Mockito.when(dataRegion.getNonSystemDatabaseName()).thenReturn(Optional.of(DATABASE)); + + table = new TsTable(TABLE); + table.addColumnSchema(new TagColumnSchema("device", TSDataType.STRING)); + table.addColumnSchema( + new FieldColumnSchema("value", TSDataType.INT32, TSEncoding.PLAIN, CompressionType.LZ4)); + DataNodeTableCache.getInstance().preUpdateTable(DATABASE, table, null); + DataNodeTableCache.getInstance().commitUpdateTable(DATABASE, TABLE, null); + } + + @After + public void tearDown() { + DataNodeTableCache.getInstance().invalid(DATABASE); + IoTDBDescriptor.getInstance().getConfig().setLoadTsFileDirs(originalLoadDirs); + } + + @Test + public void testLoadedTableRemainsReadableByPipe() throws Exception { + assertLoadedTableReadable(false, false); + } + + @Test + public void testRecoveredTableRemainsReadableByPipe() throws Exception { + assertLoadedTableReadable(true, false); + } + + @Test + public void testLoadedTableWithNullTagsRemainsReadableByPipe() throws Exception { + assertLoadedTableReadable(false, true); + } + + private void assertLoadedTableReadable(final boolean recover, final boolean nullTags) + throws Exception { + final File source = new File(temporaryFolder.getRoot(), "source.tsfile"); + final TableSchema schema = TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute(table); + try (final TsFileWriter writer = new TsFileWriter(source)) { + writer.registerTableSchema(schema); + final Tablet tablet = + new Tablet( + TABLE, + Arrays.asList("device", "value"), + Arrays.asList(TSDataType.STRING, TSDataType.INT32), + Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD), + ROW_COUNT); + for (int i = 0; i < ROW_COUNT; i++) { + tablet.addTimestamp(i, i); + tablet.addValue("device", i, nullTags ? null : "d" + i % 2); + tablet.addValue("value", i, i); + } + writer.writeTable(tablet); + } + + final File taskDir = temporaryFolder.newFolder("task"); + TsFileWriterManager manager = new TsFileWriterManager(dataRegion, taskDir); + try { + final ChunkOffsetCalculator calculator = new ChunkOffsetCalculator(); + final List pieces = new ArrayList<>(); + new TsFileSplitter( + source, + data -> { + calculator.assign((ChunkData) data); + pieces.add(data); + return true; + }) + .splitTsFileByDataPartition(); + manager.writePiece(pieces); + if (recover) { + manager.close(true); + manager = new TsFileWriterManager(dataRegion, taskDir).recoverFromDisk(); + } + manager.prepare(false, Collections.emptyMap(), Collections.emptyMap()); + + final File staged = new File(taskDir, DATABASE + "-0-0.tsfile"); + int rowCount = 0; + try (final TsFileInsertionEventTableParser parser = + new TsFileInsertionEventTableParser( + staged, + new TablePattern(true, null, null), + Long.MIN_VALUE, + Long.MAX_VALUE, + null, + null, + null, + false)) { + for (final TabletInsertionEvent event : parser.toTabletInsertionEvents()) { + final PipeRawTabletInsertionEvent tabletEvent = (PipeRawTabletInsertionEvent) event; + rowCount += tabletEvent.convertToTablet().getRowSize(); + tabletEvent.clearReferenceCount(getClass().getName()); + } + } + assertEquals(ROW_COUNT, rowCount); + try (final TsFileSequenceReader reader = new TsFileSequenceReader(staged.getPath())) { + assertEquals(schema, reader.getTableSchemaMap().get(TABLE)); + } + } finally { + manager.close(); + } + } +}