change to use opensearch core code

Signed-off-by: xinyual <xinyual@amazon.com>

Author: xinyual

core/build.gradle

@@ -46,6 +46,7 @@ pitest {
 }
 
 dependencies {
+    api group: 'org.opensearch', name: 'opensearch', version: "${opensearch_version}"
     api group: 'com.google.guava', name: 'guava', version: "${guava_version}"
     api group: 'org.apache.commons', name: 'commons-lang3', version: "${commons_lang3_version}"
     api group: 'org.apache.commons', name: 'commons-text', version: "${commons_text_version}"

core/src/main/java/org/opensearch/sql/calcite/udf/udaf/DistinctCountApproxAggFunction.java

@@ -5,7 +5,12 @@
 
 package org.opensearch.sql.calcite.udf.udaf;
 
-import com.google.zetasketch.HyperLogLogPlusPlus;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+import org.opensearch.common.hash.MurmurHash3;
+import org.opensearch.common.util.BigArrays;
+import org.opensearch.search.aggregations.metrics.HyperLogLogPlusPlus;
 import org.opensearch.sql.calcite.udf.UserDefinedAggFunction;
 
 /** The function use HyperLogLogPlusPlus to count distinct count approximate value */
@@ -26,26 +31,51 @@ public Object result(HLLAccumulator accumulator) {
   public HLLAccumulator add(HLLAccumulator acc, Object... values) {
     for (Object value : values) {
       if (value != null) {
-        acc.add(value.toString());
+        acc.add(value);
       }
     }
     return acc;
   }
 
   public static class HLLAccumulator implements UserDefinedAggFunction.Accumulator {
-    private final HyperLogLogPlusPlus<String> hll;
+    private final HyperLogLogPlusPlus hll;
 
     public HLLAccumulator() {
-      this.hll = new HyperLogLogPlusPlus.Builder().buildForStrings();
+      this.hll =
+          new HyperLogLogPlusPlus(
+              HyperLogLogPlusPlus.DEFAULT_PRECISION, BigArrays.NON_RECYCLING_INSTANCE, 1);
     }
 
-    public void add(String value) {
-      hll.add(value);
+    public void add(Object value) {
+      hll.collect(0, hash(value));
     }
 
     @Override
     public Object value(Object... args) {
-      return hll.result();
+      return hll.cardinality(0);
     }
   }
+
+  private static long hash(Object data) {
+    MurmurHash3.Hash128 hash = new MurmurHash3.Hash128();
+    if (data == null) {
+      return 0L;
+    }
+
+    byte[] bytes;
+
+    if (data instanceof byte[]) {
+      bytes = (byte[]) data;
+    } else if (data instanceof String) {
+      bytes = ((String) data).getBytes(StandardCharsets.UTF_8);
+    } else if (data instanceof Number) {
+      long value = ((Number) data).longValue();
+      bytes = ByteBuffer.allocate(Long.BYTES).order(ByteOrder.LITTLE_ENDIAN).putLong(value).array();
+    } else {
+      bytes = data.toString().getBytes(StandardCharsets.UTF_8);
+    }
+
+    MurmurHash3.hash128(bytes, 0, bytes.length, 0, hash);
+    return hash.h1 ^ hash.h2;
+  }
 }