Changing point processor

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeProcessorIT.java

@@ -150,4 +150,167 @@ public void testTumblingTimeSamplingProcessor() throws Exception {
           receiverEnv, "select * from root.vehicle.**", "Time,root.vehicle.d0.s1,", expectedResSet);
     }
   }
+
+  @Test
+  public void testChangingValueProcessor() throws Exception {
+    final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+    final String receiverIp = receiverDataNode.getIp();
+    final int receiverPort = receiverDataNode.getPort();
+
+    try (final SyncConfigNodeIServiceClient client =
+        (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+      // Test empty tsFile parsing
+      // Assert that an empty tsFile will not be parsed by the processor then block
+      // the subsequent data processing
+      // Do not fail if the failure has nothing to do with pipe
+      // Because the failures will randomly generate due to resource limitation
+      if (!TestUtils.tryExecuteNonQueriesWithRetry(
+          senderEnv,
+          Arrays.asList(
+              "insert into root.vehicle.d0(time, s1) values (0, 1)", "delete from root.**"))) {
+        return;
+      }
+
+      final Map<String, String> extractorAttributes = new HashMap<>();
+      final Map<String, String> processorAttributes = new HashMap<>();
+      final Map<String, String> connectorAttributes = new HashMap<>();
+
+      extractorAttributes.put("source.realtime.mode", "log");
+
+      processorAttributes.put("processor", "changing-value-sampling-processor");
+      processorAttributes.put("processor.changing-value.compression-deviation", "10");
+      processorAttributes.put("processor.changing-value.min-time-interval", "10000");
+      processorAttributes.put("processor.changing-value.max-time-interval", "20000");
+
+      connectorAttributes.put("sink", "iotdb-thrift-sink");
+      connectorAttributes.put("sink.batch.enable", "false");
+      connectorAttributes.put("sink.ip", receiverIp);
+      connectorAttributes.put("sink.port", Integer.toString(receiverPort));
+
+      final TSStatus status =
+          client.createPipe(
+              new TCreatePipeReq("testPipe", connectorAttributes)
+                  .setExtractorAttributes(extractorAttributes)
+                  .setProcessorAttributes(processorAttributes));
+
+      Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("testPipe").getCode());
+
+      if (!TestUtils.tryExecuteNonQueriesWithRetry(
+          senderEnv,
+          Arrays.asList(
+              "insert into root.vehicle.d0(time, s1) values (0, 10)",
+              "insert into root.vehicle.d0(time, s1) values (9999, 20)",
+              "insert into root.vehicle.d0(time, s1) values (10000, 30)",
+              "insert into root.vehicle.d0(time, s1) values (19000, 40)",
+              "insert into root.vehicle.d0(time, s1) values (20000, 50)",
+              "insert into root.vehicle.d0(time, s1) values (29001, 60)",
+              "insert into root.vehicle.d0(time, s1) values (50000, 70)",
+              "insert into root.vehicle.d0(time, s1) values (60000, 71)",
+              "flush"))) {
+        return;
+      }
+
+      final Set<String> expectedResSet = new HashSet<>();
+
+      expectedResSet.add("0,10.0,");
+      expectedResSet.add("10000,30.0,");
+      expectedResSet.add("20000,50.0,");
+      expectedResSet.add("50000,70.0,");
+
+      TestUtils.assertDataEventuallyOnEnv(
+          receiverEnv, "select * from root.**", "Time,root.vehicle.d0.s1,", expectedResSet);
+    }
+  }
+
+  @Test
+  public void testChangingPointProcessor() throws Exception {
+    final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+    final String receiverIp = receiverDataNode.getIp();
+    final int receiverPort = receiverDataNode.getPort();
+
+    try (final SyncConfigNodeIServiceClient client =
+        (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+      // Test empty tsFile parsing
+      // Assert that an empty tsFile will not be parsed by the processor then block
+      // the subsequent data processing
+      // Do not fail if the failure has nothing to do with pipe
+      // Because the failures will randomly generate due to resource limitation
+      if (!TestUtils.tryExecuteNonQueriesWithRetry(
+          senderEnv,
+          Arrays.asList(
+              "insert into root.vehicle.d0(time, s1) values (0, 1)", "delete from root.**"))) {
+        return;
+      }
+
+      final Map<String, String> extractorAttributes = new HashMap<>();
+      final Map<String, String> processorAttributes = new HashMap<>();
+      final Map<String, String> connectorAttributes = new HashMap<>();
+
+      extractorAttributes.put("source.realtime.mode", "log");
+
+      processorAttributes.put("processor", "changing-point-sampling-processor");
+      processorAttributes.put("processor.changing-point.value-variation", "10");
+      processorAttributes.put("processor.changing-point.arrival-time.min-interval", "10000");
+      processorAttributes.put("processor.changing-point.arrival-time.max-interval", "30000");
+      processorAttributes.put("processor.changing-point.event-time.min-interval", "10000");
+      processorAttributes.put("processor.changing-point.event-time.max-interval", "30000");
+
+      connectorAttributes.put("sink", "iotdb-thrift-sink");
+      connectorAttributes.put("sink.batch.enable", "false");
+      connectorAttributes.put("sink.ip", receiverIp);
+      connectorAttributes.put("sink.port", Integer.toString(receiverPort));
+
+      final TSStatus status =
+          client.createPipe(
+              new TCreatePipeReq("testPipe", connectorAttributes)
+                  .setExtractorAttributes(extractorAttributes)
+                  .setProcessorAttributes(processorAttributes));
+
+      Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("testPipe").getCode());
+
+      if (!TestUtils.tryExecuteNonQueriesWithRetry(
+          senderEnv,
+          Arrays.asList(
+              "insert into root.vehicle.d0(time, s1) values (0, 10)",
+              "insert into root.vehicle.d0(time, s1) values (100000, 20)",
+              "insert into root.vehicle.d0(time, s1) values (110000, 30)"))) {
+        return;
+      }
+
+      Thread.sleep(10000);
+
+      if (!TestUtils.tryExecuteNonQueriesWithRetry(
+          senderEnv,
+          Arrays.asList(
+              "insert into root.vehicle.d0(time, s1) values (100000, 40)",
+              "insert into root.vehicle.d0(time, s1) values (400000, 50)",
+              "insert into root.vehicle.d0(time, s1) values (500000, 60)"))) {
+        return;
+      }
+
+      Thread.sleep(10000);
+
+      if (!TestUtils.tryExecuteNonQueriesWithRetry(
+          senderEnv,
+          Arrays.asList("insert into root.vehicle.d0(time, s1) values (100000, 41)", "flush"))) {
+        return;
+      }
+
+      final Set<String> expectedResSet = new HashSet<>();
+
+      expectedResSet.add("0,10.0,");
+      expectedResSet.add("100000,40.0,");
+
+      TestUtils.assertDataEventuallyOnEnv(
+          receiverEnv, "select * from root.**", "Time,root.vehicle.d0.s1,", expectedResSet);
+    }
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionProcessorConstructor.java

@@ -27,7 +27,7 @@
 import org.apache.iotdb.db.pipe.processor.aggregate.AggregateProcessor;
 import org.apache.iotdb.db.pipe.processor.aggregate.operator.processor.StandardStatisticsOperatorProcessor;
 import org.apache.iotdb.db.pipe.processor.aggregate.window.processor.TumblingWindowingProcessor;
-import org.apache.iotdb.db.pipe.processor.downsampling.changing.ChangingValueSamplingProcessor;
+import org.apache.iotdb.db.pipe.processor.downsampling.changing.ChangingPointSamplingProcessor;
 import org.apache.iotdb.db.pipe.processor.downsampling.sdt.SwingingDoorTrendingSamplingProcessor;
 import org.apache.iotdb.db.pipe.processor.downsampling.tumbling.TumblingTimeSamplingProcessor;
 import org.apache.iotdb.db.pipe.processor.pipeconsensus.PipeConsensusProcessor;
@@ -52,7 +52,10 @@ protected void initConstructors() {
         SwingingDoorTrendingSamplingProcessor::new);
     pluginConstructors.put(
         BuiltinPipePlugin.CHANGING_VALUE_SAMPLING_PROCESSOR.getPipePluginName(),
-        ChangingValueSamplingProcessor::new);
+        ChangingPointSamplingProcessor::new);
+    pluginConstructors.put(
+        BuiltinPipePlugin.CHANGING_POINT_SAMPLING_PROCESSOR.getPipePluginName(),
+        ChangingPointSamplingProcessor::new);
     pluginConstructors.put(
         BuiltinPipePlugin.THROWING_EXCEPTION_PROCESSOR.getPipePluginName(),
         ThrowingExceptionProcessor::new);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/DownSamplingFilter.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.processor.downsampling;
+
+import org.apache.tsfile.utils.Accountable;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.time.LocalDate;
+
+public abstract class DownSamplingFilter implements Accountable {
+
+  protected static final long SIZE_OF_DATE =
+      RamUsageEstimator.shallowSizeOfInstance(LocalDate.class);
+
+  protected static final long SIZE_OF_LONG = RamUsageEstimator.shallowSizeOfInstance(Long.class);
+
+  protected static final long SIZE_OF_BINARY =
+      RamUsageEstimator.shallowSizeOfInstance(Binary.class);
+
+  protected boolean isFilteredByArrivalTime = true;
+
+  protected long lastPointArrivalTime;
+
+  protected long lastPointEventTime;
+
+  public DownSamplingFilter(
+      final long arrivalTime, final long eventTime, final boolean isFilteredByArrivalTime) {
+    this.lastPointArrivalTime = arrivalTime;
+    this.lastPointEventTime = eventTime;
+    this.isFilteredByArrivalTime = isFilteredByArrivalTime;
+  }
+
+  public DownSamplingFilter(final long arrivalTime, final long eventTime) {
+    this.lastPointArrivalTime = arrivalTime;
+    this.lastPointEventTime = eventTime;
+  }
+
+  public void reset(final long arrivalTime, final long eventTime) {
+    this.lastPointArrivalTime = arrivalTime;
+    this.lastPointEventTime = eventTime;
+  }
+
+  public long getLastPointArrivalTime() {
+    return lastPointArrivalTime;
+  }
+
+  public long getLastPointEventTime() {
+    return lastPointEventTime;
+  }
+
+  public boolean isFilteredByArrivalTime() {
+    return isFilteredByArrivalTime;
+  }
+}