[iotdb] branch xingtanzjr/operator-design updated: complete DeviceMergeOperator and TimeJoinOperator

[xi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

xingtanzjr pushed a commit to branch xingtanzjr/operator-design
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/xingtanzjr/operator-design by this push:
     new cd8ac4d  complete DeviceMergeOperator and TimeJoinOperator
cd8ac4d is described below

commit cd8ac4d0af573d374a242d5562f77a86a1f003f9
Author: Jinrui.Zhang <xi...@gmail.com>
AuthorDate: Tue Mar 1 21:36:12 2022 +0800

    complete DeviceMergeOperator and TimeJoinOperator
---
 .../distribution/common/DeviceMergeOrder.java      |  8 +++++
 .../iotdb/db/query/distribution/common/Tablet.java | 29 ++++++++++++++++
 .../query/distribution/common/TabletMetadata.java  | 21 ++++++++++++
 .../query/distribution/common/WithoutPolicy.java   |  6 ++++
 .../distribution/operator/DeviceMergeOperator.java | 40 ++++++++++++++++++++++
 .../distribution/operator/TimeJoinOperator.java    | 16 +++++++--
 6 files changed, 117 insertions(+), 3 deletions(-)

diff --git a/server/src/main/java/org/apache/iotdb/db/query/distribution/common/DeviceMergeOrder.java b/server/src/main/java/org/apache/iotdb/db/query/distribution/common/DeviceMergeOrder.java
new file mode 100644
index 0000000..986beac
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/distribution/common/DeviceMergeOrder.java
@@ -0,0 +1,8 @@
+package org.apache.iotdb.db.query.distribution.common;
+
+public enum DeviceMergeOrder {
+    TIMESTAMP_ASC,
+    TIMESTAMP_DESC,
+    DEVICE_NAME_ASC,
+    DEVICE_NAME_DESC,
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/distribution/common/Tablet.java b/server/src/main/java/org/apache/iotdb/db/query/distribution/common/Tablet.java
new file mode 100644
index 0000000..956ccb0
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/distribution/common/Tablet.java
@@ -0,0 +1,29 @@
+package org.apache.iotdb.db.query.distribution.common;
+
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+
+/**
+ * Intermediate result for most of ExecOperators.
+ * The Tablet contains data from one or more series and constructs them as a row based view
+ * The Tablet constructed with n series has n+1 columns where one column is timestamp and the other n columns are values
+ * from each series.
+ * The Tablet also contains the metadata of owned series.
+ *
+ * TODO: consider the detailed data store model in memory. (using column based or row based ?)
+ */
+public class Tablet {
+
+    private TabletMetadata metadata;
+
+    public boolean hasNext() {
+        return false;
+    }
+
+    public RowRecord getNext() {
+        return null;
+    }
+
+    public TabletMetadata getMetadata() {
+        return metadata;
+    }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/distribution/common/TabletMetadata.java b/server/src/main/java/org/apache/iotdb/db/query/distribution/common/TabletMetadata.java
new file mode 100644
index 0000000..0a01959
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/distribution/common/TabletMetadata.java
@@ -0,0 +1,21 @@
+package org.apache.iotdb.db.query.distribution.common;
+
+import org.apache.iotdb.tsfile.read.common.Path;
+
+import java.util.List;
+
+public class TabletMetadata {
+    // list of all columns in current Tablet
+    // The column list not only contains the series column, but also contains other column to construct the final result
+    // set such as timestamp and deviceName
+    private List<String> columnList;
+
+    // Indicate whether the result set should be aligned by device. This parameter can be used for downstream operators
+    // when processing data from current Tablet. The RowRecord produced by Tablet with `alignedByDevice = true` will contain
+    // n + 1 fields which are n series field and 1 deviceName field.
+    // For example, when the FilterOperator execute the filter operation, it may need the deviceName field when matching
+    // the series with corresponding column in Tablet
+    //
+    // If alignedByDevice is true, the owned series should belong to one device
+    private boolean alignedByDevice;
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/distribution/common/WithoutPolicy.java b/server/src/main/java/org/apache/iotdb/db/query/distribution/common/WithoutPolicy.java
new file mode 100644
index 0000000..02d1e43
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/distribution/common/WithoutPolicy.java
@@ -0,0 +1,6 @@
+package org.apache.iotdb.db.query.distribution.common;
+
+public enum WithoutPolicy {
+    CONTAINS_NULL,
+    ALL_NULL
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/distribution/operator/DeviceMergeOperator.java b/server/src/main/java/org/apache/iotdb/db/query/distribution/operator/DeviceMergeOperator.java
new file mode 100644
index 0000000..16099ca
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/distribution/operator/DeviceMergeOperator.java
@@ -0,0 +1,40 @@
+package org.apache.iotdb.db.query.distribution.operator;
+
+import org.apache.iotdb.db.query.distribution.common.DeviceMergeOrder;
+import org.apache.iotdb.db.query.distribution.common.Tablet;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * DeviceMergeOperator is responsible for constructing a device-based view of a set of series. And output the result with
+ * specific order. The order could be 'order by device' or 'order by timestamp'
+ *
+ * The types of involved devices should be same. If the device contains n series, the device-based view will contain n+2
+ * columns, which are timestamp column, device name column and n value columns of involved series.
+ *
+ * Children type: [TimeJoinOperator]
+ */
+public class DeviceMergeOperator extends ExecOperator<Tablet> {
+    // The result output order that this operator
+    private DeviceMergeOrder mergeOrder;
+
+    // Owned devices
+    private List<String> ownedDeviceNameList;
+
+    // The map from deviceName to corresponding query result operator responsible for that device.
+    private Map<String, TimeJoinOperator> upstreamMap;
+
+    @Override
+    public boolean hasNext() {
+        return false;
+    }
+
+    // If the Tablet from TimeJoinOperator has n columns, the output of DeviceMergeOperator will contain n+1 columns where
+    // the additional column is `deviceName`
+    // And, the `alignedByDevice` in the TabletMetadata will be `true`
+    @Override
+    public Tablet getNextBatch() {
+        return null;
+    }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/distribution/operator/TimeJoinOperator.java b/server/src/main/java/org/apache/iotdb/db/query/distribution/operator/TimeJoinOperator.java
index 8361276..dbf1deb 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/distribution/operator/TimeJoinOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/distribution/operator/TimeJoinOperator.java
@@ -1,6 +1,8 @@
 package org.apache.iotdb.db.query.distribution.operator;
 
-import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.db.query.distribution.common.Tablet;
+import org.apache.iotdb.db.query.distribution.common.TraversalOrder;
+import org.apache.iotdb.db.query.distribution.common.WithoutPolicy;
 
 /**
  * TimeJoinOperator is responsible for join two or more series.
@@ -9,7 +11,15 @@ import org.apache.iotdb.tsfile.read.common.RowRecord;
  *
  * Children type: [SeriesScanOperator]
  */
-public class TimeJoinOperator extends ExecOperator<RowRecord> {
+public class TimeJoinOperator extends ExecOperator<Tablet> {
+
+    // This parameter indicates the order when executing multiway merge sort.
+    private TraversalOrder mergeOrder;
+
+    // The policy to decide whether a row should be discarded
+    // The without policy is able to be push down to the TimeJoinOperator because we can know whether a row contains
+    // null or not in this operator the situation won't be changed by the downstream operators.
+    private WithoutPolicy withoutPolicy;
 
     @Override
     public boolean hasNext() {
@@ -17,7 +27,7 @@ public class TimeJoinOperator extends ExecOperator<RowRecord> {
     }
 
     @Override
-    public RowRecord getNextBatch() {
+    public Tablet getNextBatch() {
         return null;
     }
 }