reduce log

refactor Refactor Update PipeInsertNodeTabletInsertionEvent.java
apache · Mar 12, 2025 · d032771 · d032771
1 parent cc0a99a
commit d032771
Show file tree

Hide file tree

Showing 7 changed files with 26 additions and 34 deletions.
diff --git a/...de/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcda/OpcDaServerHandle.java b/...de/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcda/OpcDaServerHandle.java
@@ -194,9 +194,7 @@ void transfer(final Tablet tablet) {
         addItem(itemId, schemas.get(i).getType());
       }
       for (int j = tablet.getRowSize() - 1; j >= 0; --j) {
-        if (Objects.isNull(tablet.getBitMaps())
-            || Objects.isNull(tablet.getBitMaps()[i])
-            || !tablet.getBitMaps()[i].isMarked(j)) {
+        if (!tablet.isNull(j, i)) {
           if (serverTimestampMap.get(itemId) <= tablet.getTimestamp(j)) {
             writeData(
                 itemId,

diff --git a/...anode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcua/OpcUaNameSpace.java b/...anode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcua/OpcUaNameSpace.java
@@ -124,9 +124,7 @@ private void transferTabletForClientServerModel(final Tablet tablet, final boole
 
       for (int i = 0; i < schemas.size(); ++i) {
         for (int j = tablet.getRowSize() - 1; j >= 0; --j) {
-          if (Objects.isNull(tablet.getBitMaps())
-              || Objects.isNull(tablet.getBitMaps()[i])
-              || !tablet.getBitMaps()[i].isMarked(j)) {
+          if (!tablet.isNull(j, i)) {
             newSchemas.add(schemas.get(i));
             timestamps.add(tablet.getTimestamp(j));
             values.add(

diff --git a/...core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/PipeInsertionEvent.java b/...core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/PipeInsertionEvent.java
@@ -28,9 +28,6 @@
 
 import javax.validation.constraints.NotNull;
 
-import java.util.HashSet;
-import java.util.Set;
-
 /**
  * The data model used to record the Event and the data model of the DataRegion corresponding to the
  * source data, so this type requires some specifications .
@@ -60,8 +57,6 @@ public abstract class PipeInsertionEvent extends EnrichedEvent {
   protected String treeModelDatabaseName; // lazy initialization
   protected String tableModelDatabaseName; // lazy initialization
 
-  private Set<String> noPrivilegeTableNames = new HashSet<>();
-
   protected PipeInsertionEvent(
       final String pipeName,
       final long creationTime,
@@ -174,12 +169,4 @@ public void renameTableModelDatabase(@NotNull final String tableModelDatabaseNam
     this.tableModelDatabaseName = tableModelDatabaseName.toLowerCase();
     this.treeModelDatabaseName = PathUtils.qualifyDatabaseName(tableModelDatabaseName);
   }
-
-  public void addTable(final String tableName) {
-    noPrivilegeTableNames.add(tableName);
-  }
-
-  public Set<String> getNoPrivilegeTableNames() {
-    return noPrivilegeTableNames;
-  }
 }
diff --git a/...java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java b/...java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
@@ -85,7 +85,8 @@ public class PipeInsertNodeTabletInsertionEvent extends PipeInsertionEvent
           + RamUsageEstimator.shallowSizeOfInstance(WALEntryHandler.class)
           + RamUsageEstimator.shallowSizeOfInstance(WALEntryPosition.class)
           + RamUsageEstimator.shallowSizeOfInstance(AtomicInteger.class)
-          + RamUsageEstimator.shallowSizeOfInstance(AtomicBoolean.class);
+          + RamUsageEstimator.shallowSizeOfInstance(AtomicBoolean.class)
+          + RamUsageEstimator.shallowSizeOf(Boolean.class);
   private static final long SET_SIZE = RamUsageEstimator.shallowSizeOfInstance(HashSet.class);
 
   private final WALEntryHandler walEntryHandler;
@@ -579,6 +580,12 @@ public long ramBytesUsed() {
     return INSTANCE_SIZE
         + (Objects.nonNull(devicePath) ? PartialPath.estimateSize(devicePath) : 0)
         + (Objects.nonNull(progressIndex) ? progressIndex.ramBytesUsed() : 0)
+        + (Objects.nonNull(treeModelDatabaseName)
+            ? RamUsageEstimator.sizeOf(treeModelDatabaseName)
+            : 0)
+        + (Objects.nonNull(tableModelDatabaseName)
+            ? RamUsageEstimator.sizeOf(tableModelDatabaseName)
+            : 0)
         + (Objects.nonNull(tableNames)
             ? SET_SIZE
                 + tableNames.stream().mapToLong(RamUsageEstimator::sizeOf).reduce(0L, Long::sum)

diff --git a/...c/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java b/...c/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
@@ -363,8 +363,8 @@ private TabletInsertionEventParser initEventParser() {
   }
 
   public long count() {
-    final Tablet covertedTablet = shouldParseTimeOrPattern() ? convertToTablet() : tablet;
-    return (long) covertedTablet.getRowSize() * covertedTablet.getSchemas().size();
+    final Tablet convertedTablet = shouldParseTimeOrPattern() ? convertToTablet() : tablet;
+    return (long) convertedTablet.getRowSize() * convertedTablet.getSchemas().size();
   }
 
   /////////////////////////// parsePatternOrTime ///////////////////////////

diff --git a/.../java/org/apache/iotdb/db/pipe/event/common/tablet/parser/TabletInsertionEventParser.java b/.../java/org/apache/iotdb/db/pipe/event/common/tablet/parser/TabletInsertionEventParser.java
@@ -142,8 +142,8 @@ protected void parse(final InsertRowNode insertRowNode) {
     final MeasurementSchema[] originMeasurementSchemaList = insertRowNode.getMeasurementSchemas();
     final String[] originColumnNameStringList = insertRowNode.getMeasurements();
     final TsTableColumnCategory[] originColumnCategories = insertRowNode.getColumnCategories();
-    final TSDataType[] originValueColumnDataTypes = insertRowNode.getDataTypes();
-    final Object[] originValueColumns = insertRowNode.getValues();
+    final TSDataType[] originValueDataTypes = insertRowNode.getDataTypes();
+    final Object[] originValues = insertRowNode.getValues();
 
     for (int i = 0; i < originColumnIndex2FilteredColumnIndexMapperList.length; i++) {
       if (originColumnIndex2FilteredColumnIndexMapperList[i] != null) {
@@ -154,21 +154,20 @@ protected void parse(final InsertRowNode insertRowNode) {
             originColumnCategories != null && originColumnCategories[i] != null
                 ? originColumnCategories[i].toTsFileColumnType()
                 : Tablet.ColumnCategory.FIELD;
-        this.valueColumnDataTypes[filteredColumnIndex] = originValueColumnDataTypes[i];
+        this.valueColumnDataTypes[filteredColumnIndex] = originValueDataTypes[i];
         final BitMap bitMap = new BitMap(this.timestampColumn.length);
-        if (Objects.isNull(originValueColumns[i])
-            || Objects.isNull(originValueColumnDataTypes[i])) {
+        if (Objects.isNull(originValues[i]) || Objects.isNull(originValueDataTypes[i])) {
           fillNullValue(
-              originValueColumnDataTypes[i],
+              originValueDataTypes[i],
               this.valueColumns,
               bitMap,
               filteredColumnIndex,
               rowIndexList.size());
         } else {
           this.valueColumns[filteredColumnIndex] =
               filterValueColumnsByRowIndexList(
-                  originValueColumnDataTypes[i],
-                  originValueColumns[i],
+                  originValueDataTypes[i],
+                  originValues[i],
                   rowIndexList,
                   true,
                   bitMap, // use the output bitmap since there is no bitmap in InsertRowNode

diff --git a/...he/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionHybridExtractor.java b/...he/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionHybridExtractor.java
@@ -82,13 +82,17 @@ public boolean isNeedListenToInsertNode() {
   }
 
   private void extractTabletInsertion(final PipeRealtimeEvent event) {
-    if (canNotUseTabletAnyMore(event)) {
+    final TsFileEpoch.State state = event.getTsFileEpoch().getState(this);
+
+    if (state != TsFileEpoch.State.USING_TSFILE
+        && state != TsFileEpoch.State.USING_BOTH
+        && canNotUseTabletAnyMore(event)) {
       event
           .getTsFileEpoch()
           .migrateState(
               this,
-              state -> {
-                switch (state) {
+              curState -> {
+                switch (curState) {
                   case EMPTY:
                   case USING_TSFILE:
                     return TsFileEpoch.State.USING_TSFILE;
@@ -100,7 +104,6 @@ private void extractTabletInsertion(final PipeRealtimeEvent event) {
               });
     }
 
-    final TsFileEpoch.State state = event.getTsFileEpoch().getState(this);
     switch (state) {
       case USING_TSFILE:
         // Ignore the tablet event.