Serialize filters (#657)

* make Filter serializable
apache · Dec 24, 2019 · e7adcbb · e7adcbb
1 parent c1cb94b
commit e7adcbb
Show file tree

Hide file tree

Showing 19 changed files with 518 additions and 25 deletions.
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/EngineQueryRouter.java b/server/src/main/java/org/apache/iotdb/db/query/executor/EngineQueryRouter.java
@@ -111,7 +111,7 @@ public QueryDataSet groupBy(List<Path> selectedSeries, List<String> aggres,
 
     long queryId = context.getQueryId();
 
-    GlobalTimeExpression timeExpression = new GlobalTimeExpression(new GroupByFilter(unit, slidingStep, startTime, endTime, FilterType.GROUP_BY_FILTER));
+    GlobalTimeExpression timeExpression = new GlobalTimeExpression(new GroupByFilter(unit, slidingStep, startTime, endTime));
 
     if (expression == null) {
       expression = timeExpression;

diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/GroupByFilter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/GroupByFilter.java
@@ -18,27 +18,32 @@
  */
 package org.apache.iotdb.tsfile.read.filter;
 
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.util.Objects;
 import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
 import org.apache.iotdb.tsfile.read.filter.basic.Filter;
-import org.apache.iotdb.tsfile.read.filter.factory.FilterType;
-
-import java.io.Serializable;
+import org.apache.iotdb.tsfile.read.filter.factory.FilterSerializeId;
 
 public class GroupByFilter implements Filter, Serializable {
 
   private static final long serialVersionUID = -1211805021419281440L;
-  private final long unit;
-  private final long slidingStep;
-  private final long startTime;
-  private final long endTime;
-  private final FilterType filterType;
+  private long unit;
+  private long slidingStep;
+  private long startTime;
+  private long endTime;
 
-  public GroupByFilter(long unit, long slidingStep, long startTime, long endTime, FilterType filterType) {
+  public GroupByFilter(long unit, long slidingStep, long startTime, long endTime) {
     this.unit = unit;
     this.slidingStep = slidingStep;
     this.startTime = startTime;
     this.endTime = endTime;
-    this.filterType = filterType;
+  }
+
+  public GroupByFilter() {
+
   }
 
   @Override
@@ -91,11 +96,54 @@ public boolean containStartEndTime(long startTime, long endTime) {
 
   @Override
   public Filter clone() {
-    return new GroupByFilter(unit, slidingStep, startTime, endTime, filterType);
+    return new GroupByFilter(unit, slidingStep, startTime, endTime);
   }
 
   @Override
   public String toString() {
     return "GroupByFilter{}";
   }
+
+  @Override
+  public void serialize(DataOutputStream outputStream) {
+    try {
+      outputStream.write(getSerializeId().ordinal());
+      outputStream.writeLong(unit);
+      outputStream.writeLong(slidingStep);
+      outputStream.writeLong(startTime);
+      outputStream.writeLong(endTime);
+    } catch (IOException ignored) {
+      // ignored
+    }
+  }
+
+  @Override
+  public void deserialize(ByteBuffer buffer) {
+    unit = buffer.getLong();
+    slidingStep = buffer.getLong();
+    startTime = buffer.getLong();
+    endTime = buffer.getLong();
+  }
+
+  @Override
+  public FilterSerializeId getSerializeId() {
+    return FilterSerializeId.GROUP_BY;
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (!(obj instanceof GroupByFilter)) {
+      return false;
+    }
+    GroupByFilter other = ((GroupByFilter) obj);
+    return this.unit == other.unit
+        && this.slidingStep == other.slidingStep
+        && this.startTime == other.startTime
+        && this.endTime == other.endTime;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(unit, slidingStep, startTime, endTime);
+  }
 }
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/basic/BinaryFilter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/basic/BinaryFilter.java
@@ -18,7 +18,12 @@
  */
 package org.apache.iotdb.tsfile.read.filter.basic;
 
+import java.io.DataOutputStream;
+import java.io.IOException;
 import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.util.Objects;
+import org.apache.iotdb.tsfile.read.filter.factory.FilterFactory;
 
 /**
  * Definition for binary filter operations.
@@ -27,8 +32,11 @@ public abstract class BinaryFilter implements Filter, Serializable {
 
   private static final long serialVersionUID = 1039585564327602465L;
 
-  protected final Filter left;
-  protected final Filter right;
+  protected Filter left;
+  protected Filter right;
+
+  public BinaryFilter() {
+  }
 
   protected BinaryFilter(Filter left, Filter right) {
     this.left = left;
@@ -50,4 +58,36 @@ public String toString() {
 
   @Override
   public abstract Filter clone();
+
+  @Override
+  public void serialize(DataOutputStream outputStream) {
+    try {
+      outputStream.write(getSerializeId().ordinal());
+      left.serialize(outputStream);
+      right.serialize(outputStream);
+    } catch (IOException ignored) {
+      // ignore
+    }
+  }
+
+  @Override
+  public void deserialize(ByteBuffer buffer) {
+    left = FilterFactory.deserialize(buffer);
+    right = FilterFactory.deserialize(buffer);
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (!(obj instanceof BinaryFilter)) {
+      return false;
+    }
+    BinaryFilter other = ((BinaryFilter) obj);
+    return this.left.equals(other.left) && this.right.equals(other.right)
+        && this.getSerializeId().equals(other.getSerializeId());
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(left, right, getSerializeId());
+  }
 }
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/basic/Filter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/basic/Filter.java
@@ -18,7 +18,10 @@
  */
 package org.apache.iotdb.tsfile.read.filter.basic;
 
+import java.io.DataOutputStream;
+import java.nio.ByteBuffer;
 import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
+import org.apache.iotdb.tsfile.read.filter.factory.FilterSerializeId;
 
 /**
  * Filter is a top level filter abstraction.
@@ -63,4 +66,10 @@ public interface Filter {
   boolean containStartEndTime(long startTime, long endTime);
 
   Filter clone();
+
+  void serialize(DataOutputStream outputStream);
+
+  void deserialize(ByteBuffer buffer);
+
+  FilterSerializeId getSerializeId();
 }
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/basic/UnaryFilter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/basic/UnaryFilter.java
@@ -18,8 +18,14 @@
  */
 package org.apache.iotdb.tsfile.read.filter.basic;
 
+import java.io.DataOutputStream;
+import java.io.IOException;
 import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.util.Objects;
 import org.apache.iotdb.tsfile.read.filter.factory.FilterType;
+import org.apache.iotdb.tsfile.read.filter.operator.Eq;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
 
 /**
  * Definition for unary filter operations.
@@ -29,10 +35,13 @@
 public abstract class UnaryFilter<T extends Comparable<T>> implements Filter, Serializable {
 
   private static final long serialVersionUID = 1431606024929453556L;
-  protected final T value;
+  protected T value;
 
   protected FilterType filterType;
 
+  public UnaryFilter() {
+  }
+
   protected UnaryFilter(T value, FilterType filterType) {
     this.value = value;
     this.filterType = filterType;
@@ -51,4 +60,36 @@ public FilterType getFilterType() {
 
   @Override
   public abstract Filter clone();
+
+  @Override
+  public void serialize(DataOutputStream outputStream) {
+    try {
+      outputStream.write(getSerializeId().ordinal());
+      outputStream.write(filterType.ordinal());
+      ReadWriteIOUtils.writeObject(value, outputStream);
+    } catch (IOException ignored) {
+      // ignored
+    }
+  }
+
+  @Override
+  public void deserialize(ByteBuffer buffer) {
+    filterType = FilterType.values()[buffer.get()];
+    value = (T) ReadWriteIOUtils.readObject(buffer);
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (!(obj instanceof UnaryFilter)) {
+      return false;
+    }
+    UnaryFilter other = ((UnaryFilter) obj);
+    return this.value.equals(other.value) && this.filterType.equals(other.filterType)
+        && this.getSerializeId().equals(other.getSerializeId());
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(value, filterType, getSerializeId());
+  }
 }
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/factory/FilterFactory.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/factory/FilterFactory.java
@@ -18,10 +18,20 @@
  */
 package org.apache.iotdb.tsfile.read.filter.factory;
 
+import java.nio.ByteBuffer;
+import org.apache.iotdb.tsfile.read.filter.GroupByFilter;
 import org.apache.iotdb.tsfile.read.filter.basic.Filter;
 import org.apache.iotdb.tsfile.read.filter.operator.AndFilter;
+import org.apache.iotdb.tsfile.read.filter.operator.Eq;
+import org.apache.iotdb.tsfile.read.filter.operator.Gt;
+import org.apache.iotdb.tsfile.read.filter.operator.GtEq;
+import org.apache.iotdb.tsfile.read.filter.operator.Lt;
+import org.apache.iotdb.tsfile.read.filter.operator.LtEq;
+import org.apache.iotdb.tsfile.read.filter.operator.NotEq;
 import org.apache.iotdb.tsfile.read.filter.operator.NotFilter;
 import org.apache.iotdb.tsfile.read.filter.operator.OrFilter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class FilterFactory {
 
@@ -37,4 +47,46 @@ public static NotFilter not(Filter filter) {
     return new NotFilter(filter);
   }
 
+  public static Filter deserialize(ByteBuffer buffer) {
+    FilterSerializeId id = FilterSerializeId.values()[buffer.get()];
+
+    Filter filter;
+    switch (id) {
+      case EQ:
+        filter = new Eq<>();
+        break;
+      case GT:
+        filter = new Gt<>();
+        break;
+      case LT:
+        filter = new Lt<>();
+        break;
+      case OR:
+        filter = new OrFilter();
+        break;
+      case AND:
+        filter = new AndFilter();
+        break;
+      case NEQ:
+        filter = new NotEq<>();
+        break;
+      case NOT:
+        filter = new NotFilter();
+        break;
+      case GTEQ:
+        filter = new GtEq<>();
+        break;
+      case LTEQ:
+        filter = new LtEq<>();
+        break;
+      case GROUP_BY:
+        filter = new GroupByFilter();
+        break;
+      default:
+        throw new UnsupportedOperationException("Unknown filter type " + id);
+    }
+    filter.deserialize(buffer);
+    return filter;
+  }
+
 }
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/factory/FilterSerializeId.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/factory/FilterSerializeId.java
@@ -0,0 +1,24 @@
+/*
+ * 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.tsfile.read.filter.factory;
+
+public enum FilterSerializeId {
+  AND, EQ, GROUP_BY, GT, GTEQ, LT, LTEQ, NEQ, NOT, OR
+}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/AndFilter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/filter/operator/AndFilter.java
@@ -21,6 +21,7 @@
 import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
 import org.apache.iotdb.tsfile.read.filter.basic.BinaryFilter;
 import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.filter.factory.FilterSerializeId;
 
 /**
  * Both the left and right operators of AndExpression must satisfy the condition.
@@ -29,6 +30,9 @@ public class AndFilter extends BinaryFilter {
 
   private static final long serialVersionUID = -8212850098906044102L;
 
+  public AndFilter() {
+  }
+
   public AndFilter(Filter left, Filter right) {
     super(left, right);
   }
@@ -64,4 +68,9 @@ public String toString() {
   public Filter clone() {
     return new AndFilter(left.clone(), right.clone());
   }
+
+  @Override
+  public FilterSerializeId getSerializeId() {
+    return FilterSerializeId.AND;
+  }
 }