You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datasketches.apache.org by al...@apache.org on 2024/09/27 19:29:39 UTC
(datasketches-bigquery) 01/01: added agg functions to build sketches from int64
This is an automated email from the ASF dual-hosted git repository.
alsay pushed a commit to branch agg_int64
in repository https://gitbox.apache.org/repos/asf/datasketches-bigquery.git
commit 1e92c747bfb7fbf55c1dc30d30500ec3a642dd3d
Author: AlexanderSaydakov <Al...@users.noreply.github.com>
AuthorDate: Fri Sep 27 12:29:16 2024 -0700
added agg functions to build sketches from int64
---
cpc/cpc_sketch.cpp | 1 +
cpc/sqlx/cpc_sketch_agg_int64.sqlx | 36 ++++++
cpc/sqlx/cpc_sketch_agg_int64_lgk_seed.sqlx | 134 ++++++++++++++++++++++
cpc/test/cpc_sketch_test.sql | 8 +-
hll/hll_sketch.cpp | 1 +
hll/sqlx/hll_sketch_agg_int64.sqlx | 36 ++++++
hll/sqlx/hll_sketch_agg_int64_lgk_type.sqlx | 133 +++++++++++++++++++++
hll/test/hll_sketch_test.sql | 8 +-
theta/sqlx/theta_sketch_agg_int64.sqlx | 38 ++++++
theta/sqlx/theta_sketch_agg_int64_lgk_seed_p.sqlx | 132 +++++++++++++++++++++
theta/test/theta_sketch_test.sql | 8 +-
theta/theta_sketch.cpp | 1 +
12 files changed, 524 insertions(+), 12 deletions(-)
diff --git a/cpc/cpc_sketch.cpp b/cpc/cpc_sketch.cpp
index bb70a64..4f9da7a 100644
--- a/cpc/cpc_sketch.cpp
+++ b/cpc/cpc_sketch.cpp
@@ -40,6 +40,7 @@ EMSCRIPTEN_BINDINGS(cpc_sketch) {
return new datasketches::cpc_sketch(lg_k, seed);
}))
.function("updateString", emscripten::select_overload<void(const std::string&)>(&datasketches::cpc_sketch::update))
+ .function("updateInt64", emscripten::select_overload<void(uint64_t)>(&datasketches::cpc_sketch::update))
.function("serializeAsUint8Array", emscripten::optional_override([](const datasketches::cpc_sketch& self) {
auto bytes = self.serialize();
return Uint8Array.new_(emscripten::typed_memory_view(bytes.size(), bytes.data()));
diff --git a/cpc/sqlx/cpc_sketch_agg_int64.sqlx b/cpc/sqlx/cpc_sketch_agg_int64.sqlx
new file mode 100644
index 0000000..ea2137e
--- /dev/null
+++ b/cpc/sqlx/cpc_sketch_agg_int64.sqlx
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+config { hasOutput: true }
+
+CREATE OR REPLACE AGGREGATE FUNCTION ${self()}(value INT64)
+RETURNS BYTES
+OPTIONS (
+ description = '''Creates a sketch that represents the cardinality of the given INT64 column.
+
+Param value: the INT64 column of identifiers.
+Default lg_k and seed.
+Returns: a Compact, Compressed CPC Sketch, as BYTES
+
+For more information:
+ - https://datasketches.apache.org/docs/CPC/CpcSketches.html
+'''
+) AS (
+ $BQ_DATASET.cpc_sketch_agg_int64_lgk_seed(value, STRUCT<BYTEINT, INT64>(NULL, NULL))
+);
diff --git a/cpc/sqlx/cpc_sketch_agg_int64_lgk_seed.sqlx b/cpc/sqlx/cpc_sketch_agg_int64_lgk_seed.sqlx
new file mode 100644
index 0000000..57f4ffa
--- /dev/null
+++ b/cpc/sqlx/cpc_sketch_agg_int64_lgk_seed.sqlx
@@ -0,0 +1,134 @@
+/*
+ * 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.
+ */
+
+config { hasOutput: true }
+
+CREATE OR REPLACE AGGREGATE FUNCTION ${self()}(value INT64, params STRUCT<lg_k BYTEINT, seed INT64> NOT AGGREGATE)
+RETURNS BYTES
+LANGUAGE js
+OPTIONS (
+ library=["gs://$GCS_BUCKET/cpc_sketch.mjs"],
+ description = '''Creates a sketch that represents the cardinality of the given INT64 column.
+
+Param value: the INT64 column of identifiers.
+Param lg_k: the sketch accuracy/size parameter as an integer in the range [4, 26].
+Param seed: the seed to be used by the underlying hash function.
+Returns: a Compact, Compressed CPC Sketch, as BYTES
+
+For more information:
+ - https://datasketches.apache.org/docs/CPC/CpcSketches.html
+'''
+) AS R"""
+import ModuleFactory from "gs://$GCS_BUCKET/cpc_sketch.mjs";
+var Module = await ModuleFactory();
+const default_lg_k = Number(Module.DEFAULT_LG_K);
+const default_seed = BigInt(Module.DEFAULT_SEED);
+
+function destroyState(state) {
+ if (state.sketch) {
+ state.sketch.delete();
+ state.sketch = null;
+ }
+ if (state.union) {
+ state.union.delete();
+ state.union = null;
+ }
+ state.serialized = null;
+}
+
+// UDAF interface
+export function initialState(params) {
+ return {
+ lg_k: params.lg_k == null ? default_lg_k : Number(params.lg_k),
+ seed: params.seed == null ? default_seed : BigInt(params.seed)
+ };
+}
+
+export function aggregate(state, value) {
+ try {
+ if (state.sketch == null) {
+ state.sketch = new Module.cpc_sketch(state.lg_k, state.seed);
+ }
+ state.sketch.updateInt64(value);
+ } catch (e) {
+ if (e.message != null) throw e;
+ throw new Error(Module.getExceptionMessage(e));
+ }
+}
+
+export function serialize(state) {
+ if (state.sketch == null && state.union == null) return state; // for transition deserialize-serialize
+ try {
+ // for prior transition deserialize-aggregate
+ // merge aggregated and serialized state
+ if (state.sketch != null && state.serialized != null) {
+ var u = null;
+ try {
+ u = new Module.cpc_union(state.lg_k, state.seed);
+ u.updateWithSketch(state.sketch);
+ u.updateWithBytes(state.serialized, state.seed);
+ state.serialized = u.getResultAsUint8Array();
+ } finally {
+ if (u != null) u.delete();
+ }
+ } else if (state.sketch != null) {
+ state.serialized = state.sketch.serializeAsUint8Array();
+ } else if (state.union != null) {
+ state.serialized = state.union.getResultAsUint8Array();
+ }
+ return {
+ lg_k: state.lg_k,
+ seed: state.seed,
+ serialized: state.serialized
+ };
+ } catch (e) {
+ if (e.message != null) throw e;
+ throw new Error(Module.getExceptionMessage(e));
+ } finally {
+ destroyState(state);
+ }
+}
+
+export function deserialize(state) {
+ return state;
+}
+
+export function merge(state, other_state) {
+ try {
+ if (state.union == null) {
+ state.union = new Module.cpc_union(state.lg_k, state.seed);
+ }
+ if (state.serialized) {
+ state.union.updateWithBytes(state.serialized, state.seed);
+ state.serialized = null;
+ }
+ if (other_state.serialized) {
+ state.union.updateWithBytes(other_state.serialized, other_state.seed);
+ other_state.serialized = null;
+ }
+ } catch (e) {
+ if (e.message != null) throw e;
+ throw new Error(Module.getExceptionMessage(e));
+ }
+}
+
+export function finalize(state) {
+ return serialize(state).serialized;
+}
+""";
diff --git a/cpc/test/cpc_sketch_test.sql b/cpc/test/cpc_sketch_test.sql
index d9c88b8..c5f833b 100644
--- a/cpc/test/cpc_sketch_test.sql
+++ b/cpc/test/cpc_sketch_test.sql
@@ -42,9 +42,9 @@ select $BQ_DATASET.cpc_sketch_get_estimate_seed(
create or replace table $BQ_DATASET.cpc_sketch(sketch bytes);
insert into $BQ_DATASET.cpc_sketch
-(select $BQ_DATASET.cpc_sketch_agg_string(cast(value as string)) from unnest(GENERATE_ARRAY(1, 10000, 1)) as value);
+(select $BQ_DATASET.cpc_sketch_agg_int64(value) from unnest(GENERATE_ARRAY(1, 10000, 1)) as value);
insert into $BQ_DATASET.cpc_sketch
-(select $BQ_DATASET.cpc_sketch_agg_string(cast(value as string)) from unnest(GENERATE_ARRAY(100000, 110000, 1)) as value);
+(select $BQ_DATASET.cpc_sketch_agg_int64(value) from unnest(GENERATE_ARRAY(100000, 110000, 1)) as value);
select $BQ_DATASET.cpc_sketch_to_string(sketch) from $BQ_DATASET.cpc_sketch;
@@ -64,9 +64,9 @@ drop table $BQ_DATASET.cpc_sketch;
create or replace table $BQ_DATASET.cpc_sketch(sketch bytes);
insert into $BQ_DATASET.cpc_sketch
-(select $BQ_DATASET.cpc_sketch_agg_string_lgk_seed(cast(value as string), struct<byteint, int64>(10, 111)) from unnest(GENERATE_ARRAY(1, 10000, 1)) as value);
+(select $BQ_DATASET.cpc_sketch_agg_int64_lgk_seed(value, struct<byteint, int64>(10, 111)) from unnest(GENERATE_ARRAY(1, 10000, 1)) as value);
insert into $BQ_DATASET.cpc_sketch
-(select $BQ_DATASET.cpc_sketch_agg_string_lgk_seed(cast(value as string), struct<byteint, int64>(10, 111)) from unnest(GENERATE_ARRAY(100000, 110000, 1)) as value);
+(select $BQ_DATASET.cpc_sketch_agg_int64_lgk_seed(value, struct<byteint, int64>(10, 111)) from unnest(GENERATE_ARRAY(100000, 110000, 1)) as value);
select $BQ_DATASET.cpc_sketch_to_string_seed(sketch, 111) from $BQ_DATASET.cpc_sketch;
diff --git a/hll/hll_sketch.cpp b/hll/hll_sketch.cpp
index 35b7f92..ff0137d 100644
--- a/hll/hll_sketch.cpp
+++ b/hll/hll_sketch.cpp
@@ -43,6 +43,7 @@ EMSCRIPTEN_BINDINGS(hll_sketch) {
return new datasketches::hll_sketch(lg_k, convert_tgt_type(tgt_type_str));
}))
.function("updateString", emscripten::select_overload<void(const std::string&)>(&datasketches::hll_sketch::update))
+ .function("updateInt64", emscripten::select_overload<void(uint64_t)>(&datasketches::hll_sketch::update))
.function("serializeAsUint8Array", emscripten::optional_override([](const datasketches::hll_sketch& self) {
auto bytes = self.serialize_compact();
return Uint8Array.new_(emscripten::typed_memory_view(bytes.size(), bytes.data()));
diff --git a/hll/sqlx/hll_sketch_agg_int64.sqlx b/hll/sqlx/hll_sketch_agg_int64.sqlx
new file mode 100644
index 0000000..9d39931
--- /dev/null
+++ b/hll/sqlx/hll_sketch_agg_int64.sqlx
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+config { hasOutput: true }
+
+CREATE OR REPLACE AGGREGATE FUNCTION ${self()}(value INT64)
+RETURNS BYTES
+OPTIONS (
+ description = '''Creates a sketch that represents the cardinality of the given INT64 column.
+
+Param value: the INT64 column of identifiers.
+Default lg_k and tgt_type.
+Returns: an HLL Sketch, as BYTES.
+
+For more information:
+ - https://datasketches.apache.org/docs/HLL/HllSketches.html
+'''
+) AS (
+ $BQ_DATASET.hll_sketch_agg_int64_lgk_type(value, STRUCT<BYTEINT, STRING>(NULL, NULL))
+);
diff --git a/hll/sqlx/hll_sketch_agg_int64_lgk_type.sqlx b/hll/sqlx/hll_sketch_agg_int64_lgk_type.sqlx
new file mode 100644
index 0000000..b806472
--- /dev/null
+++ b/hll/sqlx/hll_sketch_agg_int64_lgk_type.sqlx
@@ -0,0 +1,133 @@
+/*
+ * 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.
+ */
+
+config { hasOutput: true }
+
+CREATE OR REPLACE AGGREGATE FUNCTION ${self()}(value INT64, params STRUCT<lg_k BYTEINT, tgt_type STRING> NOT AGGREGATE)
+RETURNS BYTES
+LANGUAGE js
+OPTIONS (
+ library=["gs://$GCS_BUCKET/hll_sketch.mjs"],
+ description = '''Creates a sketch that represents the cardinality of the given INT64 column.
+
+Param value: the INT64 column of identifiers.
+Param lg_k: the sketch accuracy/size parameter as an integer in the range [4, 21].
+Param tgt_type: The HLL type to use: one of {"HLL_4", "HLL_6", "HLL_8"}.
+Returns: an HLL Sketch, as BYTES.
+
+For more information:
+ - https://datasketches.apache.org/docs/HLL/HllSketches.html
+'''
+) AS R"""
+import ModuleFactory from "gs://$GCS_BUCKET/hll_sketch.mjs";
+var Module = await ModuleFactory();
+const default_lg_k = Number(12);
+
+function destroyState(state) {
+ if (state.sketch) {
+ state.sketch.delete();
+ state.sketch = null;
+ }
+ if (state.union) {
+ state.union.delete();
+ state.union = null;
+ }
+ state.serialized = null;
+}
+
+// UDAF interface
+export function initialState(params) {
+ return {
+ lg_k: params.lg_k == null ? default_lg_k : Number(params.lg_k),
+ tgt_type: params.tgt_type == null ? "" : params.tgt_type
+ };
+}
+
+export function aggregate(state, value) {
+ try {
+ if (state.sketch == null) { // for transition deserialize-aggregate
+ state.sketch = new Module.hll_sketch(state.lg_k, state.tgt_type);
+ }
+ state.sketch.updateInt64(value);
+ } catch (e) {
+ if (e.message != null) throw e;
+ throw new Error(Module.getExceptionMessage(e));
+ }
+}
+
+export function serialize(state) {
+ if (state.sketch == null && state.union == null) return state; // for transition deserialize-serialize
+ try {
+ // for prior transition deserialize-aggregate
+ // merge aggregated and serialized state
+ if (state.sketch != null && state.serialized != null) {
+ var u = null;
+ try {
+ u = new Module.hll_union(state.lg_k);
+ u.updateWithSketch(state.sketch);
+ u.updateWithBytes(state.serialized);
+ state.serialized = u.getResultAsUint8Array(state.tgt_type);
+ } finally {
+ if (u != null) u.delete();
+ }
+ } else if (state.sketch != null) {
+ state.serialized = state.sketch.serializeAsUint8Array();
+ } else if (state.union != null) {
+ state.serialized = state.union.getResultAsUint8Array(state.tgt_type);
+ }
+ return {
+ lg_k: state.lg_k,
+ tgt_type: state.tgt_type,
+ serialized: state.serialized
+ };
+ } catch (e) {
+ if (e.message != null) throw e;
+ throw new Error(Module.getExceptionMessage(e));
+ } finally {
+ destroyState(state);
+ }
+}
+
+export function deserialize(state) {
+ return state;
+}
+
+export function merge(state, other_state) {
+ try {
+ if (state.union == null) {
+ state.union = new Module.hll_union(state.lg_k);
+ }
+ if (state.serialized) {
+ state.union.updateWithBytes(state.serialized);
+ state.serialized = null;
+ }
+ if (other_state.serialized) {
+ state.union.updateWithBytes(other_state.serialized);
+ other_state.serialized = null;
+ }
+ } catch (e) {
+ if (e.message != null) throw e;
+ throw new Error(Module.getExceptionMessage(e));
+ }
+}
+
+export function finalize(state) {
+ return serialize(state).serialized;
+}
+""";
diff --git a/hll/test/hll_sketch_test.sql b/hll/test/hll_sketch_test.sql
index 04cd8eb..a8c406c 100644
--- a/hll/test/hll_sketch_test.sql
+++ b/hll/test/hll_sketch_test.sql
@@ -45,9 +45,9 @@ select $BQ_DATASET.hll_sketch_to_string(
create or replace table $BQ_DATASET.hll_sketch(sketch bytes);
insert into $BQ_DATASET.hll_sketch
-(select $BQ_DATASET.hll_sketch_agg_string(cast(value as string)) from unnest(GENERATE_ARRAY(1, 10000, 1)) as value);
+(select $BQ_DATASET.hll_sketch_agg_int64(value) from unnest(GENERATE_ARRAY(1, 10000, 1)) as value);
insert into $BQ_DATASET.hll_sketch
-(select $BQ_DATASET.hll_sketch_agg_string(cast(value as string)) from unnest(GENERATE_ARRAY(100000, 110000, 1)) as value);
+(select $BQ_DATASET.hll_sketch_agg_int64(value) from unnest(GENERATE_ARRAY(100000, 110000, 1)) as value);
# expected estimate about 20000
select $BQ_DATASET.hll_sketch_to_string(
@@ -63,9 +63,9 @@ drop table $BQ_DATASET.hll_sketch;
create or replace table $BQ_DATASET.hll_sketch(sketch bytes);
insert into $BQ_DATASET.hll_sketch
-(select $BQ_DATASET.hll_sketch_agg_string_lgk_type(cast(value as string), struct<int, string>(8, "HLL_6")) from unnest(GENERATE_ARRAY(1, 10000, 1)) as value);
+(select $BQ_DATASET.hll_sketch_agg_int64_lgk_type(value, struct<int, string>(8, "HLL_6")) from unnest(GENERATE_ARRAY(1, 10000, 1)) as value);
insert into $BQ_DATASET.hll_sketch
-(select $BQ_DATASET.hll_sketch_agg_string_lgk_type(cast(value as string), struct<int, string>(8, "HLL_6")) from unnest(GENERATE_ARRAY(100000, 110000, 1)) as value);
+(select $BQ_DATASET.hll_sketch_agg_int64_lgk_type(value, struct<int, string>(8, "HLL_6")) from unnest(GENERATE_ARRAY(100000, 110000, 1)) as value);
# expected estimate about 20000
select $BQ_DATASET.hll_sketch_to_string(
diff --git a/theta/sqlx/theta_sketch_agg_int64.sqlx b/theta/sqlx/theta_sketch_agg_int64.sqlx
new file mode 100644
index 0000000..3e4348a
--- /dev/null
+++ b/theta/sqlx/theta_sketch_agg_int64.sqlx
@@ -0,0 +1,38 @@
+/*
+ * 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.
+ */
+
+config { hasOutput: true }
+
+CREATE OR REPLACE AGGREGATE FUNCTION ${self()}(value INT64)
+RETURNS BYTES
+OPTIONS (
+ description = '''Creates a sketch that represents the cardinality of the given INT64 column.
+
+Param value: the INT64 column of identifiers.
+Param lg_k: assume default = 12
+Param seed: assume default = 9001
+Param p: assume default = 1.0
+Returns: a Compact, Compressed Theta Sketch, as BYTES.
+
+For more information:
+ - https://datasketches.apache.org/docs/Theta/ThetaSketches.html
+ '''
+) AS (
+ $BQ_DATASET.theta_sketch_agg_int64_lgk_seed_p(value, STRUCT<BYTEINT, INT64, FLOAT64>(NULL, NULL, NULL))
+);
diff --git a/theta/sqlx/theta_sketch_agg_int64_lgk_seed_p.sqlx b/theta/sqlx/theta_sketch_agg_int64_lgk_seed_p.sqlx
new file mode 100644
index 0000000..3a5a096
--- /dev/null
+++ b/theta/sqlx/theta_sketch_agg_int64_lgk_seed_p.sqlx
@@ -0,0 +1,132 @@
+/*
+ * 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.
+ */
+
+config { hasOutput: true }
+
+CREATE OR REPLACE AGGREGATE FUNCTION ${self()}(value INT64, params STRUCT<lg_k BYTEINT, seed INT64, p FLOAT64> NOT AGGREGATE)
+RETURNS BYTES
+LANGUAGE js
+OPTIONS (
+ library=["gs://$GCS_BUCKET/theta_sketch.mjs"],
+ description = '''Creates a sketch that represents the cardinality of the given INT64 column.
+
+Param value: the INT64 column of identifiers.
+Param lg_k: the sketch accuracy/size parameter as a BYTEINT in the range [4, 26]. A NULL specifies the default of 12.
+Param seed: the seed to be used by the underlying hash function. A NULL specifies the default of 9001.
+Param p: up-front sampling probability. A NULL specifies the default of 1.0.
+Returns: a Compact, Compressed Theta Sketch, as BYTES.
+
+For more information:
+ - https://datasketches.apache.org/docs/Theta/ThetaSketches.html
+ '''
+) AS R"""
+import ModuleFactory from "gs://$GCS_BUCKET/theta_sketch.mjs";
+var Module = await ModuleFactory();
+const default_lg_k = Number(Module.DEFAULT_LG_K);
+const default_seed = BigInt(Module.DEFAULT_SEED);
+const default_p = 1.0;
+
+function destroyState(state) {
+ if (state.sketch) {
+ state.sketch.delete();
+ state.sketch = null;
+ }
+ if (state.union) {
+ state.union.delete();
+ state.union = null;
+ }
+ state.serialized = null;
+}
+
+// UDAF interface
+export function initialState(params) {
+ var state = {
+ lg_k: params.lg_k == null ? default_lg_k : Number(params.lg_k),
+ seed: params.seed == null ? default_seed : BigInt(params.seed),
+ p: params.p == null ? default_p : params.p
+ };
+ state.sketch = new Module.update_theta_sketch(state.lg_k, state.seed, state.p);
+ return state;
+}
+
+export function aggregate(state, value) {
+ if (state.sketch == null) {
+ state.sketch = new Module.update_theta_sketch(state.lg_k, state.seed, state.p);
+ }
+ state.sketch.updateInt64(value);
+}
+
+export function serialize(state) {
+ if (state.sketch == null && state.union == null) return state; // for transition deserialize-serialize
+ try {
+ // for prior transition deserialize-aggregate
+ // merge aggregated and serialized state
+ if (state.sketch != null && state.serialized != null) {
+ var u = null;
+ try {
+ u = new Module.theta_union(state.lg_k, state.seed);
+ u.updateWithUpdateSketch(state.sketch);
+ u.updateWithBytes(state.serialized, state.seed);
+ state.serialized = u.getResultAsUint8ArrayCompressed();
+ } finally {
+ if (u != null) u.delete();
+ }
+ } else if (state.sketch != null) {
+ state.serialized = state.sketch.serializeAsUint8ArrayCompressed();
+ } else if (state.union != null) {
+ state.serialized = state.union.getResultAsUint8ArrayCompressed();
+ }
+ return {
+ lg_k: state.lg_k,
+ seed: state.seed,
+ serialized: state.serialized
+ };
+ } catch (e) {
+ throw new Error(Module.getExceptionMessage(e));
+ } finally {
+ destroyState(state);
+ }
+}
+
+export function deserialize(state) {
+ return state;
+}
+
+export function merge(state, other_state) {
+ try {
+ if (!state.union) {
+ state.union = new Module.theta_union(state.lg_k, state.seed);
+ }
+ if (state.serialized) {
+ state.union.updateWithBytes(state.serialized, state.seed);
+ state.serialized = null;
+ }
+ if (other_state.serialized) {
+ state.union.updateWithBytes(other_state.serialized, state.seed);
+ other_state.serialized = null;
+ }
+ } catch (e) {
+ throw new Error(Module.getExceptionMessage(e));
+ }
+}
+
+export function finalize(state) {
+ return serialize(state).serialized
+}
+""";
diff --git a/theta/test/theta_sketch_test.sql b/theta/test/theta_sketch_test.sql
index bbce179..e9f55b2 100644
--- a/theta/test/theta_sketch_test.sql
+++ b/theta/test/theta_sketch_test.sql
@@ -21,9 +21,9 @@
create or replace table $BQ_DATASET.theta_sketch(sketch bytes);
insert into $BQ_DATASET.theta_sketch
-(select $BQ_DATASET.theta_sketch_agg_string(cast(value as string)) from unnest(GENERATE_ARRAY(1, 10000, 1)) as value);
+(select $BQ_DATASET.theta_sketch_agg_int64(value) from unnest(GENERATE_ARRAY(1, 10000, 1)) as value);
insert into $BQ_DATASET.theta_sketch
-(select $BQ_DATASET.theta_sketch_agg_string(cast(value as string)) from unnest(GENERATE_ARRAY(100000, 110000, 1)) as value);
+(select $BQ_DATASET.theta_sketch_agg_int64(value) from unnest(GENERATE_ARRAY(100000, 110000, 1)) as value);
# expected about 20000
select $BQ_DATASET.theta_sketch_get_estimate_and_bounds(
@@ -42,9 +42,9 @@ drop table $BQ_DATASET.theta_sketch;
create or replace table $BQ_DATASET.theta_sketch(sketch bytes);
insert into $BQ_DATASET.theta_sketch
-(select $BQ_DATASET.theta_sketch_agg_string_lgk_seed_p(cast(value as string), struct<int, int, float64>(14, 111, 0.9)) from unnest(GENERATE_ARRAY(1, 10000, 1)) as value);
+(select $BQ_DATASET.theta_sketch_agg_int64_lgk_seed_p(value, struct<int, int, float64>(14, 111, 0.9)) from unnest(GENERATE_ARRAY(1, 10000, 1)) as value);
insert into $BQ_DATASET.theta_sketch
-(select $BQ_DATASET.theta_sketch_agg_string_lgk_seed_p(cast(value as string), struct<int, int, float64>(14, 111, 0.9)) from unnest(GENERATE_ARRAY(100000, 110000, 1)) as value);
+(select $BQ_DATASET.theta_sketch_agg_int64_lgk_seed_p(value, struct<int, int, float64>(14, 111, 0.9)) from unnest(GENERATE_ARRAY(100000, 110000, 1)) as value);
# expected about 20000
select $BQ_DATASET.theta_sketch_get_estimate_and_bounds_seed(
diff --git a/theta/theta_sketch.cpp b/theta/theta_sketch.cpp
index 5ec9664..5193cbc 100644
--- a/theta/theta_sketch.cpp
+++ b/theta/theta_sketch.cpp
@@ -50,6 +50,7 @@ EMSCRIPTEN_BINDINGS(theta_sketch) {
return new update_theta_sketch(update_theta_sketch::builder().set_lg_k(lg_k).set_seed(seed).set_p(p).build());
}))
.function("updateString", emscripten::select_overload<void(const std::string&)>(&update_theta_sketch::update))
+ .function("updateInt64", emscripten::select_overload<void(uint64_t)>(&update_theta_sketch::update))
.function("serializeAsUint8ArrayCompressed", emscripten::optional_override([](const update_theta_sketch& self) {
auto bytes = self.compact().serialize_compressed();
return Uint8Array.new_(emscripten::typed_memory_view(bytes.size(), bytes.data()));
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datasketches.apache.org
For additional commands, e-mail: commits-help@datasketches.apache.org