Squashed commit of the following:

commit fcae2b4
Author: Steve Yurong Su <rong@apache.org>
Date:   Mon Jan 29 17:21:34 2024 +0800

    Pipe: support reporting progress by UserDefinedEvent (apache#12000)

commit e2f7199
Author: Zhijia Cao <caozhijia@126.com>
Date:   Mon Jan 29 15:33:18 2024 +0800

    Storage Engine: optimized DataRegion recover log (apache#11996)

commit 775f169
Author: Xuan Ronaldo <xuanronaldo@qq.com>
Date:   Mon Jan 29 14:47:51 2024 +0800

    Pipe: result of show pipes statement is incorrect when it contains back quoted pipe names (apache#11944)

commit e331ead
Author: Haonan <hhaonan@outlook.com>
Date:   Mon Jan 29 12:33:39 2024 +0800

    Fix non-latest partition last flush time cannot recover (apache#11999)

commit ea37a1f
Author: shuwenwei <55970239+shuwenwei@users.noreply.github.com>
Date:   Mon Jan 29 00:08:27 2024 +0800

    fix repair non aligned series (apache#11997)

commit 5619156
Author: Caideyipi <87789683+Caideyipi@users.noreply.github.com>
Date:   Sun Jan 28 17:58:08 2024 +0800

    Pipe: Report load tsfile points to flush point metric timeseries in root.__system  (apache#11976)

commit 7923eb9
Author: Beyyes <cgf1993@foxmail.com>
Date:   Sun Jan 28 14:43:27 2024 +0800

    Fix concurrent bug in QueryContext

commit 9831082
Author: Jackie Tien <jackietien97@gmail.com>
Date:   Sun Jan 28 10:26:04 2024 +0800

    Session insert request won't fail while rolling upgrade

commit e7b0ca5
Author: Haonan <hhaonan@outlook.com>
Date:   Sun Jan 28 09:25:29 2024 +0800

    Refactor last flush time management (apache#11946)

commit 769a151
Author: ZhangHongYin <46039728+SpriCoder@users.noreply.github.com>
Date:   Sun Jan 28 02:31:13 2024 +0800

    Revert "Fixed issue where ConfigNode might recover incorrectly under SimpleConsensus" (apache#11987)

commit a544517
Author: Chen YZ <43774645+Cpaulyz@users.noreply.github.com>
Date:   Sat Jan 27 14:09:21 2024 +0800

    Fix template device aligned properties recover unexpectly with SimpleConsensus (apache#11968)

    Fix template device aligned properties recover unexpectly with SimpleConsensus

commit 7d0a735
Author: Colin Li <105656576+shuolinl@users.noreply.github.com>
Date:   Fri Jan 26 20:55:08 2024 +0800

    fix.create role folder when init.

commit 6c15634
Author: shuwenwei <55970239+shuwenwei@users.noreply.github.com>
Date:   Fri Jan 26 19:00:37 2024 +0800

    Fix estimating compaction memory without acquire read lock (apache#11972)

commit 40fc15b
Author: Caideyipi <87789683+Caideyipi@users.noreply.github.com>
Date:   Fri Jan 26 18:30:25 2024 +0800

    Pipe: fix connector subtasks can not be stopped and restarted after exception reporting by connector subtasks (apache#11979)

commit ebbddf3
Author: Jackie Tien <jackietien97@gmail.com>
Date:   Fri Jan 26 17:07:20 2024 +0800

    Multiply 3 for ScanOperator to take unsequence data into account (apache#11978)

commit 75ce771
Author: Caideyipi <87789683+Caideyipi@users.noreply.github.com>
Date:   Fri Jan 26 16:49:34 2024 +0800

    Pipe: fixed the bug that region follower tsfiles can not record the IoTProgressIndex from region leader (apache#11980)

commit 59fe3fe
Author: Beyyes <cgf1993@foxmail.com>
Date:   Thu Jan 25 16:19:38 2024 +0800

    Add unit byte for no enough query memory

commit 82d0459
Author: Xiangpeng Hu <65238551+BUAAserein@users.noreply.github.com>
Date:   Thu Jan 25 09:52:00 2024 +0800

    Fixed issue where ConfigNode might recover incorrectly under SimpleConsensus (apache#11969)

commit 7200e21
Author: shuwenwei <55970239+shuwenwei@users.noreply.github.com>
Date:   Wed Jan 24 23:09:06 2024 +0800

    Repair unsorted file by compaction (apache#11887)

commit 092fcb5
Author: Alan Choo <43991780+HeimingZ@users.noreply.github.com>
Date:   Wed Jan 24 19:39:17 2024 +0800

    fix npe (apache#11966)

commit 7f7285c
Author: Itami Sho <42286868+MiniSho@users.noreply.github.com>
Date:   Wed Jan 24 11:02:41 2024 +0800

    Load: Correctly release memory when system is in READ_ONLY mode (apache#11961)

commit 39b5893
Author: shuwenwei <55970239+shuwenwei@users.noreply.github.com>
Date:   Wed Jan 24 10:46:13 2024 +0800

    Fix read chunk compaction performer null pointer exception (apache#11960)

commit c52da2b
Author: Yongzao <yongzao@apache.org>
Date:   Wed Jan 24 09:47:35 2024 +0800

    [IOTDB-6295] Adjust the timeout period in ConfigNode registration procedure (apache#11962)

commit 2072603
Author: Colin Li <105656576+shuolinl@users.noreply.github.com>
Date:   Wed Jan 24 09:25:26 2024 +0800

    Fix deadlock when delete timeseries after loading data.

commit 5c6c972
Author: Potato <tanxinyu@apache.org>
Date:   Tue Jan 23 21:50:58 2024 +0800

    Enhance DataNode startup probing logic (apache#11957)

    Signed-off-by: OneSizeFitQuorum <tanxinyu@apache.org>

commit b07fafe
Author: Alan Choo <43991780+HeimingZ@users.noreply.github.com>
Date:   Tue Jan 23 20:04:13 2024 +0800

    Optimized wal file deletion algorithm (apache#11948)

    Co-authored-by: Zhijia Cao <caozhijia@126.com>

commit 1416217
Author: shuwenwei <55970239+shuwenwei@users.noreply.github.com>
Date:   Tue Jan 23 17:53:16 2024 +0800

    skip insertion compaction disk space check (apache#11959)

commit 15f6330
Author: Steve Yurong Su <rong@apache.org>
Date:   Tue Jan 23 17:27:29 2024 +0800

    Pipe: support restarting pipes on datanodes when they are stuck (apache#11955)

    Co-authored-by: Caideyipi <87789683+Caideyipi@users.noreply.github.com>

commit 827735d
Author: CloudWise-Lukemiao <76942485+CloudWise-Lukemiao@users.noreply.github.com>
Date:   Tue Jan 23 16:35:09 2024 +0800

    fixed collection iotdb info unable to obtain information (apache#11938)

commit 070de20
Author: ZhaoXin <x_zhao16@163.com>
Date:   Tue Jan 23 15:57:08 2024 +0800

    PBTree: Decouple PageManager (apache#11953)

commit 3846bb7
Author: Potato <tanxinyu@apache.org>
Date:   Tue Jan 23 15:54:57 2024 +0800

    Enhance the robustness of Ratis linearizable reads for node offline scenarios (apache#11954)

    Signed-off-by: OneSizeFitQuorum <tanxinyu@apache.org>

commit 4cad6af
Author: shuwenwei <55970239+shuwenwei@users.noreply.github.com>
Date:   Tue Jan 23 10:45:16 2024 +0800

    Fix compile error caused by concurrent pr (apache#11951)

commit 6f37cf8
Author: Yongzao <yongzao@apache.org>
Date:   Tue Jan 23 10:28:20 2024 +0800

    [IOTDB-6293] Use nanotime in cluster heartbeat (apache#11916)

commit 69202a9
Author: shuwenwei <55970239+shuwenwei@users.noreply.github.com>
Date:   Tue Jan 23 09:41:28 2024 +0800

    reduce stack message of last time check exception for read chunk compaction performer (apache#11927)

commit 9e05b26
Author: Peng Junzhi <78788603+Pengzna@users.noreply.github.com>
Date:   Mon Jan 22 04:01:18 2024 -0600

    Add the ability for ClientManager to periodically clean up idle objects

commit 8885df3
Author: shuwenwei <55970239+shuwenwei@users.noreply.github.com>
Date:   Mon Jan 22 15:08:27 2024 +0800

    fix PointPriorityReader calculation of aligned series null value num (apache#11945)

commit 85e71a4
Author: shuwenwei <55970239+shuwenwei@users.noreply.github.com>
Date:   Mon Jan 22 12:22:34 2024 +0800

    Fix update time index after compact aligned series (apache#11943)

commit f1db292
Author: Steve Yurong Su <rong@apache.org>
Date:   Mon Jan 22 12:13:27 2024 +0800

    Pipe: decrease the default value of pipe_sink_selector_number to 4 for better performance (apache#11942)

commit 4ec7956
Author: William Song <48054931+SzyWilliam@users.noreply.github.com>
Date:   Sun Jan 21 23:03:57 2024 +0800

    fix remaining issues when upgrading to ratis 3.0.1 (apache#11940)

commit db7d115
Author: Alan Choo <43991780+HeimingZ@users.noreply.github.com>
Date:   Sun Jan 21 16:23:12 2024 +0800

    Skip broken wal files when iterate the PlanNodeIterator (apache#11935)

commit 3464e95
Author: Itami Sho <42286868+MiniSho@users.noreply.github.com>
Date:   Sun Jan 21 15:20:12 2024 +0800

    Load: Support configure load_clean_up_task_execution_delay_time_seconds of a load task (apache#11936)

    Co-authored-by: Steve Yurong Su <rong@apache.org>

commit f94f99a
Author: Zhihao Shen <zhihaoshen7@qq.com>
Date:   Fri Jan 19 22:32:13 2024 +0800

    Change partial aggregation symbols for variance and standard deviation aggregators

commit 0cb1499
Author: Potato <tanxinyu@apache.org>
Date:   Fri Jan 19 18:57:45 2024 +0800

    Enable SchemaRegion Leader Auto Balance (apache#11926)

    Signed-off-by: OneSizeFitQuorum <tanxinyu@apache.org>

commit c3597b6
Author: Potato <tanxinyu@apache.org>
Date:   Fri Jan 19 17:17:08 2024 +0800

    Fixed the issue that IoTConsensus synchronization is stuck when WAL file is corrupted (apache#11930)

    Signed-off-by: OneSizeFitQuorum <tanxinyu@apache.org>

commit 63605be
Author: Potato <tanxinyu@apache.org>
Date:   Fri Jan 19 15:07:13 2024 +0800

    Fix the issue where ConfigNode reports 'ConsensusGroupAlreadyExist' error during startup in some scenarios (apache#11928)

    Signed-off-by: OneSizeFitQuorum <tanxinyu@apache.org>

commit dd48a57
Author: YuFengLiu <38746920+l2280212@users.noreply.github.com>
Date:   Fri Jan 19 15:06:17 2024 +0800

    Improve  some descriptions of daemon-script

commit 771c096
Author: Steve Yurong Su <rong@apache.org>
Date:   Fri Jan 19 11:21:19 2024 +0800

    Pipe: let non enriched event forever retry when exception occurred to avoid subtask fake running status recorded at data node task agent (apache#11929)

commit 4cc5b2b
Author: Beyyes <cgf1993@foxmail.com>
Date:   Fri Jan 19 10:02:17 2024 +0800

    Fix query result is empty in order by time align by device

commit 7dd570d
Author: Steve Yurong Su <rong@apache.org>
Date:   Thu Jan 18 20:43:16 2024 +0800

    Load: Introduce LoadTsFileManager.CleanupTask to force close writer manager after exception occurring (apache#11924)

commit f40c373
Author: Haonan <hhaonan@outlook.com>
Date:   Thu Jan 18 17:42:42 2024 +0800

    fix ActiveMemTableCount update error (apache#11923)

commit 7e51623
Author: CritasWang <critas@outlook.com>
Date:   Thu Jan 18 15:11:03 2024 +0800

    conf:  1. open replication_factor,2. open dn_metric_prometheus_reporter_port (apache#11922)

commit eae4d01
Author: Zhijia Cao <caozhijia@126.com>
Date:   Thu Jan 18 14:19:50 2024 +0800

    remove enable_insertion_cross_space_compaction from iotdb-common.properties (apache#11925)

commit 23928e9
Author: CritasWang <critas@outlook.com>
Date:   Thu Jan 18 12:17:44 2024 +0800

    modify ExitOnOutOfMemoryError to CrashOnOutOfMemoryError on .bat

commit a965709
Author: William Song <48054931+SzyWilliam@users.noreply.github.com>
Date:   Thu Jan 18 11:10:10 2024 +0800

    fix remaining issues when upgrading to ratis 3.0.1 (apache#11921)

commit da79077
Author: shuwenwei <55970239+shuwenwei@users.noreply.github.com>
Date:   Wed Jan 17 19:16:21 2024 +0800

    Read chunk aligned series compaction executor (apache#11745)

commit 22f915f
Author: shuwenwei <55970239+shuwenwei@users.noreply.github.com>
Date:   Wed Jan 17 19:10:06 2024 +0800

    Strictly check inner space compaction file size (apache#11919)

commit e9e01f3
Author: Steve Yurong Su <rong@apache.org>
Date:   Wed Jan 17 18:41:45 2024 +0800

    Pipe: make PipeHistoricalDataRegionExtractor & PipeRealtimeDataRegionExtractor log info much clear with pipe name and region id added (apache#11920)

commit 9aceb9f
Author: William Song <48054931+SzyWilliam@users.noreply.github.com>
Date:   Wed Jan 17 18:02:03 2024 +0800

    [RatisConsensus] Bump ratis version from 3.0.0 to 3.0.1 (apache#11899)

commit 6d46ceb
Author: Yongzao <yongzao@apache.org>
Date:   Wed Jan 17 17:34:48 2024 +0800

    [IOTDB-6291] Update RegionGroup configuration when update DataNode configuration (apache#11914)

commit c147683
Author: shuwenwei <55970239+shuwenwei@users.noreply.github.com>
Date:   Wed Jan 17 17:15:15 2024 +0800

    fix insertion compaction target file progress index (apache#11917)

commit 96929fa
Author: Yongzao <yongzao@apache.org>
Date:   Wed Jan 17 16:02:03 2024 +0800

    [IOTDB-6292] Add the internal address column in the result of show regions (apache#11915)

commit 3f16839
Author: Jackie Tien <jackietien97@gmail.com>
Date:   Wed Jan 17 14:55:14 2024 +0800

    Dicard all null row of aligned page in AlignedPageReader (apache#11913)

commit bc3c9f4
Author: Zhijia Cao <caozhijia@126.com>
Date:   Wed Jan 17 14:33:45 2024 +0800

    remove insert cross compaction config v1 (apache#11912)

commit 85fe36d
Author: Summer <43237967+2b3c511@users.noreply.github.com>
Date:   Wed Jan 17 14:33:30 2024 +0800

    This configuration file needs to be configured only when the start-all.sh,stop-all.sh, and destroy.sh scripts are required. You also need to modify this configuration file when the cluster nodes change (apache#11911)

    Co-authored-by: 2b3c511 <rong.li@timecho.com>

commit 8fefc91
Author: YuFengLiu <38746920+l2280212@users.noreply.github.com>
Date:   Wed Jan 17 14:33:00 2024 +0800

    Daemon script change default choice from n to y (apache#11910)

integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterRestartIT.java

@@ -19,11 +19,19 @@
 
 package org.apache.iotdb.confignode.it.cluster;
 
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
 import org.apache.iotdb.commons.client.exception.ClientManagerException;
 import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
 import org.apache.iotdb.commons.cluster.NodeStatus;
 import org.apache.iotdb.confignode.it.utils.ConfigNodeTestUtils;
+import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionTableResp;
+import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
 import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp;
+import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq;
+import org.apache.iotdb.confignode.rpc.thrift.TShowRegionResp;
+import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
 import org.apache.iotdb.it.env.EnvFactory;
 import org.apache.iotdb.it.env.cluster.EnvUtils;
 import org.apache.iotdb.it.env.cluster.config.MppBaseConfig;
@@ -33,9 +41,11 @@
 import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
 import org.apache.iotdb.it.framework.IoTDBTestRunner;
 import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.apache.thrift.TException;
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
@@ -46,7 +56,9 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.apache.iotdb.consensus.ConsensusFactory.RATIS_CONSENSUS;
 
@@ -59,6 +71,8 @@ public class IoTDBClusterRestartIT {
 
   private static final int testConfigNodeNum = 3, testDataNodeNum = 2;
 
+  private static final long testTimePartitionInterval = 604800000;
+
   @Before
   public void setUp() {
     EnvFactory.getEnv()
@@ -114,6 +128,45 @@ public void clusterRestartTest() throws InterruptedException {
   @Test
   public void clusterRestartAfterUpdateDataNodeTest()
       throws InterruptedException, ClientManagerException, IOException, TException {
+    // Create default Database
+    try (SyncConfigNodeIServiceClient client =
+        (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+      TSStatus status = client.setDatabase(new TDatabaseSchema("root.database"));
+      Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+    }
+    // Create some DataPartitions to extend 2 DataRegionGroups
+    Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> partitionSlotsMap =
+        ConfigNodeTestUtils.constructPartitionSlotsMap(
+            "root.database", 0, 10, 0, 10, testTimePartitionInterval);
+    TDataPartitionReq dataPartitionReq = new TDataPartitionReq(partitionSlotsMap);
+    TDataPartitionTableResp dataPartitionTableResp = null;
+    for (int retry = 0; retry < 5; retry++) {
+      // Build new Client since it's unstable in Win8 environment
+      try (SyncConfigNodeIServiceClient configNodeClient =
+          (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+        dataPartitionTableResp = configNodeClient.getOrCreateDataPartitionTable(dataPartitionReq);
+        if (dataPartitionTableResp != null) {
+          break;
+        }
+      } catch (Exception e) {
+        // Retry sometimes in order to avoid request timeout
+        logger.error(e.getMessage());
+        TimeUnit.SECONDS.sleep(1);
+      }
+    }
+    Assert.assertNotNull(dataPartitionTableResp);
+    Assert.assertEquals(
+        TSStatusCode.SUCCESS_STATUS.getStatusCode(), dataPartitionTableResp.getStatus().getCode());
+    Assert.assertNotNull(dataPartitionTableResp.getDataPartitionTable());
+    ConfigNodeTestUtils.checkDataPartitionTable(
+        "root.database",
+        0,
+        10,
+        0,
+        10,
+        testTimePartitionInterval,
+        dataPartitionTableResp.getDataPartitionTable());
+
     // Shutdown all DataNodes
     for (int i = 0; i < testDataNodeNum; i++) {
       EnvFactory.getEnv().shutdownDataNode(i);
@@ -151,12 +204,29 @@ public void clusterRestartAfterUpdateDataNodeTest()
     // Check DataNode EndPoint
     try (SyncConfigNodeIServiceClient client =
         (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+      // Check update in NodeInfo
       TShowClusterResp showClusterResp = client.showCluster();
       ConfigNodeTestUtils.checkNodeConfig(
           showClusterResp.getConfigNodeList(),
           showClusterResp.getDataNodeList(),
           EnvFactory.getEnv().getConfigNodeWrapperList(),
           dataNodeWrapperList);
+
+      // Check update in PartitionInfo
+      TShowRegionResp showRegionResp = client.showRegion(new TShowRegionReq());
+      showRegionResp
+          .getRegionInfoList()
+          .forEach(
+              regionInfo -> {
+                AtomicBoolean matched = new AtomicBoolean(false);
+                dataNodeWrapperList.forEach(
+                    dataNodeWrapper -> {
+                      if (regionInfo.getClientRpcPort() == dataNodeWrapper.getPort()) {
+                        matched.set(true);
+                      }
+                    });
+                Assert.assertTrue(matched.get());
+              });
     }
   }
 

integration-test/src/test/java/org/apache/iotdb/db/it/IOTDBLoadTsFileIT.java

@@ -220,6 +220,15 @@ public void testLoad() throws Exception {
         }
       }
     }
+
+    // try delete after loading. Expect no deadlock
+    try (Connection connection = EnvFactory.getEnv().getConnection();
+        Statement statement = connection.createStatement()) {
+      statement.execute(
+          String.format(
+              "delete timeseries %s.%s",
+              SchemaConfig.DEVICE_0, SchemaConfig.MEASUREMENT_00.getMeasurementId()));
+    }
   }
 
   @Test

integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRepairDataIT.java

@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.it;
+
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.chunk.ChunkWriterImpl;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.iotdb.db.utils.constant.TestConstant.TIMESTAMP_STR;
+import static org.junit.Assert.assertNotNull;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({LocalStandaloneIT.class, ClusterIT.class})
+public class IoTDBRepairDataIT {
+
+  @BeforeClass
+  public static void setUp() throws Exception {
+    EnvFactory.getEnv().initClusterEnvironment();
+  }
+
+  @AfterClass
+  public static void tearDown() throws Exception {
+    EnvFactory.getEnv().initClusterEnvironment();
+  }
+
+  @Test
+  public void testRepairData() {
+    try (Connection connection = EnvFactory.getEnv().getConnection();
+        Statement statement = connection.createStatement()) {
+      statement.execute("CREATE DATABASE root.tesgsg");
+      statement.execute("CREATE TIMESERIES root.testsg.d1.s1 WITH DATATYPE=INT32, ENCODING=PLAIN");
+      File tsfile = generateUnsortedFile();
+      statement.execute(
+          String.format("load \"%s\" verify=false", tsfile.getParentFile().getAbsolutePath()));
+
+      Assert.assertFalse(validate(statement));
+      statement.execute("REPAIR DATA");
+
+      int waitTimes = 20;
+      for (int i = 0; i < waitTimes; i++) {
+        boolean sorted = validate(statement);
+        if (sorted) {
+          return;
+        }
+        Thread.sleep(TimeUnit.SECONDS.toMillis(1));
+      }
+      Assert.fail();
+    } catch (Exception e) {
+      Assert.fail(e.getMessage());
+    }
+  }
+
+  private File generateUnsortedFile() throws IOException {
+    Path tempDir = Files.createTempDirectory("");
+    tempDir.toFile().deleteOnExit();
+    String tsfileName =
+        TsFileNameGenerator.generateNewTsFileName(System.currentTimeMillis(), 1, 0, 0);
+    File tsfile = new File(tempDir + File.separator + tsfileName);
+    Files.createFile(tsfile.toPath());
+
+    try (TsFileIOWriter writer = new TsFileIOWriter(tsfile)) {
+      writer.startChunkGroup("root.testsg.d1");
+      ChunkWriterImpl chunkWriter =
+          new ChunkWriterImpl(new MeasurementSchema("s1", TSDataType.INT32));
+      chunkWriter.write(2, 1);
+      chunkWriter.write(3, 1);
+      chunkWriter.write(5, 1);
+      chunkWriter.write(4, 1);
+      chunkWriter.sealCurrentPage();
+
+      chunkWriter.writeToFileWriter(writer);
+      writer.endChunkGroup();
+      writer.endFile();
+    }
+    return tsfile;
+  }
+
+  private boolean validate(Statement statement) throws SQLException {
+    try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.testsg.d1")) {
+      assertNotNull(resultSet);
+      long time = Long.MIN_VALUE;
+      while (resultSet.next()) {
+        long currentTime = Long.parseLong(resultSet.getString(TIMESTAMP_STR));
+        if (currentTime <= time) {
+          return false;
+        }
+        time = currentTime;
+      }
+    }
+    return true;
+  }
+}

integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByLimitOffsetAlignByDeviceIT.java

@@ -98,6 +98,7 @@ public void aggregationWithHavingTest() {
 
   @Test
   public void fillTest() {
+    // linear fill can not use TopKNode
     String[] expectedHeader = new String[] {"Time,Device,s1,s2"};
     String[] retArray =
         new String[] {

integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeSyntaxIT.java

@@ -58,6 +58,7 @@ public void testValidPipeName() throws Exception {
         (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
 
       List<String> validPipeNames = Arrays.asList("Pipe_1", "null", "`33`", "`root`", "中文", "with");
+      List<String> expectedPipeNames = Arrays.asList("Pipe_1", "null", "33", "root", "中文", "with");
       for (String pipeName : validPipeNames) {
         try (Connection connection = senderEnv.getConnection();
             Statement statement = connection.createStatement()) {
@@ -77,7 +78,7 @@ public void testValidPipeName() throws Exception {
       }
 
       List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
-      for (String pipeName : validPipeNames) {
+      for (String pipeName : expectedPipeNames) {
         Assert.assertTrue(
             showPipeResult.stream()
                 .anyMatch((o) -> o.id.equals(pipeName) && o.state.equals("RUNNING")));

iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/UserDefinedEvent.java

@@ -0,0 +1,58 @@
+/*
+ * 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.pipe.api.event;
+
+/**
+ * User defined event is used to wrap data generated by users, keeping a source event to
+ * automatically report the processing progress to pipe engine.
+ */
+public abstract class UserDefinedEvent implements Event {
+
+  /** The user defined event is generated from this source event. */
+  protected final Event sourceEvent;
+
+  /**
+   * @param sourceEvent The source event of this user defined event which is used to report the
+   *     processing progress to pipe engine. Please notice that the source event should satisfy the
+   *     following conditions: 1. A source event can only be assigned to one user defined event. 2.
+   *     If more than one user defined events are generated from the same source event, only the
+   *     last generated user defined event can be assigned with the source event, others should be
+   *     assigned {@code null}, or call {@link #UserDefinedEvent()} to generate a user defined event
+   *     without source event.
+   */
+  protected UserDefinedEvent(Event sourceEvent) {
+    this.sourceEvent = parseRootSourceEvent(sourceEvent);
+  }
+
+  /** Generate a user defined event without source event. */
+  protected UserDefinedEvent() {
+    this.sourceEvent = null;
+  }
+
+  private Event parseRootSourceEvent(Event sourceEvent) {
+    return sourceEvent instanceof UserDefinedEvent
+        ? ((UserDefinedEvent) sourceEvent).getSourceEvent()
+        : sourceEvent;
+  }
+
+  public Event getSourceEvent() {
+    return sourceEvent;
+  }
+}