From 7249a7f31110234fad831c4c9188bb740a4240e3 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 23 Oct 2026 12:53:06 +0800 Subject: [PATCH 1/3] [Test] Add daily IT coverage for logical backup --- .github/workflows/logical-backup-daily-it.yml | 48 ++++++ .../manual/IoTDBPipeLogicalBackupIT.java | 141 ++++++++++++++++++ 2 files changed, 189 insertions(+) create mode 100644 .github/workflows/logical-backup-daily-it.yml diff --git a/.github/workflows/logical-backup-daily-it.yml b/.github/workflows/logical-backup-daily-it.yml new file mode 100644 index 0000000000000..21be30edbc3bb --- /dev/null +++ b/.github/workflows/logical-backup-daily-it.yml @@ -0,0 +1,48 @@ +name: Logical Backup Daily IT + +on: + pull_request: + branches: + - feat/pipe-logical-backup-sink + paths: + - .github/workflows/logical-backup-daily-it.yml + - integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeLogicalBackupIT.java + +concurrency: + group: logical-backup-daily-it-${{ github.event.pull_request.number }} + cancel-in-progress: true + +jobs: + logical-backup: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v5 + - name: Set up JDK 17 + uses: actions/setup-java@v5 + with: + distribution: corretto + java-version: 17 + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + - name: Cache Maven packages + uses: actions/cache@v5 + with: + path: ~/.m2 + key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }} + restore-keys: ${{ runner.os }}-m2- + - name: Run logical backup daily IT + run: >- + mvn -B -ntp clean verify + -Pwith-integration-tests + -DskipUTs + -Dit.test=IoTDBPipeLogicalBackupIT + -Dfailsafe.failIfNoSpecifiedTests=false + -DintegrationTest.forkCount=1 + -DConfigNodeMaxHeapSize=256 + -DDataNodeMaxHeapSize=1024 + -DDataNodeMaxDirectMemorySize=768 + -DClusterConfigurations=ScalableSingleNodeMode,ScalableSingleNodeMode + -Drat.skip=true + -pl integration-test + -am + -PMultiClusterIT2DualTreeManual diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeLogicalBackupIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeLogicalBackupIT.java index 4629bcb887506..5b1d9ab0e3915 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeLogicalBackupIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeLogicalBackupIT.java @@ -39,6 +39,7 @@ import java.io.ByteArrayInputStream; import java.io.InputStream; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; import java.nio.file.Path; import java.sql.Connection; import java.sql.ResultSet; @@ -258,6 +259,146 @@ public void testDumpPackPreflightRestoreWithScopeAndTimeFilters() throws Excepti } } + @Test + public void testTreeContinuousCapturePreservesFilteredDeletionRange() throws Exception { + TestUtils.executeNonQueries( + senderEnv, + Arrays.asList( + "create database root.logical_backup_delete", + "create timeseries root.logical_backup_delete.d.s with datatype=INT32,encoding=PLAIN", + "insert into root.logical_backup_delete.d(time,s) values(1,10)", + "flush"), + null); + final String pipeName = "logical_backup_tree_delete"; + final CompletableFuture writes = + CompletableFuture.runAsync( + () -> { + try { + Awaitility.await() + .atMost(30, TimeUnit.SECONDS) + .until( + () -> { + try (final Connection connection = senderEnv.getConnection(); + final Statement statement = connection.createStatement(); + final ResultSet result = + statement.executeQuery("show pipe " + pipeName)) { + return result.next(); + } + }); + try (final Connection connection = senderEnv.getConnection(); + final Statement statement = connection.createStatement()) { + statement.execute( + "insert into root.logical_backup_delete.d(time,s) values(2,20)"); + statement.execute( + "insert into root.logical_backup_delete.d(time,s) values(3,30)"); + statement.execute("delete from root.logical_backup_delete.d.s where time <= 2"); + } + } catch (final Exception e) { + throw new CompletionException(e); + } + }); + final Path parent = temporaryFolder.getRoot().toPath().toAbsolutePath(); + runConnected( + senderEnv, + "dump", + "--backup-dir", + parent.toString(), + "--backup-id", + "tree-delete", + "--pipe-name", + pipeName, + "--scope", + "schema,data", + "--model", + "tree", + "--include", + "root.logical_backup_delete.**", + "--start-time", + "1", + "--end-time", + "3", + "--mode", + "continuous", + "--until", + Instant.now().plusSeconds(20).toString(), + "--timeout-seconds", + "180", + "--poll-interval-seconds", + "1"); + writes.get(30, TimeUnit.SECONDS); + final Path backup = parent.resolve("tree-delete"); + assertLogicalRequests(backup); + runConnected(receiverEnv, "restore", "--input", backup.toString()); + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, + "select * from root.logical_backup_delete.d", + "Time,root.logical_backup_delete.d.s,", + new HashSet<>(Arrays.asList("3,30,"))); + } + + @Test + public void testInspectPackAndDeprecatedAliasesKeepBackupPortable() throws Exception { + TestUtils.executeNonQueries( + senderEnv, + Arrays.asList( + "create database root.logical_backup_alias", + "create timeseries root.logical_backup_alias.d.s with datatype=INT32,encoding=PLAIN", + "insert into root.logical_backup_alias.d(time,s) values(1,10)", + "flush"), + null); + final Path parent = temporaryFolder.getRoot().toPath().toAbsolutePath(); + runConnected( + senderEnv, + "dump", + "--backup-dir", + parent.toString(), + "--backup-id", + "portable", + "--scope", + "config,schema,data", + "--model", + "tree", + "--include", + "root.logical_backup_alias.**", + "--timeout-seconds", + "180", + "--poll-interval-seconds", + "1"); + final Path backup = parent.resolve("portable"); + final byte[] manifestBeforePack = Files.readAllBytes(backup.resolve("manifest.json")); + Assert.assertEquals( + 0, + PipeLogicalBackupTool.run( + new String[] {"inspect", "--input", backup.toString(), "--json"})); + Assert.assertEquals( + 0, + PipeLogicalBackupTool.run(new String[] {"verify", "--input", backup.toString(), "--json"})); + final Path archive = parent.resolve("portable.zip"); + Assert.assertEquals( + 0, + PipeLogicalBackupTool.run( + new String[] {"pack", "--input", backup.toString(), "--output", archive.toString()})); + final Path aliasArchive = parent.resolve("portable-alias.zip"); + Assert.assertEquals( + 0, + PipeLogicalBackupTool.run( + new String[] { + "export", "--input", backup.toString(), "--output", aliasArchive.toString() + })); + Assert.assertArrayEquals( + manifestBeforePack, Files.readAllBytes(backup.resolve("manifest.json"))); + Assert.assertEquals( + 0, + PipeLogicalBackupTool.run( + new String[] {"verify", "--input", archive.toString(), "--json"})); + runConnected(receiverEnv, "import", "--input", aliasArchive.toString()); + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, + "select * from root.logical_backup_alias.d", + "Time,root.logical_backup_alias.d.s,", + new HashSet<>(Arrays.asList("1,10,"))); + } + private static void assertLogicalRequests(final Path backup) throws Exception { final Set logicalRequests = Set.of( From e82d214f844d4b1ed9b9fe7678256bf5f3c33fd2 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 23 Oct 2026 14:50:00 +0800 Subject: [PATCH 2/3] [Test] Cover filtered logical backup deletes on populated targets --- .github/workflows/logical-backup-daily-it.yml | 8 +- .../manual/IoTDBPipeLogicalBackupIT.java | 75 ++++++++++++++----- 2 files changed, 64 insertions(+), 19 deletions(-) diff --git a/.github/workflows/logical-backup-daily-it.yml b/.github/workflows/logical-backup-daily-it.yml index 21be30edbc3bb..99c0b67c63097 100644 --- a/.github/workflows/logical-backup-daily-it.yml +++ b/.github/workflows/logical-backup-daily-it.yml @@ -14,7 +14,13 @@ concurrency: jobs: logical-backup: + strategy: + fail-fast: false + max-parallel: 1 + matrix: + senderCluster: [ScalableSingleNodeMode, HighPerformanceMode] runs-on: ubuntu-latest + name: Logical Backup IT (${{ matrix.senderCluster }} source) steps: - uses: actions/checkout@v5 - name: Set up JDK 17 @@ -41,7 +47,7 @@ jobs: -DConfigNodeMaxHeapSize=256 -DDataNodeMaxHeapSize=1024 -DDataNodeMaxDirectMemorySize=768 - -DClusterConfigurations=ScalableSingleNodeMode,ScalableSingleNodeMode + -DClusterConfigurations=${{ matrix.senderCluster }},ScalableSingleNodeMode -Drat.skip=true -pl integration-test -am diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeLogicalBackupIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeLogicalBackupIT.java index 5b1d9ab0e3915..513195949dfa4 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeLogicalBackupIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeLogicalBackupIT.java @@ -78,7 +78,19 @@ public void testBoundedContinuousTableCapture() throws Exception { statement.execute("create table measurements(device STRING TAG, reading INT32 FIELD)"); statement.execute("create table other_table(device STRING TAG, reading INT32 FIELD)"); statement.execute("insert into measurements(time,device,reading) values(1,'d1',10)"); - statement.execute("insert into other_table(time,device,reading) values(1,'d1',100)"); + statement.execute("insert into measurements(time,device,reading) values(4,'d1',40)"); + statement.execute("insert into other_table(time,device,reading) values(2,'d1',200)"); + statement.execute("flush"); + } + try (final Connection connection = receiverEnv.getTableConnection(); + final Statement statement = connection.createStatement()) { + statement.execute("create database backup_db"); + statement.execute("use backup_db"); + statement.execute("create table measurements(device STRING TAG, reading INT32 FIELD)"); + statement.execute("create table other_table(device STRING TAG, reading INT32 FIELD)"); + statement.execute("insert into measurements(time,device,reading) values(0,'d1',0)"); + statement.execute("insert into measurements(time,device,reading) values(4,'d1',400)"); + statement.execute("insert into other_table(time,device,reading) values(2,'d1',200)"); statement.execute("flush"); } final String pipeName = "logical_backup_continuous_table"; @@ -103,10 +115,8 @@ public void testBoundedContinuousTableCapture() throws Exception { statement.execute( "insert into measurements(time,device,reading) values(2,'d1',20)"); statement.execute( - "insert into measurements(time,device,reading) values(4,'d1',40)"); - statement.execute( - "insert into other_table(time,device,reading) values(2,'d1',200)"); - statement.execute("delete from measurements where time <= 1"); + "insert into measurements(time,device,reading) values(3,'d1',30)"); + statement.execute("delete from measurements where time <= 4"); } } catch (final Exception e) { throw new CompletionException(e); @@ -123,13 +133,13 @@ public void testBoundedContinuousTableCapture() throws Exception { "--pipe-name", pipeName, "--scope", - "schema,data", + "data", "--model", "table", "--include", "backup_db/measurements", "--start-time", - "0", + "1", "--end-time", "3", "--mode", @@ -152,15 +162,21 @@ public void testBoundedContinuousTableCapture() throws Exception { statement.execute("use backup_db"); try (final ResultSet result = statement.executeQuery("select * from measurements")) { Assert.assertTrue(result.next()); - Assert.assertEquals(2L, result.getLong("time")); + Assert.assertEquals(0L, result.getLong("time")); Assert.assertEquals("d1", result.getString("device")); - Assert.assertEquals(20, result.getInt("reading")); + Assert.assertEquals(0, result.getInt("reading")); + Assert.assertTrue(result.next()); + Assert.assertEquals(4L, result.getLong("time")); + Assert.assertEquals("d1", result.getString("device")); + Assert.assertEquals(400, result.getInt("reading")); Assert.assertFalse(result.next()); } - try (final ResultSet result = statement.executeQuery("show tables")) { - while (result.next()) { - Assert.assertNotEquals("other_table", result.getString(1)); - } + try (final ResultSet result = statement.executeQuery("select * from other_table")) { + Assert.assertTrue(result.next()); + Assert.assertEquals(2L, result.getLong("time")); + Assert.assertEquals("d1", result.getString("device")); + Assert.assertEquals(200, result.getInt("reading")); + Assert.assertFalse(result.next()); } } } @@ -260,13 +276,27 @@ public void testDumpPackPreflightRestoreWithScopeAndTimeFilters() throws Excepti } @Test - public void testTreeContinuousCapturePreservesFilteredDeletionRange() throws Exception { + public void testTreeContinuousCaptureFiltersDeleteByPathAndTime() throws Exception { TestUtils.executeNonQueries( senderEnv, Arrays.asList( "create database root.logical_backup_delete", "create timeseries root.logical_backup_delete.d.s with datatype=INT32,encoding=PLAIN", + "create timeseries root.logical_backup_delete.d.other with datatype=INT32,encoding=PLAIN", "insert into root.logical_backup_delete.d(time,s) values(1,10)", + "insert into root.logical_backup_delete.d(time,s) values(4,40)", + "insert into root.logical_backup_delete.d(time,other) values(2,200)", + "flush"), + null); + TestUtils.executeNonQueries( + receiverEnv, + Arrays.asList( + "create database root.logical_backup_delete", + "create timeseries root.logical_backup_delete.d.s with datatype=INT32,encoding=PLAIN", + "create timeseries root.logical_backup_delete.d.other with datatype=INT32,encoding=PLAIN", + "insert into root.logical_backup_delete.d(time,s) values(0,0)", + "insert into root.logical_backup_delete.d(time,s) values(4,400)", + "insert into root.logical_backup_delete.d(time,other) values(2,200)", "flush"), null); final String pipeName = "logical_backup_tree_delete"; @@ -291,7 +321,9 @@ public void testTreeContinuousCapturePreservesFilteredDeletionRange() throws Exc "insert into root.logical_backup_delete.d(time,s) values(2,20)"); statement.execute( "insert into root.logical_backup_delete.d(time,s) values(3,30)"); - statement.execute("delete from root.logical_backup_delete.d.s where time <= 2"); + statement.execute( + "delete from root.logical_backup_delete.d.s, " + + "root.logical_backup_delete.d.other where time <= 4"); } } catch (final Exception e) { throw new CompletionException(e); @@ -312,7 +344,7 @@ public void testTreeContinuousCapturePreservesFilteredDeletionRange() throws Exc "--model", "tree", "--include", - "root.logical_backup_delete.**", + "root.logical_backup_delete.d.s", "--start-time", "1", "--end-time", @@ -331,9 +363,16 @@ public void testTreeContinuousCapturePreservesFilteredDeletionRange() throws Exc runConnected(receiverEnv, "restore", "--input", backup.toString()); TestUtils.assertDataEventuallyOnEnv( receiverEnv, - "select * from root.logical_backup_delete.d", + "select s from root.logical_backup_delete.d", "Time,root.logical_backup_delete.d.s,", - new HashSet<>(Arrays.asList("3,30,"))); + new HashSet<>(Arrays.asList("0,0,", "4,400,")), + 30); + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, + "select other from root.logical_backup_delete.d", + "Time,root.logical_backup_delete.d.other,", + new HashSet<>(Arrays.asList("2,200,")), + 30); } @Test From 5d2738bca7c0fa8e69aa00b13523ab8ea3288905 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 23 Oct 2026 21:10:33 +0800 Subject: [PATCH 3/3] [Test] Exercise logical backup delete rewrites and checkpoint recovery --- .github/workflows/logical-backup-daily-it.yml | 28 ++ .../manual/IoTDBPipeLogicalBackupIT.java | 433 ++++++++++++++++-- 2 files changed, 416 insertions(+), 45 deletions(-) diff --git a/.github/workflows/logical-backup-daily-it.yml b/.github/workflows/logical-backup-daily-it.yml index 99c0b67c63097..1882aefe58bbf 100644 --- a/.github/workflows/logical-backup-daily-it.yml +++ b/.github/workflows/logical-backup-daily-it.yml @@ -1,3 +1,20 @@ +# 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. + name: Logical Backup Daily IT on: @@ -41,6 +58,8 @@ jobs: mvn -B -ntp clean verify -Pwith-integration-tests -DskipUTs + -Dtest=NoSuchSurefireTest + -Dsurefire.failIfNoSpecifiedTests=false -Dit.test=IoTDBPipeLogicalBackupIT -Dfailsafe.failIfNoSpecifiedTests=false -DintegrationTest.forkCount=1 @@ -52,3 +71,12 @@ jobs: -pl integration-test -am -PMultiClusterIT2DualTreeManual + - name: Upload test logs + if: failure() + uses: actions/upload-artifact@v6 + with: + name: logical-backup-${{ matrix.senderCluster }} + path: | + integration-test/target/cluster-logs + integration-test/target/failsafe-reports + retention-days: 7 diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeLogicalBackupIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeLogicalBackupIT.java index 513195949dfa4..d85462f4e7793 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeLogicalBackupIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeLogicalBackupIT.java @@ -19,10 +19,30 @@ package org.apache.iotdb.pipe.it.dual.treemodel.manual; +import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; import org.apache.iotdb.commons.pipe.sink.logicalbackup.LogicalBackupArchiveReader; import org.apache.iotdb.commons.pipe.sink.logicalbackup.LogicalBackupCatalog; +import org.apache.iotdb.commons.pipe.sink.logicalbackup.LogicalBackupFormat; +import org.apache.iotdb.commons.pipe.sink.logicalbackup.LogicalBackupRecord; +import org.apache.iotdb.commons.pipe.sink.logicalbackup.LogicalBackupRecordType; +import org.apache.iotdb.commons.pipe.sink.logicalbackup.LogicalBackupSegmentReader; import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq; import org.apache.iotdb.db.it.utils.TestUtils; +import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferPlanNodeReq; +import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletInsertNodeReqV2; +import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletRawReqV2; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractDeleteDataNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; +import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; +import org.apache.iotdb.isession.SessionConfig; import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.MultiClusterIT2DualTreeManual; import org.apache.iotdb.itbase.env.BaseEnv; @@ -37,6 +57,7 @@ import org.junit.runner.RunWith; import java.io.ByteArrayInputStream; +import java.io.IOException; import java.io.InputStream; import java.nio.charset.StandardCharsets; import java.nio.file.Files; @@ -53,11 +74,16 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; @RunWith(IoTDBTestRunner.class) @Category({MultiClusterIT2DualTreeManual.class}) public class IoTDBPipeLogicalBackupIT extends AbstractPipeDualTreeModelManualIT { + private static final String DELETE_DEVICE = "root.logical_backup_delete.d"; + private static final long CANARY_TIME = 2; + private static final int CANARY_VALUE = 2000; + @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); @Override @@ -79,6 +105,8 @@ public void testBoundedContinuousTableCapture() throws Exception { statement.execute("create table other_table(device STRING TAG, reading INT32 FIELD)"); statement.execute("insert into measurements(time,device,reading) values(1,'d1',10)"); statement.execute("insert into measurements(time,device,reading) values(4,'d1',40)"); + statement.execute("insert into measurements(time,device,reading) values(2,'d2',220)"); + statement.execute("insert into measurements(time,reading) values(2,222)"); statement.execute("insert into other_table(time,device,reading) values(2,'d1',200)"); statement.execute("flush"); } @@ -90,39 +118,43 @@ public void testBoundedContinuousTableCapture() throws Exception { statement.execute("create table other_table(device STRING TAG, reading INT32 FIELD)"); statement.execute("insert into measurements(time,device,reading) values(0,'d1',0)"); statement.execute("insert into measurements(time,device,reading) values(4,'d1',400)"); + statement.execute("insert into measurements(time,device,reading) values(3,'d2',333)"); + statement.execute("insert into measurements(time,reading) values(3,334)"); statement.execute("insert into other_table(time,device,reading) values(2,'d1',200)"); statement.execute("flush"); } final String pipeName = "logical_backup_continuous_table"; + final Path parent = temporaryFolder.getRoot().toPath().toAbsolutePath(); + final Path backup = parent.resolve("bounded"); final CompletableFuture writes = CompletableFuture.runAsync( () -> { try { Awaitility.await() .atMost(30, TimeUnit.SECONDS) - .until( - () -> { - try (final Connection connection = senderEnv.getConnection(); - final Statement statement = connection.createStatement(); - final ResultSet result = - statement.executeQuery("show pipe " + pipeName)) { - return result.next(); - } - }); + .until(() -> isPipeRunning(senderEnv, pipeName)); try (final Connection connection = senderEnv.getTableConnection(); final Statement statement = connection.createStatement()) { statement.execute("use backup_db"); - statement.execute( - "insert into measurements(time,device,reading) values(2,'d1',20)"); + Awaitility.await() + .atMost(30, TimeUnit.SECONDS) + .pollInterval(1, TimeUnit.SECONDS) + .until( + () -> { + statement.execute( + "insert into measurements(time,device,reading) values(2,'d1',2000)"); + return hasCommittedCanary(backup, "measurements", "reading"); + }); statement.execute( "insert into measurements(time,device,reading) values(3,'d1',30)"); - statement.execute("delete from measurements where time <= 4"); + statement.execute("delete from measurements where device = 'd1' and time <= 4"); + statement.execute( + "insert into measurements(time,device,reading) values(3,'d1',300)"); } } catch (final Exception e) { throw new CompletionException(e); } }); - final Path parent = temporaryFolder.getRoot().toPath().toAbsolutePath(); runConnected( senderEnv, "dump", @@ -145,31 +177,57 @@ public void testBoundedContinuousTableCapture() throws Exception { "--mode", "continuous", "--until", - Instant.now().plusSeconds(20).toString(), + Instant.now().plusSeconds(45).toString(), "--timeout-seconds", "180", "--poll-interval-seconds", "1"); writes.get(30, TimeUnit.SECONDS); - final Path backup = parent.resolve("bounded"); final LogicalBackupCatalog catalog = LogicalBackupCatalog.load(backup.resolve("manifest.json")); Assert.assertEquals("SEALED", catalog.status); Assert.assertEquals("continuous", catalog.mode); + try (final Connection connection = senderEnv.getTableConnection(); + final Statement statement = connection.createStatement()) { + statement.execute("use backup_db"); + try (final ResultSet result = + statement.executeQuery( + "select time,reading from measurements where device = 'd1' and time = 3")) { + Assert.assertTrue("The source must retain the rewrite after deletion", result.next()); + Assert.assertEquals(3, result.getLong("time")); + Assert.assertEquals(300, result.getInt("reading")); + Assert.assertFalse(result.next()); + } + } assertLogicalRequests(backup); + assertTableDeleteIsFiltered(backup); + final List replayTrace = capturedRewriteAndDeletes(backup); runConnected(receiverEnv, "restore", "--input", backup.toString()); try (final Connection connection = receiverEnv.getTableConnection(); final Statement statement = connection.createStatement()) { statement.execute("use backup_db"); try (final ResultSet result = statement.executeQuery("select * from measurements")) { - Assert.assertTrue(result.next()); - Assert.assertEquals(0L, result.getLong("time")); - Assert.assertEquals("d1", result.getString("device")); - Assert.assertEquals(0, result.getInt("reading")); - Assert.assertTrue(result.next()); - Assert.assertEquals(4L, result.getLong("time")); - Assert.assertEquals("d1", result.getString("device")); - Assert.assertEquals(400, result.getInt("reading")); - Assert.assertFalse(result.next()); + final Set rows = new HashSet<>(); + while (result.next()) { + Assert.assertTrue( + rows.add( + result.getLong("time") + + "," + + result.getString("device") + + "," + + result.getInt("reading"))); + } + Assert.assertEquals( + "Archived rewrite and deletion order: " + replayTrace, + new HashSet<>( + Arrays.asList( + "0,d1,0", + "2,d2,220", + "2,null,222", + "3,d1,300", + "3,d2,333", + "3,null,334", + "4,d1,400")), + rows); } try (final ResultSet result = statement.executeQuery("select * from other_table")) { Assert.assertTrue(result.next()); @@ -273,6 +331,43 @@ public void testDumpPackPreflightRestoreWithScopeAndTimeFilters() throws Excepti } } } + final Path conflictCheckpoint = parent.resolve("restore-conflict-checkpoint.json"); + runConnectedExpectingFailureWithMessage( + receiverEnv, + conflictCheckpoint, + "already been created as database", + "restore", + "--input", + zip.toString(), + "--checkpoint", + conflictCheckpoint.toString()); + try (final Connection connection = receiverEnv.getConnection(); + final Statement statement = connection.createStatement()) { + statement.execute("drop database root.backup"); + } + runConnected( + receiverEnv, + "restore", + "--input", + zip.toString(), + "--checkpoint", + conflictCheckpoint.toString(), + "--resume"); + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, + "select * from root.backup.**", + "Time,root.backup.d.s1,root.backup.d.s2,", + new HashSet<>(Arrays.asList("2,null,true,", "3,30,false,"))); + runConnectedExpectingFailureWithMessage( + senderEnv, + conflictCheckpoint, + "checkpoint does not match the source or target", + "restore", + "--input", + zip.toString(), + "--checkpoint", + conflictCheckpoint.toString(), + "--resume"); } @Test @@ -288,37 +383,42 @@ public void testTreeContinuousCaptureFiltersDeleteByPathAndTime() throws Excepti "insert into root.logical_backup_delete.d(time,other) values(2,200)", "flush"), null); - TestUtils.executeNonQueries( - receiverEnv, - Arrays.asList( - "create database root.logical_backup_delete", - "create timeseries root.logical_backup_delete.d.s with datatype=INT32,encoding=PLAIN", - "create timeseries root.logical_backup_delete.d.other with datatype=INT32,encoding=PLAIN", - "insert into root.logical_backup_delete.d(time,s) values(0,0)", - "insert into root.logical_backup_delete.d(time,s) values(4,400)", - "insert into root.logical_backup_delete.d(time,other) values(2,200)", - "flush"), - null); + try (final Connection connection = receiverEnv.getConnection(); + final Statement statement = connection.createStatement()) { + statement.execute("create database root.logical_backup_delete"); + statement.execute( + "create timeseries root.logical_backup_delete.d.s with datatype=INT32,encoding=PLAIN"); + statement.execute( + "create timeseries root.logical_backup_delete.d.other with datatype=INT32,encoding=PLAIN"); + statement.execute("insert into root.logical_backup_delete.d(time,s) values(0,0)"); + statement.execute("insert into root.logical_backup_delete.d(time,s) values(4,400)"); + statement.execute("insert into root.logical_backup_delete.d(time,other) values(2,200)"); + statement.execute("flush"); + } final String pipeName = "logical_backup_tree_delete"; + final Path parent = temporaryFolder.getRoot().toPath().toAbsolutePath(); + final Path backup = parent.resolve("tree-delete"); final CompletableFuture writes = CompletableFuture.runAsync( () -> { try { Awaitility.await() .atMost(30, TimeUnit.SECONDS) + .until(() -> isPipeRunning(senderEnv, pipeName)); + Awaitility.await() + .atMost(30, TimeUnit.SECONDS) + .pollInterval(1, TimeUnit.SECONDS) .until( () -> { try (final Connection connection = senderEnv.getConnection(); - final Statement statement = connection.createStatement(); - final ResultSet result = - statement.executeQuery("show pipe " + pipeName)) { - return result.next(); + final Statement statement = connection.createStatement()) { + statement.execute( + "insert into root.logical_backup_delete.d(time,s) values(2,2000)"); } + return hasCommittedCanary(backup, DELETE_DEVICE, "s"); }); try (final Connection connection = senderEnv.getConnection(); final Statement statement = connection.createStatement()) { - statement.execute( - "insert into root.logical_backup_delete.d(time,s) values(2,20)"); statement.execute( "insert into root.logical_backup_delete.d(time,s) values(3,30)"); statement.execute( @@ -329,7 +429,6 @@ public void testTreeContinuousCaptureFiltersDeleteByPathAndTime() throws Excepti throw new CompletionException(e); } }); - final Path parent = temporaryFolder.getRoot().toPath().toAbsolutePath(); runConnected( senderEnv, "dump", @@ -340,7 +439,7 @@ public void testTreeContinuousCaptureFiltersDeleteByPathAndTime() throws Excepti "--pipe-name", pipeName, "--scope", - "schema,data", + "data", "--model", "tree", "--include", @@ -352,14 +451,14 @@ public void testTreeContinuousCaptureFiltersDeleteByPathAndTime() throws Excepti "--mode", "continuous", "--until", - Instant.now().plusSeconds(20).toString(), + Instant.now().plusSeconds(45).toString(), "--timeout-seconds", "180", "--poll-interval-seconds", "1"); writes.get(30, TimeUnit.SECONDS); - final Path backup = parent.resolve("tree-delete"); assertLogicalRequests(backup); + assertTreeDeleteIsFiltered(backup); runConnected(receiverEnv, "restore", "--input", backup.toString()); TestUtils.assertDataEventuallyOnEnv( receiverEnv, @@ -453,6 +552,238 @@ private static void assertLogicalRequests(final Path backup) throws Exception { record -> Assert.assertTrue(logicalRequests.contains(record.getRequestType()))); } + private static List readDeletes(final Path backup) throws Exception { + final List deletes = new ArrayList<>(); + for (final LogicalBackupArchiveReader.BackupStream stream : + new LogicalBackupArchiveReader().read(backup, false)) { + if (!"data".equals(stream.getManifest().streamType)) { + continue; + } + for (final LogicalBackupArchiveReader.EventGroup group : stream.getEventGroups()) { + for (final LogicalBackupRecord request : group.getRequests()) { + if (request.getRequestType() == PipeRequestType.TRANSFER_PLAN_NODE.getType()) { + final PlanNode planNode = + PipeTransferPlanNodeReq.fromTPipeTransferReq(request.toTPipeTransferReq()) + .getPlanNode(); + if (planNode instanceof AbstractDeleteDataNode) { + deletes.add((AbstractDeleteDataNode) planNode); + } + } + } + } + } + Assert.assertFalse(deletes.isEmpty()); + return deletes; + } + + private static void assertTreeDeleteIsFiltered(final Path backup) throws Exception { + for (final AbstractDeleteDataNode node : readDeletes(backup)) { + Assert.assertTrue(node instanceof DeleteDataNode); + final DeleteDataNode delete = (DeleteDataNode) node; + Assert.assertEquals(1, delete.getPathList().size()); + Assert.assertEquals( + "root.logical_backup_delete.d.s", delete.getPathList().get(0).getFullPath()); + Assert.assertEquals(1, delete.getDeleteStartTime()); + Assert.assertEquals(3, delete.getDeleteEndTime()); + } + } + + private static void assertTableDeleteIsFiltered(final Path backup) throws Exception { + for (final AbstractDeleteDataNode node : readDeletes(backup)) { + Assert.assertTrue(node instanceof RelationalDeleteDataNode); + final RelationalDeleteDataNode delete = (RelationalDeleteDataNode) node; + Assert.assertEquals("backup_db", delete.getDatabaseName()); + Assert.assertFalse(delete.getModEntries().isEmpty()); + for (final TableDeletionEntry entry : delete.getModEntries()) { + Assert.assertEquals("measurements", entry.getTableName()); + Assert.assertEquals(1, entry.getStartTime()); + Assert.assertEquals(3, entry.getEndTime()); + } + } + } + + private static boolean hasCommittedCanary( + final Path backup, final String device, final String measurement) throws Exception { + if (!Files.isDirectory(backup)) { + return false; + } + final LogicalBackupSegmentReader reader = + new LogicalBackupSegmentReader(LogicalBackupFormat.MAX_RECORD_BYTES); + try (final Stream paths = Files.walk(backup)) { + for (final Path segment : + paths + .filter(path -> path.getFileName().toString().endsWith(".pwal")) + .filter( + path -> + path.getParent() != null + && path.getParent().getFileName().toString().startsWith("data-")) + .toList()) { + final LogicalBackupSegmentReader.ScanResult scan; + try { + scan = reader.scan(segment, true); + } catch (final IOException e) { + // The active segment may be changing while this poll reads it. + continue; + } + final List requests = new ArrayList<>(); + boolean openEventGroup = false; + for (final LogicalBackupRecord record : scan.getRecords()) { + if (record.getRecordType() == LogicalBackupRecordType.EVENT_BEGIN) { + requests.clear(); + openEventGroup = true; + } else if (record.getRecordType() == LogicalBackupRecordType.PIPE_REQUEST + && openEventGroup) { + requests.add(record); + } else if (record.getRecordType() == LogicalBackupRecordType.EVENT_COMMIT + && openEventGroup) { + if (requests.stream() + .anyMatch(request -> isCanaryRequest(request, device, measurement))) { + return true; + } + requests.clear(); + openEventGroup = false; + } + } + } + } + return false; + } + + private static boolean isCanaryRequest( + final LogicalBackupRecord record, final String device, final String measurement) { + return containsPoint(insertStatement(record), device, measurement, CANARY_TIME, CANARY_VALUE); + } + + private static InsertBaseStatement insertStatement(final LogicalBackupRecord record) { + if (record.getRequestType() == PipeRequestType.TRANSFER_TABLET_RAW_V2.getType()) { + return PipeTransferTabletRawReqV2.fromTPipeTransferReq(record.toTPipeTransferReq()) + .constructStatement(); + } + if (record.getRequestType() == PipeRequestType.TRANSFER_TABLET_INSERT_NODE_V2.getType()) { + return PipeTransferTabletInsertNodeReqV2.fromTPipeTransferReq(record.toTPipeTransferReq()) + .constructStatement(); + } + return null; + } + + private static List capturedRewriteAndDeletes(final Path backup) throws Exception { + final List trace = new ArrayList<>(); + boolean capturedRewrite = false; + for (final LogicalBackupArchiveReader.BackupStream stream : + new LogicalBackupArchiveReader().read(backup, false)) { + if (!"data".equals(stream.getManifest().streamType)) { + continue; + } + for (final LogicalBackupArchiveReader.EventGroup group : stream.getEventGroups()) { + for (final LogicalBackupRecord request : group.getRequests()) { + final String position = stream.getManifest().streamId + "#" + request.getSequence(); + if (containsPoint(insertStatement(request), "measurements", "reading", 3, 300)) { + capturedRewrite = true; + trace.add(position + " INSERT d1@3=300"); + } else if (request.getRequestType() == PipeRequestType.TRANSFER_PLAN_NODE.getType()) { + trace.add( + position + + " " + + PipeTransferPlanNodeReq.fromTPipeTransferReq(request.toTPipeTransferReq()) + .getPlanNode()); + } + } + } + } + Assert.assertTrue("The archive must contain the rewrite: " + trace, capturedRewrite); + return trace; + } + + private static boolean containsPoint( + final InsertBaseStatement statement, + final String device, + final String measurement, + final long expectedTime, + final int expectedValue) { + if (statement instanceof InsertTabletStatement) { + final InsertTabletStatement tablet = (InsertTabletStatement) statement; + return tablet.getDevicePath() != null + && device.equals(tablet.getDevicePath().getFullPath()) + && containsPoint( + tablet.getTimes(), + tablet.getMeasurements(), + tablet.getColumns(), + measurement, + expectedTime, + expectedValue); + } + if (statement instanceof InsertRowStatement) { + final InsertRowStatement row = (InsertRowStatement) statement; + return row.getDevicePath() != null + && device.equals(row.getDevicePath().getFullPath()) + && containsPoint( + new long[] {row.getTime()}, + row.getMeasurements(), + row.getValues(), + measurement, + expectedTime, + expectedValue); + } + if (statement instanceof InsertRowsStatement) { + final List rows = + ((InsertRowsStatement) statement).getInsertRowStatementList(); + return rows != null + && rows.stream() + .anyMatch( + row -> containsPoint(row, device, measurement, expectedTime, expectedValue)); + } + return false; + } + + private static boolean containsPoint( + final long[] times, + final String[] measurements, + final Object[] values, + final String measurement, + final long expectedTime, + final int expectedValue) { + if (times == null || measurements == null || values == null) { + return false; + } + for (int column = 0; column < measurements.length; column++) { + if (!measurement.equals(measurements[column])) { + continue; + } + if (column >= values.length || values[column] == null) { + continue; + } + final Object valuesColumn = values[column]; + if (valuesColumn instanceof int[]) { + final int[] intValues = (int[]) valuesColumn; + for (int row = 0; row < Math.min(times.length, intValues.length); row++) { + if (times[row] == expectedTime && intValues[row] == expectedValue) { + return true; + } + } + } else if (valuesColumn instanceof Integer) { + for (final long time : times) { + if (time == expectedTime && ((Integer) valuesColumn) == expectedValue) { + return true; + } + } + } + } + return false; + } + + private static boolean isPipeRunning(final BaseEnv env, final String pipeName) throws Exception { + try (final SyncConfigNodeIServiceClient client = + (SyncConfigNodeIServiceClient) env.getLeaderConfigNodeConnection()) { + return client + .showPipe(new TShowPipeReq().setUserName(SessionConfig.DEFAULT_USER)) + .pipeInfoList + .stream() + .filter(pipe -> pipeName.equals(pipe.id)) + .map(TShowPipeInfo::getState) + .anyMatch("RUNNING"::equals); + } + } + private static void runConnected(final BaseEnv env, final String command, final String... options) throws Exception { final List args = @@ -475,4 +806,16 @@ private static void runConnected(final BaseEnv env, final String command, final System.setIn(previous); } } + + private static void runConnectedExpectingFailureWithMessage( + final BaseEnv env, + final Path checkpoint, + final String expectedMessage, + final String command, + final String... options) { + final IOException failure = + Assert.assertThrows(IOException.class, () -> runConnected(env, command, options)); + Assert.assertTrue(failure.getMessage(), failure.getMessage().contains(expectedMessage)); + Assert.assertTrue(Files.exists(checkpoint)); + } }