This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch xianyi
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/xianyi by this push:
new b28786f add trigger moving_extrem
b28786f is described below
commit b28786fb10d17e47a308a3652a8007f896c70db7
Author: Steve Yurong Su <[email protected]>
AuthorDate: Fri Nov 19 17:07:58 2021 +0800
add trigger moving_extrem
---
.../main/java/org/apache/iotdb/TriggerExample.java | 139 +++++++++++++++++++++
.../trigger/builtin/MovingExtremeTrigger.java | 120 ++++++++++++++++++
2 files changed, 259 insertions(+)
diff --git a/example/session/src/main/java/org/apache/iotdb/TriggerExample.java
b/example/session/src/main/java/org/apache/iotdb/TriggerExample.java
new file mode 100644
index 0000000..9559366
--- /dev/null
+++ b/example/session/src/main/java/org/apache/iotdb/TriggerExample.java
@@ -0,0 +1,139 @@
+/*
+ * 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;
+
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.SessionDataSet;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
+import org.apache.iotdb.tsfile.write.schema.UnaryMeasurementSchema;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+@SuppressWarnings("squid:S106")
+public class TriggerExample {
+
+ private static Session session;
+
+ private static final String ROOT_SG1_D1_S1 = "root.sg1.d1.s1";
+
+ private static final String LOCAL_HOST = "127.0.0.1";
+
+ public static void main(String[] args)
+ throws IoTDBConnectionException, StatementExecutionException {
+ session =
+ new
Session.Builder().host(LOCAL_HOST).port(6667).username("root").password("root").build();
+ session.open(false);
+
+ createTimeseries();
+
+ createTrigger();
+
+ insertTablet();
+
+ query();
+
+ dropTrigger();
+
+ session.close();
+ }
+
+ private static void createTimeseries()
+ throws IoTDBConnectionException, StatementExecutionException {
+ if (!session.checkTimeseriesExists(ROOT_SG1_D1_S1)) {
+ session.createTimeseries(
+ ROOT_SG1_D1_S1, TSDataType.DOUBLE, TSEncoding.GORILLA,
CompressionType.SNAPPY);
+ }
+ }
+
+ private static void createTrigger() throws IoTDBConnectionException,
StatementExecutionException {
+ session.executeNonQueryStatement(
+ "CREATE TRIGGER moving_extreme "
+ + "AFTER INSERT "
+ + "ON root.sg1.d1.s1 "
+ + "AS
'org.apache.iotdb.db.engine.trigger.builtin.MovingExtremeTrigger'"
+ + "WITH ("
+ + " 'device' = 'root.extreme.sg1.d1', "
+ + " 'measurement' = 's1'"
+ + ")");
+ }
+
+ private static void dropTrigger() throws IoTDBConnectionException,
StatementExecutionException {
+ session.executeNonQueryStatement("drop trigger moving_extreme");
+ }
+
+ /**
+ * insert the data of a device. For each timestamp, the number of
measurements is the same.
+ *
+ * <p>Users need to control the count of Tablet and write a batch when it
reaches the maxBatchSize
+ */
+ private static void insertTablet() throws IoTDBConnectionException,
StatementExecutionException {
+ /*
+ * A Tablet example:
+ * device1
+ * time s1, s2, s3
+ * 1, 1, 1, 1
+ * 2, 2, 2, 2
+ * 3, 3, 3, 3
+ */
+ // The schema of measurements of one device
+ // only measurementId and data type in MeasurementSchema take effects in
Tablet
+ List<IMeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new UnaryMeasurementSchema("s1", TSDataType.DOUBLE));
+
+ Tablet tablet = new Tablet("root.sg1.d1", schemaList, 100000);
+
+ // Method 1 to add tablet data
+ long timestamp = System.currentTimeMillis();
+
+ for (long row = 0; row < 10000000; row++) {
+ int rowIndex = tablet.rowSize++;
+ tablet.addTimestamp(rowIndex, timestamp);
+ tablet.addValue(schemaList.get(0).getMeasurementId(), rowIndex, new
Random().nextDouble());
+
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ session.insertTablet(tablet, true);
+ tablet.reset();
+ }
+ timestamp++;
+ }
+
+ if (tablet.rowSize != 0) {
+ session.insertTablet(tablet);
+ tablet.reset();
+ }
+ }
+
+ private static void query() throws IoTDBConnectionException,
StatementExecutionException {
+ try (SessionDataSet dataSet = session.executeQueryStatement("select **
from root")) {
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/trigger/builtin/MovingExtremeTrigger.java
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/builtin/MovingExtremeTrigger.java
new file mode 100644
index 0000000..7cd8690
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/trigger/builtin/MovingExtremeTrigger.java
@@ -0,0 +1,120 @@
+/*
+ * 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.engine.trigger.builtin;
+
+import org.apache.iotdb.db.engine.trigger.api.Trigger;
+import org.apache.iotdb.db.engine.trigger.api.TriggerAttributes;
+import org.apache.iotdb.db.engine.trigger.sink.local.LocalIoTDBConfiguration;
+import org.apache.iotdb.db.engine.trigger.sink.local.LocalIoTDBEvent;
+import org.apache.iotdb.db.engine.trigger.sink.local.LocalIoTDBHandler;
+import org.apache.iotdb.db.utils.windowing.api.Evaluator;
+import org.apache.iotdb.db.utils.windowing.api.Window;
+import
org.apache.iotdb.db.utils.windowing.configuration.SlidingSizeWindowConfiguration;
+import
org.apache.iotdb.db.utils.windowing.handler.SlidingSizeWindowEvaluationHandler;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+public class MovingExtremeTrigger implements Trigger {
+
+ private final LocalIoTDBHandler localIoTDBHandler = new LocalIoTDBHandler();
+
+ private SlidingSizeWindowEvaluationHandler windowEvaluationHandler;
+
+ private String device;
+ private String[] measurements;
+ private TSDataType[] dataTypes;
+
+ @Override
+ public void onCreate(TriggerAttributes attributes) throws Exception {
+ device = attributes.getString("device");
+ measurements = new String[] {attributes.getString("measurement")};
+ dataTypes = new TSDataType[] {TSDataType.DOUBLE};
+
+ openSinkHandlers();
+
+ windowEvaluationHandler =
+ new SlidingSizeWindowEvaluationHandler(
+ new SlidingSizeWindowConfiguration(TSDataType.DOUBLE, 100, 100),
+ new Evaluator() {
+
+ @Override
+ public void evaluate(Window window) throws Exception {
+ double extreme = 0;
+ double[] array = window.getDoubleArray();
+ for (int i = 0, n = window.size(); i < n; ++i) {
+ extreme = Math.max(extreme, Math.abs(array[i]));
+ }
+
+ localIoTDBHandler.onEvent(new
LocalIoTDBEvent(window.getTime(0), extreme));
+ }
+
+ @Override
+ public void onRejection(Window window) {
+ double extreme = 0;
+ double[] array = window.getDoubleArray();
+ for (int i = 0, n = window.size(); i < n; ++i) {
+ extreme = Math.max(extreme, Math.abs(array[i]));
+ }
+
+ try {
+ localIoTDBHandler.onEvent(new
LocalIoTDBEvent(window.getTime(0), extreme));
+ } catch (Exception e) {
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+ });
+ }
+
+ @Override
+ public void onDrop() throws Exception {
+ closeSinkHandlers();
+ }
+
+ @Override
+ public void onStart() throws Exception {
+ openSinkHandlers();
+ }
+
+ @Override
+ public void onStop() throws Exception {
+ closeSinkHandlers();
+ }
+
+ @Override
+ public Double fire(long timestamp, Double value) {
+ windowEvaluationHandler.collect(timestamp, value);
+ return value;
+ }
+
+ @Override
+ public double[] fire(long[] timestamps, double[] values) {
+ for (int i = 0, n = timestamps.length; i < n; ++i) {
+ windowEvaluationHandler.collect(timestamps[i], values[i]);
+ }
+ return values;
+ }
+
+ private void openSinkHandlers() throws Exception {
+ localIoTDBHandler.open(new LocalIoTDBConfiguration(device, measurements,
dataTypes));
+ }
+
+ private void closeSinkHandlers() throws Exception {
+ localIoTDBHandler.close();
+ }
+}
