Fix unit tests

.github/actions/etcdRunner/action.yml

@@ -0,0 +1,19 @@
+name: 'etcd-runner'
+description: 'etcd runner'
+
+runs:
+  using: "composite" # Mandatory parameter
+  steps:
+    - name: Start ETCD
+      shell: bash
+      run: |
+        if [ "$RUNNER_OS" == "Linux" ]; then
+          chmod +x "${GITHUB_WORKSPACE}/.github/etcd.sh"
+          "${GITHUB_WORKSPACE}/.github/etcd.sh"
+        elif [ "$RUNNER_OS" == "macOS" ]; then
+          chmod +x "${GITHUB_WORKSPACE}/.github/etcd_macos.sh"
+          "${GITHUB_WORKSPACE}/.github/etcd_macos.sh"
+        else
+          echo "$RUNNER_OS is not supported"
+          exit 1
+        fi

.github/etcd_macos.sh

@@ -0,0 +1,23 @@
+#!/bin/sh
+
+set -e
+
+sh -c "wget -nv https://golang.org/dl/go1.16.5.linux-amd64.tar.gz"
+
+sh -c "tar -zxvf go1.16.5.linux-amd64.tar.gz"
+
+export GOROOT=$PWD
+
+export GOPATH=$GOROOT/go
+
+export GOBIN=$GOPATH/bin
+
+export PATH=$PATH:$GOBIN
+
+sh -c "wget -nv https://github.com/etcd-io/etcd/releases/download/v3.5.0/etcd-v3.5.0-darwin-amd64.zip"
+
+sh -c "unzip etcd-v3.5.0-darwin-amd64.zip"
+
+sh -c "mv etcd-v3.5.0-darwin-amd64/etcd* $GOBIN"
+
+sh -c "nohup etcd >> run.log 2>&1 &"

.github/workflows/unit-test.yml

@@ -0,0 +1,39 @@
+name: "UTTest"
+
+on:
+  pull_request:
+    types: [opened, reopened]
+    branches:
+      - main
+env:
+  VERSION: 0.6.0-SNAPSHOT
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  Unit-Test:
+    strategy:
+      fail-fast: false
+      matrix:
+        java: [ 8 ]
+        python-version: [ "3.7" ]
+        os: [ ubuntu-latest, macos-latest ]
+    runs-on: ${{ matrix.os }}
+    steps:
+      - uses: actions/checkout@v2
+      - name: Environmet Dependence
+        uses: ./.github/actions/dependence
+        with:
+          python-version: ${{ matrix.python-version }}
+          java: ${{ matrix.java }}
+
+      - name: Run ZooKeeper
+        uses: ./.github/actions/zookeeperRunner
+
+      - name: Run ETCD
+        uses: ./.github/actions/etcdRunner
+
+      - name: Test with Maven
+        run: mvn clean package -pl core -am

core/src/main/java/cn/edu/tsinghua/iginx/engine/logical/utils/ExprUtils.java

@@ -336,7 +336,7 @@ private static void extractTimeRange(List<TimeRange> timeRanges, Filter f) {
                 break;
             case Or:
                 List<TimeRange> ranges = getTimeRangeFromOrFilter((OrFilter) f);
-                if (ranges != null && !ranges.isEmpty()) {
+                if (!ranges.isEmpty()) {
                     timeRanges.addAll(ranges);
                 }
                 break;
@@ -452,8 +452,6 @@ private static Filter setTrue(Filter filter, TimeSeriesRange interval) {
                     andChildren.set(i, childFilter);
                 }
                 return new AndFilter(andChildren);
-            case Key:
-                return filter;
             case Value:
                 String path = ((ValueFilter) filter).getPath();
                 if (interval.getStartTimeSeries() != null && interval.getStartTimeSeries().compareTo(path) > 0) {

core/src/main/java/cn/edu/tsinghua/iginx/engine/shared/operator/filter/NotFilter.java

@@ -56,7 +56,7 @@ public Filter copy() {
 
     @Override
     public String toString() {
-        return "!" + child.toString();
+        return "!(" + child.toString() + ")";
     }
 
     @Override

core/src/main/java/cn/edu/tsinghua/iginx/metadata/sync/protocol/etcd/ETCDSyncProtocolImpl.java

@@ -86,7 +86,7 @@ public ETCDSyncProtocolImpl(String category, Client client, ProposalListener lis
 
     private void registerProposalListener() {
         this.proposalWatcher = client.getWatchClient().watch(ByteSequence.from(String.format(PROTOCOL_PROPOSAL_CONTAINER_TEMPLATE, category).getBytes()),
-                WatchOption.newBuilder().withPrefix(ByteSequence.from(String.format(PROTOCOL_PROPOSAL_CONTAINER_TEMPLATE, category).getBytes())).build(),
+                WatchOption.newBuilder().withPrefix(ByteSequence.from(String.format(PROTOCOL_PROPOSAL_CONTAINER_TEMPLATE, category).getBytes())).withPrevKV(true).build(),
                 new Watch.Listener() {
 
                     @Override
@@ -187,10 +187,10 @@ public boolean startProposal(String key, SyncProposal syncProposal, VoteListener
         try {
             // lock proposal + category
             leaseId = client.getLeaseClient().grant(MAX_LOCK_TIME).get().getID();
-            client.getLockClient().lock(ByteSequence.from(lockPath.getBytes()), leaseId);
-
-            GetResponse response = client.getKVClient().get(ByteSequence.from(String.format(PROTOCOL_PROPOSAL_TEMPLATE, this.category, key).getBytes()),
-                    GetOption.newBuilder().withSortOrder(GetOption.SortOrder.DESCEND).withLimit(1L).build()).get();
+            client.getLockClient().lock(ByteSequence.from(lockPath.getBytes()), leaseId).get();
+            String proposalPrefix = String.format(PROTOCOL_PROPOSAL_TEMPLATE, this.category, key);
+            GetResponse response = client.getKVClient().get(ByteSequence.from(proposalPrefix.getBytes()),
+                    GetOption.newBuilder().withSortOrder(GetOption.SortOrder.DESCEND).withPrefix(ByteSequence.from(proposalPrefix.getBytes())).withLimit(1L).build()).get();
             if (response.getCount() >= 1L) {
                 long lastCreateTime = Long.parseLong(new String(response.getKvs().get(0).getKey().getBytes()).split("_")[1]);
                 if (lastCreateTime + MAX_NETWORK_LATENCY > createTime) {
@@ -204,7 +204,8 @@ public boolean startProposal(String key, SyncProposal syncProposal, VoteListener
             proposalLock.writeLock().unlock();
 
             syncProposal.setCreateTime(createTime);
-            client.getKVClient().put(ByteSequence.from((String.format(PROTOCOL_PROPOSAL_TEMPLATE, this.category, key) + PATH_SEPARATOR + "proposal_" + createTime).getBytes()), ByteSequence.from(JsonUtils.toJson(syncProposal))).get();
+            String proposalPath = String.format(PROTOCOL_PROPOSAL_TEMPLATE, this.category, key) + PATH_SEPARATOR + "proposal_" + createTime;
+            client.getKVClient().put(ByteSequence.from(proposalPath.getBytes()), ByteSequence.from(JsonUtils.toJson(syncProposal))).get();
             return true;
         } catch (Exception e) {
             logger.error("start proposal failure: ", e);
@@ -259,10 +260,11 @@ public void endProposal(String key, SyncProposal syncProposal) throws NetworkExc
         try {
             // lock proposal + category
             leaseId = client.getLeaseClient().grant(MAX_LOCK_TIME).get().getID();
-            client.getLockClient().lock(ByteSequence.from(lockPath.getBytes()), leaseId);
+            client.getLockClient().lock(ByteSequence.from(lockPath.getBytes()), leaseId).get();
 
-            GetResponse response = client.getKVClient().get(ByteSequence.from(String.format(PROTOCOL_PROPOSAL_TEMPLATE, this.category, key).getBytes()),
-                    GetOption.newBuilder().withSortOrder(GetOption.SortOrder.DESCEND).withLimit(1L).build()).get();
+            String proposalPrefix = String.format(PROTOCOL_PROPOSAL_TEMPLATE, this.category, key);
+            GetResponse response = client.getKVClient().get(ByteSequence.from(proposalPrefix.getBytes()),
+                    GetOption.newBuilder().withSortOrder(GetOption.SortOrder.DESCEND).withPrefix(ByteSequence.from(proposalPrefix.getBytes())).withLimit(1L).build()).get();
             if (response.getCount() == 0) {
                 throw new ExecutionException("can't find proposal for " + key);
             }

core/src/main/java/cn/edu/tsinghua/iginx/sql/IginXSqlVisitor.java

@@ -555,8 +555,8 @@ private Operator parseOperator(ExpressionContext ctx) {
     private void parseSpecialClause(SpecialClauseContext ctx, SelectStatement selectStatement) {
         if (ctx.downsampleWithLevelClause() != null) {
             // downsampleWithLevelClause = downsampleClause + aggregateWithLevelClause
-            parseDownsampleClause(ctx.downsampleClause(), selectStatement);
-            parseAggregateWithLevelClause(ctx.aggregateWithLevelClause().INT(), selectStatement);
+            parseDownsampleClause(ctx.downsampleWithLevelClause().downsampleClause(), selectStatement);
+            parseAggregateWithLevelClause(ctx.downsampleWithLevelClause().aggregateWithLevelClause().INT(), selectStatement);
         }
         if (ctx.downsampleClause() != null) {
             parseDownsampleClause(ctx.downsampleClause(), selectStatement);

core/src/test/java/cn/edu/tsinghua/iginx/engine/logical/utils/ExprUtilsTest.java

@@ -2,20 +2,20 @@
 
 import cn.edu.tsinghua.iginx.engine.shared.TimeRange;
 import cn.edu.tsinghua.iginx.engine.shared.operator.filter.*;
+import cn.edu.tsinghua.iginx.exceptions.SQLParserException;
 import cn.edu.tsinghua.iginx.metadata.entity.TimeSeriesInterval;
 import cn.edu.tsinghua.iginx.sql.TestUtils;
 import cn.edu.tsinghua.iginx.sql.statement.DeleteStatement;
 import cn.edu.tsinghua.iginx.sql.statement.SelectStatement;
 import org.junit.Test;
 
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.stream.Collectors;
 
 import static org.junit.Assert.assertEquals;
 
-class FilterTransformer {
+class InfluxDBFilterTransformer {
 
     public static String toString(Filter filter) {
         switch (filter.getType()) {
@@ -35,7 +35,7 @@ public static String toString(Filter filter) {
     }
 
     private static String toString(AndFilter filter) {
-        return filter.getChildren().stream().map(FilterTransformer::toString).collect(Collectors.joining(" and ", "(", ")"));
+        return filter.getChildren().stream().map(InfluxDBFilterTransformer::toString).collect(Collectors.joining(" and ", "(", ")"));
     }
 
     private static String toString(NotFilter filter) {
@@ -51,7 +51,7 @@ private static String toString(ValueFilter filter) {
     }
 
     private static String toString(OrFilter filter) {
-        return filter.getChildren().stream().map(FilterTransformer::toString).collect(Collectors.joining(" or ", "(", ")"));
+        return filter.getChildren().stream().map(InfluxDBFilterTransformer::toString).collect(Collectors.joining(" or ", "(", ")"));
     }
 
 
@@ -118,38 +118,38 @@ public void testToCNF() {
         SelectStatement statement = (SelectStatement) TestUtils.buildStatement(select);
         Filter filter = statement.getFilter();
         System.out.println(filter.toString());
-        System.out.println(FilterTransformer.toString(filter));
+        System.out.println(InfluxDBFilterTransformer.toString(filter));
         System.out.println(ExprUtils.toCNF(filter).toString());
-        System.out.println(FilterTransformer.toString(ExprUtils.toCNF(filter)));
+        System.out.println(InfluxDBFilterTransformer.toString(ExprUtils.toCNF(filter)));
 
         select = "SELECT a FROM root WHERE (a > 5 AND b <= 10) OR (c > 7 AND d == 8);";
         statement = (SelectStatement) TestUtils.buildStatement(select);
         filter = statement.getFilter();
         System.out.println(filter.toString());
-        System.out.println(FilterTransformer.toString(filter));
+        System.out.println(InfluxDBFilterTransformer.toString(filter));
         System.out.println(ExprUtils.toCNF(filter).toString());
-        System.out.println(FilterTransformer.toString(ExprUtils.toCNF(filter)));
+        System.out.println(InfluxDBFilterTransformer.toString(ExprUtils.toCNF(filter)));
 
         select = "SELECT a FROM root WHERE (a > 5 AND b <= 10) OR (c > 7 OR d == 8) OR (e < 3 AND f != 2);";
         statement = (SelectStatement) TestUtils.buildStatement(select);
         filter = statement.getFilter();
         System.out.println(filter.toString());
-        System.out.println(FilterTransformer.toString(filter));
+        System.out.println(InfluxDBFilterTransformer.toString(filter));
         System.out.println(ExprUtils.toCNF(filter).toString());
-        System.out.println(FilterTransformer.toString(ExprUtils.toCNF(filter)));
+        System.out.println(InfluxDBFilterTransformer.toString(ExprUtils.toCNF(filter)));
 
         select = "SELECT a FROM root WHERE (a > 5 OR b <= 10) OR (c > 7 AND d == 8);";
         statement = (SelectStatement) TestUtils.buildStatement(select);
         filter = statement.getFilter();
         System.out.println(filter.toString());
-        System.out.println(FilterTransformer.toString(filter));
+        System.out.println(InfluxDBFilterTransformer.toString(filter));
         System.out.println(ExprUtils.toCNF(filter).toString());
-        System.out.println(FilterTransformer.toString(ExprUtils.toCNF(filter)));
+        System.out.println(InfluxDBFilterTransformer.toString(ExprUtils.toCNF(filter)));
     }
 
     @Test
     public void testTimeRange() {
-        String delete = "DELETE FROM root.a WHERE (time > 5 AND time <= 10) OR (time > 12 AND time < 15);";
+        String delete = "DELETE FROM root.a WHERE (key > 5 AND key <= 10) OR (key > 12 AND key < 15);";
         DeleteStatement statement = (DeleteStatement) TestUtils.buildStatement(delete);
         assertEquals(
             Arrays.asList(
@@ -159,7 +159,7 @@ public void testTimeRange() {
             statement.getTimeRanges()
         );
 
-        delete = "DELETE FROM root.a WHERE (time > 1 AND time <= 8) OR (time >= 5 AND time < 11) OR time >= 66;";
+        delete = "DELETE FROM root.a WHERE (key > 1 AND key <= 8) OR (key >= 5 AND key < 11) OR key >= 66;";
         statement = (DeleteStatement) TestUtils.buildStatement(delete);
         assertEquals(
             Arrays.asList(
@@ -169,7 +169,7 @@ public void testTimeRange() {
             statement.getTimeRanges()
         );
 
-        delete = "DELETE FROM root.a WHERE time >= 16 AND time < 61;";
+        delete = "DELETE FROM root.a WHERE key >= 16 AND key < 61;";
         statement = (DeleteStatement) TestUtils.buildStatement(delete);
         assertEquals(
             Collections.singletonList(
@@ -178,7 +178,7 @@ public void testTimeRange() {
             statement.getTimeRanges()
         );
 
-        delete = "DELETE FROM root.a WHERE time >= 16;";
+        delete = "DELETE FROM root.a WHERE key >= 16;";
         statement = (DeleteStatement) TestUtils.buildStatement(delete);
         assertEquals(
             Collections.singletonList(
@@ -187,7 +187,7 @@ public void testTimeRange() {
             statement.getTimeRanges()
         );
 
-        delete = "DELETE FROM root.a WHERE time < 61;";
+        delete = "DELETE FROM root.a WHERE key < 61;";
         statement = (DeleteStatement) TestUtils.buildStatement(delete);
         assertEquals(
             Collections.singletonList(
@@ -196,46 +196,47 @@ public void testTimeRange() {
             statement.getTimeRanges()
         );
 
-        delete = "DELETE FROM root.a WHERE time < 61 AND time > 616;";
-        statement = (DeleteStatement) TestUtils.buildStatement(delete);
-        assertEquals(
-            new ArrayList<>(),
-            statement.getTimeRanges()
-        );
-
         delete = "DELETE FROM root.a;";
         statement = (DeleteStatement) TestUtils.buildStatement(delete);
         assertEquals(
-            new ArrayList<>(),
+            Collections.singletonList(
+                new TimeRange(0, Long.MAX_VALUE)
+            ),
             statement.getTimeRanges()
         );
     }
 
+    @Test(expected=SQLParserException.class)
+    public void testErrDelete() {
+        String delete = "DELETE FROM root.a WHERE key < 61 AND key > 616;";
+        TestUtils.buildStatement(delete);
+    }
+
     @Test
     public void testGetSubFilterFromFragment() {
         // sub1
-        String select = "SELECT a FROM root WHERE (a > 5 OR d < 15) AND !(e < 27) AND (c < 10 OR b > 2) AND time > 10 AND time <= 100;";
+        String select = "SELECT a FROM root WHERE (a > 5 OR d < 15) AND !(e < 27) AND (c < 10 OR b > 2) AND key > 10 AND key <= 100;";
         SelectStatement statement = (SelectStatement) TestUtils.buildStatement(select);
         Filter filter = statement.getFilter();
         assertEquals(
-            "((((root.a > 5) || (root.d < 15)) && !((root.e < 27)) && ((root.c < 10) || (root.b > 2)) && time > 10 && time <= 100))",
+            "((root.a > 5 || root.d < 15) && !(root.e < 27) && (root.c < 10 || root.b > 2) && key > 10 && key <= 100)",
             filter.toString()
         );
         assertEquals(
-            "((time > 10 && time <= 100))",
+            "(key > 10 && key <= 100)",
             ExprUtils.getSubFilterFromFragment(filter, new TimeSeriesInterval("root.a", "root.c")).toString()
         );
 
         // sub2
-        select = "SELECT a FROM root WHERE (a > 5 OR d < 15) AND !(e < 27) AND (c < 10 OR b > 2) AND time > 10 AND time <= 100;";
+        select = "SELECT a FROM root WHERE (a > 5 OR d < 15) AND !(e < 27) AND (c < 10 OR b > 2) AND key > 10 AND key <= 100;";
         statement = (SelectStatement) TestUtils.buildStatement(select);
         filter = statement.getFilter();
         assertEquals(
-            "((((root.a > 5) || (root.d < 15)) && !((root.e < 27)) && ((root.c < 10) || (root.b > 2)) && time > 10 && time <= 100))",
+            "((root.a > 5 || root.d < 15) && !(root.e < 27) && (root.c < 10 || root.b > 2) && key > 10 && key <= 100)",
             filter.toString()
         );
         assertEquals(
-            "(((root.e >= 27) && time > 10 && time <= 100))",
+            "(root.e >= 27 && key > 10 && key <= 100)",
             ExprUtils.getSubFilterFromFragment(filter, new TimeSeriesInterval("root.c", "root.z")).toString()
         );
 
@@ -244,11 +245,11 @@ public void testGetSubFilterFromFragment() {
         statement = (SelectStatement) TestUtils.buildStatement(select);
         filter = statement.getFilter();
         assertEquals(
-            "((((root.a > 5) || (root.d < 15)) && !((root.e < 27)) && ((root.c < 10) || (root.b > 2))))",
+            "((root.a > 5 || root.d < 15) && !(root.e < 27) && (root.c < 10 || root.b > 2))",
             filter.toString()
         );
         assertEquals(
-            "(((root.a > 5 || root.d < 15) && (root.e >= 27) && (root.c < 10 || root.b > 2)))",
+            "((root.a > 5 || root.d < 15) && root.e >= 27 && (root.c < 10 || root.b > 2))",
             ExprUtils.getSubFilterFromFragment(filter, new TimeSeriesInterval("root.a", "root.z")).toString()
         );
 
@@ -257,7 +258,7 @@ public void testGetSubFilterFromFragment() {
         statement = (SelectStatement) TestUtils.buildStatement(select);
         filter = statement.getFilter();
         assertEquals(
-            "((((root.a > 5) || (root.d < 15)) && !((root.e < 27)) && ((root.c < 10) || (root.b > 2))))",
+            "((root.a > 5 || root.d < 15) && !(root.e < 27) && (root.c < 10 || root.b > 2))",
             filter.toString()
         );
         assertEquals(