Merge branch 'master' into sub-iot-v2

Author: VGalaxies

LICENSE

@@ -270,3 +270,18 @@ The following files include code modified from Apache Kafka project.
 
 Project page: https://github.com/apache/kafka
 License: https://github.com/apache/kafka/blob/trunk/LICENSE
+
+--------------------------------------------------------------------------------
+
+The following files include code modified from Stream-Lib project.
+
+./iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/approximate/Counter.java
+./iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/approximate/DoublyLinkedList.java
+./iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/approximate/ExternalizableUtil.java
+./iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/approximate/ITopK.java
+./iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/approximate/ListNode2.java
+./iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/approximate/Pair.java
+./iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/approximate/StreamSummary.java
+
+Project page: https://github.com/addthis/stream-lib
+License: https://github.com/addthis/stream-lib/blob/master/LICENSE.txt

integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationIT.java

@@ -4165,6 +4165,51 @@ public void approxCountDistinctTest() {
         DATABASE_NAME);
   }
 
+  @Test
+  public void approxMostFrequentTest() {
+    String[] expectedHeader = buildHeaders(7);
+    String[] retArray =
+        new String[] {
+          "{\"50000\":8},{\"30.0\":8},{\"55.0\":8},{\"true\":12},{\"0xcafebabe55\":8},{\"1727158540000\":12},{\"20240924\":20},"
+        };
+    tableResultSetEqualTest(
+        "select approx_most_frequent(s2, 1, 10), approx_most_frequent(s3, 1, 10), approx_most_frequent(s4, 1, 10), approx_most_frequent(s5, 1, 10), approx_most_frequent(s8, 1, 10), approx_most_frequent(s9, 1, 10), approx_most_frequent(s10, 1, 10) from table1",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+
+    expectedHeader = new String[] {"time", "province", "_col2"};
+    retArray =
+        new String[] {
+          "2024-09-24T06:15:30.000Z,beijing,{},",
+          "2024-09-24T06:15:31.000Z,beijing,{\"31000\":2},",
+          "2024-09-24T06:15:35.000Z,beijing,{\"35000\":2},",
+          "2024-09-24T06:15:36.000Z,beijing,{},",
+          "2024-09-24T06:15:40.000Z,beijing,{\"40000\":2},",
+          "2024-09-24T06:15:41.000Z,beijing,{},",
+          "2024-09-24T06:15:46.000Z,beijing,{\"46000\":2},",
+          "2024-09-24T06:15:50.000Z,beijing,{\"50000\":4},",
+          "2024-09-24T06:15:51.000Z,beijing,{},",
+          "2024-09-24T06:15:55.000Z,beijing,{},",
+          "2024-09-24T06:15:30.000Z,shanghai,{},",
+          "2024-09-24T06:15:31.000Z,shanghai,{\"31000\":2},",
+          "2024-09-24T06:15:35.000Z,shanghai,{\"35000\":2},",
+          "2024-09-24T06:15:36.000Z,shanghai,{},",
+          "2024-09-24T06:15:40.000Z,shanghai,{\"40000\":2},",
+          "2024-09-24T06:15:41.000Z,shanghai,{},",
+          "2024-09-24T06:15:46.000Z,shanghai,{\"46000\":2},",
+          "2024-09-24T06:15:50.000Z,shanghai,{\"50000\":4},",
+          "2024-09-24T06:15:51.000Z,shanghai,{},",
+          "2024-09-24T06:15:55.000Z,shanghai,{},",
+        };
+
+    tableResultSetEqualTest(
+        "SELECT time,province,approx_most_frequent(s2, 1, 10) from table1 group by 1,2 order by 2,1",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+  }
+
   @Test
   public void exceptionTest() {
     tableAssertTestFail(
@@ -4211,6 +4256,14 @@ public void exceptionTest() {
         "select approx_count_distinct(province, 'test') from table1",
         "701: Second argument of Aggregate functions [approx_count_distinct] should be numberic type and do not use expression",
         DATABASE_NAME);
+    tableAssertTestFail(
+        "select approx_most_frequent(province, -10, 100) from table1",
+        "701: The second and third argument must be greater than 0, but got k=-10, capacity=100",
+        DATABASE_NAME);
+    tableAssertTestFail(
+        "select approx_most_frequent(province, 'test', 100) from table1",
+        "701: The second and third argument of 'approx_most_frequent' function must be numeric literal",
+        DATABASE_NAME);
   }
 
   // ==================================================================

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AbstractApproxMostFrequentAccumulator.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed 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.queryengine.execution.operator.source.relational.aggregation;
+
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.approximate.SpaceSavingStateFactory;
+
+import com.google.gson.Gson;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.file.metadata.statistics.Statistics;
+import org.apache.tsfile.utils.Binary;
+
+import java.nio.charset.StandardCharsets;
+
+public abstract class AbstractApproxMostFrequentAccumulator<T> implements TableAccumulator {
+
+  protected SpaceSavingStateFactory.SingleSpaceSavingState<T> state =
+      SpaceSavingStateFactory.createSingleState();
+
+  @Override
+  public void evaluateIntermediate(ColumnBuilder columnBuilder) {
+    columnBuilder.writeBinary(new Binary(state.getSpaceSaving().serialize()));
+  }
+
+  @Override
+  public void evaluateFinal(ColumnBuilder columnBuilder) {
+    columnBuilder.writeBinary(
+        new Binary(new Gson().toJson(state.getSpaceSaving().getBuckets()), StandardCharsets.UTF_8));
+  }
+
+  @Override
+  public boolean hasFinalResult() {
+    return false;
+  }
+
+  @Override
+  public void addStatistics(Statistics[] statistics) {
+    throw new UnsupportedOperationException(
+        "ApproxMostFrequentAccumulator does not support statistics");
+  }
+
+  @Override
+  public void reset() {
+    state.getSpaceSaving().reset();
+  }
+}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AccumulatorFactory.java

@@ -22,6 +22,11 @@
 import org.apache.iotdb.common.rpc.thrift.TAggregationType;
 import org.apache.iotdb.commons.udf.utils.UDFDataTypeTransformer;
 import org.apache.iotdb.db.queryengine.execution.aggregation.VarianceAccumulator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.BinaryGroupedApproxMostFrequentAccumulator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.BlobGroupedApproxMostFrequentAccumulator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.BooleanGroupedApproxMostFrequentAccumulator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.DoubleGroupedApproxMostFrequentAccumulator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.FloatGroupedApproxMostFrequentAccumulator;
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedAccumulator;
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedApproxCountDistinctAccumulator;
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedAvgAccumulator;
@@ -41,6 +46,8 @@
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedSumAccumulator;
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedUserDefinedAggregateAccumulator;
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedVarianceAccumulator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.IntGroupedApproxMostFrequentAccumulator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.LongGroupedApproxMostFrequentAccumulator;
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.UpdateMemory;
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.hash.MarkDistinctHash;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
@@ -57,6 +64,7 @@
 import org.apache.tsfile.read.common.block.column.IntColumn;
 import org.apache.tsfile.read.common.type.Type;
 import org.apache.tsfile.read.common.type.TypeFactory;
+import org.apache.tsfile.write.UnSupportedDataTypeException;
 
 import java.util.Arrays;
 import java.util.List;
@@ -246,6 +254,8 @@ private static GroupedAccumulator createBuiltinGroupedAccumulator(
             inputDataTypes.get(0), VarianceAccumulator.VarianceType.VAR_POP);
       case APPROX_COUNT_DISTINCT:
         return new GroupedApproxCountDistinctAccumulator(inputDataTypes.get(0));
+      case APPROX_MOST_FREQUENT:
+        return getGroupedApproxMostFrequentAccumulator(inputDataTypes.get(0));
       default:
         throw new IllegalArgumentException("Invalid Aggregation function: " + aggregationType);
     }
@@ -313,11 +323,63 @@ public static TableAccumulator createBuiltinAccumulator(
             inputDataTypes.get(0), VarianceAccumulator.VarianceType.VAR_POP);
       case APPROX_COUNT_DISTINCT:
         return new ApproxCountDistinctAccumulator(inputDataTypes.get(0));
+      case APPROX_MOST_FREQUENT:
+        return getApproxMostFrequentAccumulator(inputDataTypes.get(0));
       default:
         throw new IllegalArgumentException("Invalid Aggregation function: " + aggregationType);
     }
   }
 
+  public static GroupedAccumulator getGroupedApproxMostFrequentAccumulator(TSDataType type) {
+    switch (type) {
+      case BOOLEAN:
+        return new BooleanGroupedApproxMostFrequentAccumulator();
+      case INT32:
+      case DATE:
+        return new IntGroupedApproxMostFrequentAccumulator();
+      case INT64:
+      case TIMESTAMP:
+        return new LongGroupedApproxMostFrequentAccumulator();
+      case FLOAT:
+        return new FloatGroupedApproxMostFrequentAccumulator();
+      case DOUBLE:
+        return new DoubleGroupedApproxMostFrequentAccumulator();
+      case TEXT:
+      case STRING:
+        return new BinaryGroupedApproxMostFrequentAccumulator();
+      case BLOB:
+        return new BlobGroupedApproxMostFrequentAccumulator();
+      default:
+        throw new UnSupportedDataTypeException(
+            String.format("Unsupported data type in APPROX_COUNT_DISTINCT Aggregation: %s", type));
+    }
+  }
+
+  public static TableAccumulator getApproxMostFrequentAccumulator(TSDataType type) {
+    switch (type) {
+      case BOOLEAN:
+        return new BooleanApproxMostFrequentAccumulator();
+      case INT32:
+      case DATE:
+        return new IntApproxMostFrequentAccumulator();
+      case INT64:
+      case TIMESTAMP:
+        return new LongApproxMostFrequentAccumulator();
+      case FLOAT:
+        return new FloatApproxMostFrequentAccumulator();
+      case DOUBLE:
+        return new DoubleApproxMostFrequentAccumulator();
+      case TEXT:
+      case STRING:
+        return new BinaryApproxMostFrequentAccumulator();
+      case BLOB:
+        return new BlobApproxMostFrequentAccumulator();
+      default:
+        throw new UnSupportedDataTypeException(
+            String.format("Unsupported data type in APPROX_COUNT_DISTINCT Aggregation: %s", type));
+    }
+  }
+
   public static boolean isMultiInputAggregation(TAggregationType aggregationType) {
     switch (aggregationType) {
       case MAX_BY:

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/ApproxCountDistinctAccumulator.java

@@ -14,6 +14,9 @@
 
 package org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation;
 
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.approximate.HyperLogLog;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.approximate.HyperLogLogStateFactory;
+
 import org.apache.tsfile.block.column.Column;
 import org.apache.tsfile.block.column.ColumnBuilder;
 import org.apache.tsfile.enums.TSDataType;
@@ -24,7 +27,7 @@
 import org.apache.tsfile.write.UnSupportedDataTypeException;
 
 import static com.google.common.base.Preconditions.checkArgument;
-import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.HyperLogLog.DEFAULT_STANDARD_ERROR;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.approximate.HyperLogLog.DEFAULT_STANDARD_ERROR;
 
 public class ApproxCountDistinctAccumulator implements TableAccumulator {
   private static final long INSTANCE_SIZE =

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/BinaryApproxMostFrequentAccumulator.java

@@ -0,0 +1,130 @@
+/*
+ * Licensed 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.queryengine.execution.operator.source.relational.aggregation;
+
+import org.apache.iotdb.db.exception.sql.SemanticException;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.approximate.Counter;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.approximate.SpaceSaving;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.approximate.SpaceSavingStateFactory;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.RamUsageEstimator;
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+public class BinaryApproxMostFrequentAccumulator
+    extends AbstractApproxMostFrequentAccumulator<Binary> {
+
+  private static final long INSTANCE_SIZE =
+      RamUsageEstimator.shallowSizeOfInstance(BinaryApproxMostFrequentAccumulator.class);
+
+  public BinaryApproxMostFrequentAccumulator() {}
+
+  @Override
+  public long getEstimatedSize() {
+    return INSTANCE_SIZE
+        + RamUsageEstimator.shallowSizeOfInstance(BinaryApproxMostFrequentAccumulator.class)
+        + state.getEstimatedSize();
+  }
+
+  @Override
+  public TableAccumulator copy() {
+    return new BinaryApproxMostFrequentAccumulator();
+  }
+
+  @Override
+  public void addInput(Column[] arguments, AggregationMask mask) {
+    int maxBuckets = arguments[1].getInt(0);
+    int capacity = arguments[2].getInt(0);
+    if (maxBuckets <= 0 || capacity <= 0) {
+      throw new SemanticException(
+          "The second and third argument must be greater than 0, but got k="
+              + maxBuckets
+              + ", capacity="
+              + capacity);
+    }
+    SpaceSaving<Binary> spaceSaving = getOrCreateSpaceSaving(state, maxBuckets, capacity);
+
+    int positionCount = mask.getPositionCount();
+
+    Column valueColumn = arguments[0];
+
+    if (mask.isSelectAll()) {
+      for (int i = 0; i < valueColumn.getPositionCount(); i++) {
+        if (!valueColumn.isNull(i)) {
+          spaceSaving.add(valueColumn.getBinary(i));
+        }
+      }
+    } else {
+      int[] selectedPositions = mask.getSelectedPositions();
+      int position;
+      for (int i = 0; i < positionCount; i++) {
+        position = selectedPositions[i];
+        if (!valueColumn.isNull(position)) {
+          spaceSaving.add(valueColumn.getBinary(position));
+        }
+      }
+    }
+  }
+
+  @Override
+  public void addIntermediate(Column argument) {
+    for (int i = 0; i < argument.getPositionCount(); i++) {
+      if (!argument.isNull(i)) {
+        SpaceSaving<Binary> current =
+            new SpaceSaving<>(
+                argument.getBinary(i).getValues(),
+                BinaryApproxMostFrequentAccumulator::serializeBucket,
+                BinaryApproxMostFrequentAccumulator::deserializeBucket,
+                BinaryApproxMostFrequentAccumulator::calculateKeyByte);
+        state.merge(current);
+      }
+    }
+  }
+
+  public static SpaceSaving<Binary> getOrCreateSpaceSaving(
+      SpaceSavingStateFactory.SingleSpaceSavingState<Binary> state, int maxBuckets, int capacity) {
+    SpaceSaving<Binary> spaceSaving = state.getSpaceSaving();
+    if (spaceSaving == null) {
+      spaceSaving =
+          new SpaceSaving<>(
+              maxBuckets,
+              capacity,
+              BinaryApproxMostFrequentAccumulator::serializeBucket,
+              BinaryApproxMostFrequentAccumulator::deserializeBucket,
+              BinaryApproxMostFrequentAccumulator::calculateKeyByte);
+      state.setSpaceSaving(spaceSaving);
+    }
+    return spaceSaving;
+  }
+
+  public static void serializeBucket(Binary key, long count, ByteBuffer output) {
+    ReadWriteIOUtils.write(key, output);
+    ReadWriteIOUtils.write(count, output);
+  }
+
+  public static void deserializeBucket(ByteBuffer input, SpaceSaving<Binary> spaceSaving) {
+    Binary key = ReadWriteIOUtils.readBinary(input);
+    long count = ReadWriteIOUtils.readLong(input);
+    spaceSaving.add(key, count);
+  }
+
+  public static int calculateKeyByte(List<Counter<Binary>> counters) {
+    return counters.stream().mapToInt(counter -> counter.getItem().getLength()).sum();
+  }
+}