Add bit and bool aggregate functions

geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/schema/function/BuildInSqlFunctionTable.java

@@ -52,6 +52,9 @@
 import org.apache.geaflow.dsl.udf.graph.SingleSourceShortestPath;
 import org.apache.geaflow.dsl.udf.graph.TriangleCount;
 import org.apache.geaflow.dsl.udf.graph.WeakConnectedComponents;
+import org.apache.geaflow.dsl.udf.table.agg.BitAndInteger;
+import org.apache.geaflow.dsl.udf.table.agg.BitOrInteger;
+import org.apache.geaflow.dsl.udf.table.agg.BoolAnd;
 import org.apache.geaflow.dsl.udf.table.agg.PercentileDouble;
 import org.apache.geaflow.dsl.udf.table.agg.PercentileInteger;
 import org.apache.geaflow.dsl.udf.table.agg.PercentileLong;
@@ -222,6 +225,9 @@ public class BuildInSqlFunctionTable extends ListSqlOperatorTable {
             .add(GeaFlowFunction.of(PercentileLong.class))
             .add(GeaFlowFunction.of(PercentileInteger.class))
             .add(GeaFlowFunction.of(PercentileDouble.class))
+            .add(GeaFlowFunction.of(BitAndInteger.class))
+            .add(GeaFlowFunction.of(BitOrInteger.class))
+            .add(GeaFlowFunction.of(BoolAnd.class))
             // UDGA
             .add(GeaFlowFunction.of(SingleSourceShortestPath.class))
             .add(GeaFlowFunction.of(AllSourceShortestPath.class))

geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/table/agg/BitAndInteger.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.geaflow.dsl.udf.table.agg;
+
+import java.io.Serializable;
+import org.apache.geaflow.dsl.common.function.Description;
+import org.apache.geaflow.dsl.common.function.UDAF;
+import org.apache.geaflow.dsl.udf.table.agg.BitAndInteger.Accumulator;
+
+@Description(name = "bit_and", description = "The bitwise AND aggregate function for int.")
+public class BitAndInteger extends UDAF<Integer, Accumulator, Integer> {
+
+    @Override
+    public Accumulator createAccumulator() {
+        return new Accumulator(null);
+    }
+
+    @Override
+    public void accumulate(Accumulator accumulator, Integer input) {
+        if (input != null) {
+            accumulator.value = accumulator.value == null ? input : accumulator.value & input;
+        }
+    }
+
+    @Override
+    public void merge(Accumulator accumulator, Iterable<Accumulator> its) {
+        for (Accumulator toMerge : its) {
+            if (toMerge.value != null) {
+                accumulate(accumulator, toMerge.value);
+            }
+        }
+    }
+
+    @Override
+    public void resetAccumulator(Accumulator accumulator) {
+        accumulator.value = null;
+    }
+
+    @Override
+    public Integer getValue(Accumulator accumulator) {
+        return accumulator.value;
+    }
+
+    public static class Accumulator implements Serializable {
+
+        public Accumulator() {
+        }
+
+        public Integer value;
+
+        public Accumulator(Integer value) {
+            this.value = value;
+        }
+    }
+}

geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/table/agg/BitOrInteger.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.geaflow.dsl.udf.table.agg;
+
+import java.io.Serializable;
+import org.apache.geaflow.dsl.common.function.Description;
+import org.apache.geaflow.dsl.common.function.UDAF;
+import org.apache.geaflow.dsl.udf.table.agg.BitOrInteger.Accumulator;
+
+@Description(name = "bit_or", description = "The bitwise OR aggregate function for int.")
+public class BitOrInteger extends UDAF<Integer, Accumulator, Integer> {
+
+    @Override
+    public Accumulator createAccumulator() {
+        return new Accumulator(null);
+    }
+
+    @Override
+    public void accumulate(Accumulator accumulator, Integer input) {
+        if (input != null) {
+            accumulator.value = accumulator.value == null ? input : accumulator.value | input;
+        }
+    }
+
+    @Override
+    public void merge(Accumulator accumulator, Iterable<Accumulator> its) {
+        for (Accumulator toMerge : its) {
+            if (toMerge.value != null) {
+                accumulate(accumulator, toMerge.value);
+            }
+        }
+    }
+
+    @Override
+    public void resetAccumulator(Accumulator accumulator) {
+        accumulator.value = null;
+    }
+
+    @Override
+    public Integer getValue(Accumulator accumulator) {
+        return accumulator.value;
+    }
+
+    public static class Accumulator implements Serializable {
+
+        public Accumulator() {
+        }
+
+        public Integer value;
+
+        public Accumulator(Integer value) {
+            this.value = value;
+        }
+    }
+}

geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/table/agg/BoolAnd.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.geaflow.dsl.udf.table.agg;
+
+import java.io.Serializable;
+import org.apache.geaflow.dsl.common.function.Description;
+import org.apache.geaflow.dsl.common.function.UDAF;
+import org.apache.geaflow.dsl.udf.table.agg.BoolAnd.Accumulator;
+
+@Description(name = "bool_and", description = "The boolean AND aggregate function.")
+public class BoolAnd extends UDAF<Boolean, Accumulator, Boolean> {
+
+    @Override
+    public Accumulator createAccumulator() {
+        return new Accumulator(null);
+    }
+
+    @Override
+    public void accumulate(Accumulator accumulator, Boolean input) {
+        if (input != null) {
+            accumulator.value = accumulator.value == null ? input : accumulator.value && input;
+        }
+    }
+
+    @Override
+    public void merge(Accumulator accumulator, Iterable<Accumulator> its) {
+        for (Accumulator toMerge : its) {
+            if (toMerge.value != null) {
+                accumulate(accumulator, toMerge.value);
+            }
+        }
+    }
+
+    @Override
+    public void resetAccumulator(Accumulator accumulator) {
+        accumulator.value = null;
+    }
+
+    @Override
+    public Boolean getValue(Accumulator accumulator) {
+        return accumulator.value;
+    }
+
+    public static class Accumulator implements Serializable {
+
+        public Accumulator() {
+        }
+
+        public Boolean value;
+
+        public Accumulator(Boolean value) {
+            this.value = value;
+        }
+    }
+}

geaflow/geaflow-dsl/geaflow-dsl-plan/src/test/java/org/apache/geaflow/dsl/udf/agg/UDAFTest.java

@@ -23,6 +23,9 @@
 import org.apache.geaflow.dsl.udf.table.agg.AvgDouble;
 import org.apache.geaflow.dsl.udf.table.agg.AvgInteger;
 import org.apache.geaflow.dsl.udf.table.agg.AvgLong;
+import org.apache.geaflow.dsl.udf.table.agg.BitAndInteger;
+import org.apache.geaflow.dsl.udf.table.agg.BitOrInteger;
+import org.apache.geaflow.dsl.udf.table.agg.BoolAnd;
 import org.apache.geaflow.dsl.udf.table.agg.Count;
 import org.apache.geaflow.dsl.udf.table.agg.MaxDouble;
 import org.apache.geaflow.dsl.udf.table.agg.MaxInteger;
@@ -86,6 +89,62 @@ public void testCount() {
         Assert.assertEquals((long) af.getValue(accumulator), 0);
     }
 
+    @Test
+    public void testBitAndInteger() {
+        BitAndInteger af = new BitAndInteger();
+        BitAndInteger.Accumulator accumulator = af.createAccumulator();
+        af.accumulate(accumulator, null);
+        Assert.assertNull(af.getValue(accumulator));
+        af.accumulate(accumulator, 7);
+        af.accumulate(accumulator, 3);
+        Assert.assertEquals((int) af.getValue(accumulator), 3);
+
+        BitAndInteger.Accumulator toMerge = af.createAccumulator();
+        af.accumulate(toMerge, 1);
+        af.merge(accumulator, Lists.newArrayList(toMerge));
+        Assert.assertEquals((int) af.getValue(accumulator), 1);
+
+        af.resetAccumulator(accumulator);
+        Assert.assertNull(af.getValue(accumulator));
+    }
+
+    @Test
+    public void testBitOrInteger() {
+        BitOrInteger af = new BitOrInteger();
+        BitOrInteger.Accumulator accumulator = af.createAccumulator();
+        af.accumulate(accumulator, null);
+        Assert.assertNull(af.getValue(accumulator));
+        af.accumulate(accumulator, 4);
+        af.accumulate(accumulator, 2);
+        Assert.assertEquals((int) af.getValue(accumulator), 6);
+
+        BitOrInteger.Accumulator toMerge = af.createAccumulator();
+        af.accumulate(toMerge, 1);
+        af.merge(accumulator, Lists.newArrayList(toMerge));
+        Assert.assertEquals((int) af.getValue(accumulator), 7);
+
+        af.resetAccumulator(accumulator);
+        Assert.assertNull(af.getValue(accumulator));
+    }
+
+    @Test
+    public void testBoolAnd() {
+        BoolAnd af = new BoolAnd();
+        BoolAnd.Accumulator accumulator = af.createAccumulator();
+        af.accumulate(accumulator, null);
+        Assert.assertNull(af.getValue(accumulator));
+        af.accumulate(accumulator, true);
+        Assert.assertTrue(af.getValue(accumulator));
+
+        BoolAnd.Accumulator toMerge = af.createAccumulator();
+        af.accumulate(toMerge, false);
+        af.merge(accumulator, Lists.newArrayList(toMerge));
+        Assert.assertFalse(af.getValue(accumulator));
+
+        af.resetAccumulator(accumulator);
+        Assert.assertNull(af.getValue(accumulator));
+    }
+
     @Test
     public void testMaxDouble() {
         MaxDouble af = new MaxDouble();

geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/plan/PhysicAggregateRelNode.java

@@ -66,6 +66,9 @@
 import org.apache.geaflow.dsl.udf.table.agg.AvgDouble;
 import org.apache.geaflow.dsl.udf.table.agg.AvgInteger;
 import org.apache.geaflow.dsl.udf.table.agg.AvgLong;
+import org.apache.geaflow.dsl.udf.table.agg.BitAndInteger;
+import org.apache.geaflow.dsl.udf.table.agg.BitOrInteger;
+import org.apache.geaflow.dsl.udf.table.agg.BoolAnd;
 import org.apache.geaflow.dsl.udf.table.agg.Count;
 import org.apache.geaflow.dsl.udf.table.agg.MaxBinaryString;
 import org.apache.geaflow.dsl.udf.table.agg.MaxDouble;
@@ -102,6 +105,12 @@ public class PhysicAggregateRelNode extends Aggregate implements PhysicRelNode<R
 
     public static final String UDAF_PERCENTILE = "PERCENTILE";
 
+    public static final String UDAF_BIT_AND = "BIT_AND";
+
+    public static final String UDAF_BIT_OR = "BIT_OR";
+
+    public static final String UDAF_BOOL_AND = "BOOL_AND";
+
 
     public PhysicAggregateRelNode(
         RelOptCluster cluster,
@@ -456,6 +465,15 @@ private List<AggFunctionCall> buildAggFunctionCalls() {
                     aggClasses.add(PercentileInteger.class);
                     aggClasses.add(PercentileDouble.class);
                     break;
+                case UDAF_BIT_AND:
+                    aggClasses.add(BitAndInteger.class);
+                    break;
+                case UDAF_BIT_OR:
+                    aggClasses.add(BitOrInteger.class);
+                    break;
+                case UDAF_BOOL_AND:
+                    aggClasses.add(BoolAnd.class);
+                    break;
                 default:
                     throw new GeaFlowDSLException("Not support aggregate function " + aggName);
             }

geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/AggregateTest.java

@@ -131,6 +131,15 @@ public void testAggregate_012() throws Exception {
             .checkSinkResult();
     }
 
+    @Test
+    public void testAggregate_013() throws Exception {
+        QueryTester
+            .build()
+            .withQueryPath("/query/aggregate_013.sql")
+            .execute()
+            .checkSinkResult();
+    }
+
     @Test
     public void testStreamAggregate_001() throws Exception {
         QueryTester

geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/expect/aggregate_013.txt

@@ -0,0 +1 @@
+0,63,false