Pipe: Removed the failed partial insert from auth check & Implemented skipReportOnCommit for PipeRealtimeEvent to avoid premature report for unordered flush (#17561)

* schema

* sptls

* fix

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java

@@ -325,6 +325,10 @@ private void checkTreePattern(final IDeviceID deviceID, final String[] measureme
       throws IllegalPathException {
     final List<MeasurementPath> measurementList = new ArrayList<>();
     for (final String measurement : measurements) {
+      // Ignore failed measurements in partial inserts, consistent with downstream matching/parsing.
+      if (measurement == null) {
+        continue;
+      }
       if (treePattern.matchesMeasurement(deviceID, measurement)) {
         measurementList.add(new MeasurementPath(deviceID, measurement));
       }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java

@@ -164,6 +164,16 @@ public ProgressIndex getProgressIndex() {
     return event.getProgressIndex();
   }
 
+  @Override
+  public void skipReportOnCommit() {
+    event.skipReportOnCommit();
+  }
+
+  @Override
+  public boolean isShouldReportOnCommit() {
+    return event.isShouldReportOnCommit();
+  }
+
   @Override
   public void skipParsingPattern() {
     event.skipParsingPattern();

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeRealtimeEventTest.java

@@ -0,0 +1,101 @@
+/*
+ * 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.pipe.event;
+
+import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
+import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class PipeRealtimeEventTest {
+
+  @Test
+  public void testSkipReportOnCommitIsDelegatedToInnerEvent() {
+    final TestEnrichedEvent innerEvent = new TestEnrichedEvent();
+    final PipeRealtimeEvent realtimeEvent = new PipeRealtimeEvent(innerEvent, null, null);
+
+    Assert.assertTrue(innerEvent.isShouldReportOnCommit());
+    Assert.assertTrue(realtimeEvent.isShouldReportOnCommit());
+
+    realtimeEvent.skipReportOnCommit();
+
+    Assert.assertFalse(innerEvent.isShouldReportOnCommit());
+    Assert.assertFalse(realtimeEvent.isShouldReportOnCommit());
+  }
+
+  private static class TestEnrichedEvent extends EnrichedEvent {
+
+    private TestEnrichedEvent() {
+      super(null, 0, null, null, null, null, null, null, false, Long.MIN_VALUE, Long.MAX_VALUE);
+    }
+
+    @Override
+    public boolean internallyIncreaseResourceReferenceCount(final String holderMessage) {
+      return true;
+    }
+
+    @Override
+    public boolean internallyDecreaseResourceReferenceCount(final String holderMessage) {
+      return true;
+    }
+
+    @Override
+    public ProgressIndex getProgressIndex() {
+      return MinimumProgressIndex.INSTANCE;
+    }
+
+    @Override
+    public EnrichedEvent shallowCopySelfAndBindPipeTaskMetaForProgressReport(
+        final String pipeName,
+        final long creationTime,
+        final PipeTaskMeta pipeTaskMeta,
+        final TreePattern treePattern,
+        final TablePattern tablePattern,
+        final String userId,
+        final String userName,
+        final String cliHostname,
+        final boolean skipIfNoPrivileges,
+        final long startTime,
+        final long endTime) {
+      return this;
+    }
+
+    @Override
+    public boolean isGeneratedByPipe() {
+      return false;
+    }
+
+    @Override
+    public boolean mayEventTimeOverlappedWithTimeRange() {
+      return true;
+    }
+
+    @Override
+    public boolean mayEventPathsOverlappedWithPattern() {
+      return true;
+    }
+  }
+}

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTabletInsertionEventTest.java

@@ -19,6 +19,9 @@
 
 package org.apache.iotdb.db.pipe.event;
 
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.audit.IAuditEntity;
+import org.apache.iotdb.commons.auth.entity.PrivilegeType;
 import org.apache.iotdb.commons.exception.IllegalPathException;
 import org.apache.iotdb.commons.exception.auth.AccessDeniedException;
 import org.apache.iotdb.commons.path.PartialPath;
@@ -471,4 +474,59 @@ public void testAuthCheck() {
       event.close();
     }
   }
+
+  @Test
+  public void testAuthCheckIgnoresNullMeasurementInPartialInsert() throws Exception {
+    insertRowNode.markFailedMeasurement(1);
+
+    final PipeInsertNodeTabletInsertionEvent event =
+        new PipeInsertNodeTabletInsertionEvent(
+            false,
+            "root.db",
+            insertRowNode,
+            null,
+            0,
+            null,
+            new PrefixTreePattern(pattern),
+            new TablePattern(true, null, null),
+            "0",
+            "user",
+            "localhost",
+            false,
+            Long.MIN_VALUE,
+            Long.MAX_VALUE);
+    final AccessControl oldControl = AuthorityChecker.getAccessControl();
+    final NullMeasurementRejectingAccessControl accessControl =
+        new NullMeasurementRejectingAccessControl();
+    try {
+      AuthorityChecker.setAccessControl(accessControl);
+
+      event.throwIfNoPrivilege();
+
+      Assert.assertFalse(accessControl.hasNullMeasurementPath);
+      Assert.assertFalse(event.shouldParse4Privilege());
+    } finally {
+      AuthorityChecker.setAccessControl(oldControl);
+      event.close();
+    }
+  }
+
+  private static class NullMeasurementRejectingAccessControl
+      extends PipeTsFileInsertionEventTest.TestAccessControl {
+
+    private boolean hasNullMeasurementPath = false;
+
+    @Override
+    public TSStatus checkSeriesPrivilege4Pipe(
+        final IAuditEntity context,
+        final java.util.List<? extends PartialPath> checkedPathsSupplier,
+        final PrivilegeType permission) {
+      hasNullMeasurementPath =
+          checkedPathsSupplier.stream().anyMatch(path -> path.getFullPath().endsWith(".null"));
+      return hasNullMeasurementPath
+          ? AuthorityChecker.getTSStatus(
+              Collections.singletonList(0), checkedPathsSupplier, permission)
+          : new TSStatus(org.apache.iotdb.rpc.TSStatusCode.SUCCESS_STATUS.getStatusCode());
+    }
+  }
 }