may-comp

Author: Caideyipi

iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/collector/DataCollector.java

@@ -24,5 +24,5 @@
 import java.util.List;
 
 public interface DataCollector {
-    List<TabletInsertionEvent> convertToTabletInsertionEvents(final boolean shouldReport);
+  List<TabletInsertionEvent> convertToTabletInsertionEvents(final boolean shouldReport);
 }

iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/collector/RowCollector.java

@@ -22,15 +22,14 @@
 import org.apache.iotdb.pipe.api.access.Row;
 import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
 
-import javax.xml.crypto.Data;
 import java.io.IOException;
 import java.util.function.BiConsumer;
 
 /**
  * Used to collect rows generated by {@link TabletInsertionEvent#processRowByRow(BiConsumer)},
  * {@link TabletInsertionEvent#processTablet(BiConsumer)}.
  */
-public interface RowCollector extends DataCollector {
+public interface RowCollector {
 
   /**
    * Collects a row.

iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/collector/TabletCollector.java

@@ -19,11 +19,10 @@
 
 package org.apache.iotdb.pipe.api.collector;
 
-import org.apache.iotdb.pipe.api.access.Row;
 import org.apache.tsfile.write.record.Tablet;
 
 import java.io.IOException;
 
-public interface TabletCollector extends DataCollector {
-    void collectTablet(Tablet tablet) throws IOException;
+public interface TabletCollector {
+  void collectTablet(Tablet tablet) throws IOException;
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/row/PipeRowCollector.java

@@ -22,6 +22,7 @@
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
 import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
 import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent;
+import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletEventConverter;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
 import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil;
 import org.apache.iotdb.pipe.api.access.Row;
@@ -38,38 +39,19 @@
 import java.util.Arrays;
 import java.util.List;
 
-public class PipeRowCollector implements RowCollector {
-
-  private final List<TabletInsertionEvent> tabletInsertionEventList = new ArrayList<>();
+public class PipeRowCollector extends PipeRawTabletEventConverter implements RowCollector {
   private Tablet tablet = null;
-  private boolean isAligned = false;
-  private final PipeTaskMeta pipeTaskMeta; // Used to report progress
-  private final EnrichedEvent sourceEvent; // Used to report progress
-  private final String sourceEventDataBaseName;
-  private final Boolean isTableModel;
 
   public PipeRowCollector(PipeTaskMeta pipeTaskMeta, EnrichedEvent sourceEvent) {
-    this.pipeTaskMeta = pipeTaskMeta;
-    this.sourceEvent = sourceEvent;
-    if (sourceEvent instanceof PipeInsertionEvent) {
-      sourceEventDataBaseName =
-          ((PipeInsertionEvent) sourceEvent).getSourceDatabaseNameFromDataRegion();
-      isTableModel = ((PipeInsertionEvent) sourceEvent).getRawIsTableModelEvent();
-    } else {
-      sourceEventDataBaseName = null;
-      isTableModel = null;
-    }
+    super(pipeTaskMeta, sourceEvent);
   }
 
   public PipeRowCollector(
       PipeTaskMeta pipeTaskMeta,
       EnrichedEvent sourceEvent,
       String sourceEventDataBase,
       Boolean isTableModel) {
-    this.pipeTaskMeta = pipeTaskMeta;
-    this.sourceEvent = sourceEvent;
-    this.sourceEventDataBaseName = sourceEventDataBase;
-    this.isTableModel = isTableModel;
+    super(pipeTaskMeta, sourceEvent, sourceEventDataBase, isTableModel);
   }
 
   @Override
@@ -145,12 +127,6 @@ private void collectTabletInsertionEvent() {
   @Override
   public List<TabletInsertionEvent> convertToTabletInsertionEvents(final boolean shouldReport) {
     collectTabletInsertionEvent();
-
-    final int eventListSize = tabletInsertionEventList.size();
-    if (eventListSize > 0 && shouldReport) { // The last event should report progress
-      ((PipeRawTabletInsertionEvent) tabletInsertionEventList.get(eventListSize - 1))
-          .markAsNeedToReport();
-    }
-    return tabletInsertionEventList;
+    return super.convertToTabletInsertionEvents(shouldReport);
   }
 }

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

@@ -0,0 +1,72 @@
+/*
+ * 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.common.tablet;
+
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent;
+import org.apache.iotdb.pipe.api.collector.DataCollector;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public abstract class PipeRawTabletEventConverter implements DataCollector {
+  protected final List<TabletInsertionEvent> tabletInsertionEventList = new ArrayList<>();
+  protected boolean isAligned = false;
+  protected final PipeTaskMeta pipeTaskMeta; // Used to report progress
+  protected final EnrichedEvent sourceEvent; // Used to report progress
+  protected final String sourceEventDataBaseName;
+  protected final Boolean isTableModel;
+
+  public PipeRawTabletEventConverter(PipeTaskMeta pipeTaskMeta, EnrichedEvent sourceEvent) {
+    this.pipeTaskMeta = pipeTaskMeta;
+    this.sourceEvent = sourceEvent;
+    if (sourceEvent instanceof PipeInsertionEvent) {
+      sourceEventDataBaseName =
+          ((PipeInsertionEvent) sourceEvent).getSourceDatabaseNameFromDataRegion();
+      isTableModel = ((PipeInsertionEvent) sourceEvent).getRawIsTableModelEvent();
+    } else {
+      sourceEventDataBaseName = null;
+      isTableModel = null;
+    }
+  }
+
+  public PipeRawTabletEventConverter(
+      PipeTaskMeta pipeTaskMeta,
+      EnrichedEvent sourceEvent,
+      String sourceEventDataBase,
+      Boolean isTableModel) {
+    this.pipeTaskMeta = pipeTaskMeta;
+    this.sourceEvent = sourceEvent;
+    this.sourceEventDataBaseName = sourceEventDataBase;
+    this.isTableModel = isTableModel;
+  }
+
+  @Override
+  public List<TabletInsertionEvent> convertToTabletInsertionEvents(final boolean shouldReport) {
+    final int eventListSize = tabletInsertionEventList.size();
+    if (eventListSize > 0 && shouldReport) { // The last event should report progress
+      ((PipeRawTabletInsertionEvent) tabletInsertionEventList.get(eventListSize - 1))
+          .markAsNeedToReport();
+    }
+    return tabletInsertionEventList;
+  }
+}

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

@@ -19,5 +19,45 @@
 
 package org.apache.iotdb.db.pipe.event.common.tablet;
 
-public class PipeTabletCollector {
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent;
+import org.apache.iotdb.pipe.api.collector.TabletCollector;
+
+import org.apache.tsfile.write.record.Tablet;
+
+import java.io.IOException;
+
+public class PipeTabletCollector extends PipeRawTabletEventConverter implements TabletCollector {
+
+  public PipeTabletCollector(PipeTaskMeta pipeTaskMeta, EnrichedEvent sourceEvent) {
+    super(pipeTaskMeta, sourceEvent);
+  }
+
+  public PipeTabletCollector(
+      PipeTaskMeta pipeTaskMeta,
+      EnrichedEvent sourceEvent,
+      String sourceEventDataBase,
+      Boolean isTableModel) {
+    super(pipeTaskMeta, sourceEvent, sourceEventDataBase, isTableModel);
+  }
+
+  @Override
+  public void collectTablet(final Tablet tablet) {
+    final PipeInsertionEvent pipeInsertionEvent =
+        sourceEvent instanceof PipeInsertionEvent ? ((PipeInsertionEvent) sourceEvent) : null;
+    tabletInsertionEventList.add(
+        new PipeRawTabletInsertionEvent(
+            isTableModel,
+            sourceEventDataBaseName,
+            pipeInsertionEvent == null ? null : pipeInsertionEvent.getRawTableModelDataBase(),
+            pipeInsertionEvent == null ? null : pipeInsertionEvent.getRawTreeModelDataBase(),
+            tablet,
+            isAligned,
+            sourceEvent == null ? null : sourceEvent.getPipeName(),
+            sourceEvent == null ? 0 : sourceEvent.getCreationTime(),
+            pipeTaskMeta,
+            sourceEvent,
+            false));
+  }
 }