Pipe: Fixed the shouldMarkAsPipeRequest for CreateTable and AlterLogicalView Sync (apache#16619)

* unwebbed

* fix

Author: Caideyipi

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java

@@ -654,7 +654,7 @@ private TSStatus executePlan(final ConfigPhysicalPlan plan) throws ConsensusExce
       case DeleteTriggerInTable:
         return configManager.dropTrigger(
             new TDropTriggerReq(((DeleteTriggerInTablePlan) plan).getTriggerName())
-                .setIsGeneratedByPipe(true));
+                .setIsGeneratedByPipe(shouldMarkAsPipeRequest.get()));
       case SetTTL:
         return ((SetTTLPlan) plan).getTTL() == TTLCache.NULL_TTL
             ? configManager
@@ -664,7 +664,8 @@ private TSStatus executePlan(final ConfigPhysicalPlan plan) throws ConsensusExce
                 .getTTLManager()
                 .setTTL((SetTTLPlan) plan, shouldMarkAsPipeRequest.get());
       case PipeCreateTableOrView:
-        return executeIdempotentCreateTableOrView((PipeCreateTableOrViewPlan) plan, queryId);
+        return executeIdempotentCreateTableOrView(
+            (PipeCreateTableOrViewPlan) plan, queryId, shouldMarkAsPipeRequest.get());
       case AddTableColumn:
         return configManager
             .getProcedureManager()
@@ -941,7 +942,10 @@ private TSStatus executePlan(final ConfigPhysicalPlan plan) throws ConsensusExce
   }
 
   private TSStatus executeIdempotentCreateTableOrView(
-      final PipeCreateTableOrViewPlan plan, final String queryId) throws ConsensusException {
+      final PipeCreateTableOrViewPlan plan,
+      final String queryId,
+      final boolean shouldMarkAsPipeRequest)
+      throws ConsensusException {
     final String database = plan.getDatabase();
     final TsTable table = plan.getTable();
     final boolean isView = TreeViewSchema.isTreeViewTable(table);
@@ -957,8 +961,8 @@ private TSStatus executeIdempotentCreateTableOrView(
                     ? ProcedureType.CREATE_TABLE_VIEW_PROCEDURE
                     : ProcedureType.CREATE_TABLE_PROCEDURE,
                 isView
-                    ? new CreateTableViewProcedure(database, table, true, true)
-                    : new CreateTableProcedure(database, table, true));
+                    ? new CreateTableViewProcedure(database, table, true, shouldMarkAsPipeRequest)
+                    : new CreateTableProcedure(database, table, shouldMarkAsPipeRequest));
     // Note that the view and its column won't be auto created
     // Skip it to avoid affecting the existing base table
     if (!isView && result.getCode() == TSStatusCode.TABLE_ALREADY_EXISTS.getStatusCode()) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java

@@ -685,7 +685,8 @@ private TPipeTransferResp handleTransferSchemaPlan(final PipeTransferPlanNodeReq
       }
       return new TPipeTransferResp(
           ClusterConfigTaskExecutor.getInstance()
-              .alterLogicalViewByPipe((AlterLogicalViewNode) req.getPlanNode()));
+              .alterLogicalViewByPipe(
+                  (AlterLogicalViewNode) req.getPlanNode(), shouldMarkAsPipeRequest.get()));
     }
     final Object statement = PLAN_TO_STATEMENT_VISITOR.process(req.getPlanNode(), null);
     return statement instanceof Statement

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java

@@ -3018,7 +3018,8 @@ public SettableFuture<ConfigTaskResult> alterLogicalView(
   }
 
   @Override
-  public TSStatus alterLogicalViewByPipe(final AlterLogicalViewNode alterLogicalViewNode) {
+  public TSStatus alterLogicalViewByPipe(
+      final AlterLogicalViewNode alterLogicalViewNode, final boolean shouldMarkAsPipeRequest) {
     final Map<PartialPath, ViewExpression> viewPathToSourceMap =
         alterLogicalViewNode.getViewPathToSourceMap();
 
@@ -3037,7 +3038,7 @@ public TSStatus alterLogicalViewByPipe(final AlterLogicalViewNode alterLogicalVi
         new TAlterLogicalViewReq(
                 Coordinator.getInstance().createQueryId().getId(),
                 ByteBuffer.wrap(stream.toByteArray()))
-            .setIsGeneratedByPipe(true);
+            .setIsGeneratedByPipe(shouldMarkAsPipeRequest);
     TSStatus tsStatus;
     try (final ConfigNodeClient client =
         CLUSTER_DELETION_CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java

@@ -250,7 +250,8 @@ SettableFuture<ConfigTaskResult> renameLogicalView(
   SettableFuture<ConfigTaskResult> alterLogicalView(
       AlterLogicalViewStatement alterLogicalViewStatement, MPPQueryContext context);
 
-  TSStatus alterLogicalViewByPipe(AlterLogicalViewNode alterLogicalViewNode);
+  TSStatus alterLogicalViewByPipe(
+      AlterLogicalViewNode alterLogicalViewNode, boolean shouldMarkAsPipeRequest);
 
   SettableFuture<ConfigTaskResult> getRegionId(GetRegionIdStatement getRegionIdStatement);