diff --git a/Cargo.lock b/Cargo.lock index a2ca6ee..fa005ba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -783,9 +783,9 @@ dependencies = [ [[package]] name = "datafusion" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96f76f0167ed0842b29a3d1e41be3c034c0a46409a3a703cc4cc84ee8c24abf4" +checksum = "14313430439ef858ed5f52ca9c840e9eea9844da34d3716d05a476449634da5d" dependencies = [ "arrow", "arrow-schema", @@ -836,9 +836,9 @@ dependencies = [ [[package]] name = "datafusion-catalog" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d79ec3460f6ed5c58f9b3f2d873fbc77748b82653bff1b4cdaf06de33bb4e05f" +checksum = "2c0756ebc770bcebae366dfd273dad50935f6103b4d212253cb17e2e53c5427f" dependencies = [ "arrow", "async-trait", @@ -861,9 +861,9 @@ dependencies = [ [[package]] name = "datafusion-catalog-listing" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b48cef241e2efcfd496fe05ae4d0d5de20793451862faefe406c397a467e12d4" +checksum = "bbde2961698705d8402f0f382f08876fc6f1126ffde790ef17833cf0ab87cd59" dependencies = [ "arrow", "async-trait", @@ -885,9 +885,9 @@ dependencies = [ [[package]] name = "datafusion-common" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3f72810485975c258f1b4d00baab31728470676c60c5546f366ebd0d99f05ab6" +checksum = "a6e06fe711ff815c8cf3143158547aea599702fd43f76f5866c70d5a5b3693a0" dependencies = [ "arrow", "arrow-ipc", @@ -912,9 +912,9 @@ dependencies = [ [[package]] name = "datafusion-common-runtime" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "533c28e75dba52f41bde187d23a1cb24ab91c7c097966824fa471e67b60320ea" +checksum = "f7a5107acff4e3ed50a7732acec06bba40e74ab49df021397ed250d60f04e42a" dependencies = [ "futures", "log", @@ -923,9 +923,9 @@ dependencies = [ [[package]] name = "datafusion-datasource" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b00a1fa0da26f6087136a82fea7f13c76a672cbab452d4086952a7cf770a19b" +checksum = "f69f45d2d700785be8df9d4b44410e1c8e4a856f10cc728b418a1066c0b594fb" dependencies = [ "arrow", "async-compression", @@ -959,9 +959,9 @@ dependencies = [ [[package]] name = "datafusion-datasource-arrow" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5ad17ec881bff2ed7768b4bfe971d3efbf3473f2fd1f9d365447bccbdf908678" +checksum = "81650af64894ef34b46d2ef818ef244b40a81d90818f3bc642bf8fbf6758f8e0" dependencies = [ "arrow", "arrow-ipc", @@ -983,9 +983,9 @@ dependencies = [ [[package]] name = "datafusion-datasource-csv" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b5345285b0c3eaab412e7539b706973c083bd7e5bce575de5e0a3da488d08d1d" +checksum = "b5e2519bed59a6907c79298b9c21fe807bbc34e4ff8414577d699b0741083593" dependencies = [ "arrow", "async-trait", @@ -1006,9 +1006,9 @@ dependencies = [ [[package]] name = "datafusion-datasource-json" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "da02fb9324f56bd8c53f1ee2e949547425cb66f76adc6832b10d44f80a1221d2" +checksum = "635e61290e171597f6e50427e0cb486eda468ab47e880fa24d15bbb0fa24ac7e" dependencies = [ "arrow", "async-trait", @@ -1029,9 +1029,9 @@ dependencies = [ [[package]] name = "datafusion-datasource-parquet" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3c0b0dc1453952952fd5c69ad1c7f6042176e69ed233011d47e07cf74ed0949e" +checksum = "07cd518f85b3a027d913047b4ab0811111a2b734fdf84a51b78234455f10d8fa" dependencies = [ "arrow", "arrow-schema", @@ -1061,15 +1061,15 @@ dependencies = [ [[package]] name = "datafusion-doc" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a88fd985bc0550c36f557db69543cc9d6393b1509783520b30e902f23c555da6" +checksum = "d44c5919b560862dd1cb20a5c75b3e48c3d9e49719c91d78da5c38eaad5495c4" [[package]] name = "datafusion-execution" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a98f1052f91b4991f0bf2ce1e4e36dfbdcda454a956b8c8d562c7c845e8fce1d" +checksum = "7620ad535af8730186bdec05d720e40d9a4584b40c7eb75f72c43aa7978e5484" dependencies = [ "arrow", "arrow-buffer", @@ -1093,9 +1093,9 @@ dependencies = [ [[package]] name = "datafusion-expr" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "464625a1f0e4b9df552d894fafcc8aac953ebbc8b0fa0acdaf20975fd615040e" +checksum = "ea229af8b73e566fc6ebb1c8dd42ab250f8855c5304bd173f4b50ead78b64304" dependencies = [ "arrow", "arrow-schema", @@ -1116,9 +1116,9 @@ dependencies = [ [[package]] name = "datafusion-expr-common" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2604994999d5aeca1d1df645ffc98bc787447aaff05dde27aad0342b48fc1fe0" +checksum = "6a18baa840f77760dcc9eaad281329bc57536c5c39e65e8649596f2a51216be7" dependencies = [ "arrow", "datafusion-common", @@ -1128,9 +1128,9 @@ dependencies = [ [[package]] name = "datafusion-functions" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "051e97533e6af53e4aa0a0667cadc886abcaf36c4a5925019c55c0aa4c218fde" +checksum = "763bbfeec40b55a8a961d7349c0fcfcec4f3f66cc234d6295250b95371b85e34" dependencies = [ "arrow", "arrow-buffer", @@ -1160,9 +1160,9 @@ dependencies = [ [[package]] name = "datafusion-functions-aggregate" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2d0f1bb166d3572b6ed40e1afb2faaacade962abc08c2fcf04babee74681c56b" +checksum = "46c0c2b2e856883ecf6a4fa8b23e7c6041183af8a68d0583dc95d1d21e4eaee0" dependencies = [ "arrow", "datafusion-common", @@ -1181,9 +1181,9 @@ dependencies = [ [[package]] name = "datafusion-functions-aggregate-common" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7ed756770f5f98369e181d692fd5ee6b1127ffd7322caba92f3730f9f5c92333" +checksum = "e35e42c87dfbff06934ea0e99cfed3440ac614c3f3bfb8c33be35ca0587d4140" dependencies = [ "arrow", "datafusion-common", @@ -1193,9 +1193,9 @@ dependencies = [ [[package]] name = "datafusion-functions-nested" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91173fdb5c0ff2a41169a8ffa1b385b8844f18728747bb0a37e35ad7d5772a4f" +checksum = "08079b0604cdcc43c7d6cf006cfd5d3ed9c235ade0c1ebe91085a62f875603df" dependencies = [ "arrow", "arrow-ord", @@ -1218,9 +1218,9 @@ dependencies = [ [[package]] name = "datafusion-functions-table" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b1bcdfb286a745461b126719c32700777e83df4f17cc44db5d71ebce5731e840" +checksum = "fe2070ecf5638a93c81c8babc7858339ef52165a9f54da1411afd69adf7d85ce" dependencies = [ "arrow", "async-trait", @@ -1234,9 +1234,9 @@ dependencies = [ [[package]] name = "datafusion-functions-window" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9ec4b508f1f93f00038ba3e737e894ec6c775528b4369413386655ae6125f0fc" +checksum = "0e9a164ebe1998c5c3f9e15c9a0c4e6b599a0c8739b5471bfb089752441963f1" dependencies = [ "arrow", "datafusion-common", @@ -1251,9 +1251,9 @@ dependencies = [ [[package]] name = "datafusion-functions-window-common" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0b352020834140073fbf5b46ee0ceb926e5074a9d0bcae1dbd91d0586d999cde" +checksum = "98965b183f92201c16798994820ae032649c4821e089c4c80e7d0f4485d41df5" dependencies = [ "datafusion-common", "datafusion-physical-expr-common", @@ -1261,9 +1261,9 @@ dependencies = [ [[package]] name = "datafusion-macros" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "15192effab05d38cce10e92a6fb48c967b5f166b27b7195a165a72b232569c58" +checksum = "3a711bd5a06ec85d9683e142d28a98ad0679a7220669f5c1b8589e57e9b29e46" dependencies = [ "datafusion-doc", "quote", @@ -1272,9 +1272,9 @@ dependencies = [ [[package]] name = "datafusion-optimizer" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "854445d9f7847e1e46089cf61b8d341a64382f14484e912c83a0f23b31216896" +checksum = "ff3256b62ff44616ec83c7f04741d21124970833bb377ed56d9abc5ec6f707ec" dependencies = [ "arrow", "chrono", @@ -1292,11 +1292,12 @@ dependencies = [ [[package]] name = "datafusion-physical-expr" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "671558dad1d2aa253c39c0a4c52515958b99eb91abf649f4b88d5e69cc55282f" +checksum = "c1e37a25b5aa1f3a1db114f0ee3605bf01639b83a831dcde35119ee61f0b11d6" dependencies = [ "arrow", + "arrow-schema", "datafusion-common", "datafusion-expr", "datafusion-expr-common", @@ -1314,9 +1315,9 @@ dependencies = [ [[package]] name = "datafusion-physical-expr-adapter" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ffae3d78c2da80ecc829cb58536cc5aca2e99cf1365eda694fc75bfe288861e0" +checksum = "3a4d12787d42e88b26b0d3ec006192fa23ceb832ef2a9e87439ce820a8e038cc" dependencies = [ "arrow", "datafusion-common", @@ -1329,9 +1330,9 @@ dependencies = [ [[package]] name = "datafusion-physical-expr-common" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3d9092ed15e7203fbd0903215172f7c9d18f10d94cba35137f3b3836f7c46f16" +checksum = "d13bf7f50fd902f5afc46d50759f5185b5620c40b24054f80f7c648897d3e5a6" dependencies = [ "arrow", "chrono", @@ -1346,9 +1347,9 @@ dependencies = [ [[package]] name = "datafusion-physical-optimizer" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9005b6cf50b57b72d476c6ed4662b04be7ca6be5320ba9127c6d0b7e4218095b" +checksum = "11fbe7e6665072957d44ed9587676e17a1f9ab3acfadfc287dac23750585f18a" dependencies = [ "arrow", "datafusion-common", @@ -1366,9 +1367,9 @@ dependencies = [ [[package]] name = "datafusion-physical-plan" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5787e4fcff4adc4fce8948441103a99705018b49c8dff0720b650bd7a15da112" +checksum = "1265d58e5bce07d154e642a51ff43033b576a6ae40d50a29ac2b9f311004eb52" dependencies = [ "arrow", "arrow-data", @@ -1401,9 +1402,9 @@ dependencies = [ [[package]] name = "datafusion-pruning" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e651c8df0b90daed6a7be5921ec0ee379e6909705f063eeff70fd4e35010e4c" +checksum = "74b06b333405c05015836ed36cea02cc29d1e2ee8fb414a77532d8346fb56492" dependencies = [ "arrow", "datafusion-common", @@ -1417,9 +1418,9 @@ dependencies = [ [[package]] name = "datafusion-session" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb56667ee38217efab19b895d9a936052cfb47ed438a19663351bdc42a6214a1" +checksum = "e7616b17b1d2fee3e9b13c651eac91b768d153265e6ba88e06e0b4af7883f69c" dependencies = [ "arrow-schema", "async-trait", @@ -1432,9 +1433,9 @@ dependencies = [ [[package]] name = "datafusion-sql" -version = "55.0.0" +version = "55.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9c29067cb9d32f8e603c45e15d61ea18f1069f96ceafeceb4e18466b8e5b31d9" +checksum = "5b4848be66dd008d55adbe1dbc8cbd6a2502588ad0c0b8483bab0f43fd4495d0" dependencies = [ "arrow", "bigdecimal", diff --git a/Cargo.toml b/Cargo.toml index f64a365..10383ba 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,7 +11,7 @@ exclude = [ [dependencies] # Core dependencies -datafusion = "55.0.0" +datafusion = "55.0.1" async-trait = "0.1" tokio = {version = "1", features = ["full"] } url = "2" diff --git a/src/algorithm/centrality/hyperanf.rs b/src/algorithm/centrality/hyperanf.rs index 9aaab34..897e251 100644 --- a/src/algorithm/centrality/hyperanf.rs +++ b/src/algorithm/centrality/hyperanf.rs @@ -43,7 +43,7 @@ impl<'a> HyperANFBuilder<'a> { graph: graph, directed: true, n_hops: 2, - lg_k: 12, + lg_k: 8, checkpoint_config: CheckpointConfig::default_local_fs(), } } @@ -126,7 +126,7 @@ impl<'a> HyperANFBuilder<'a> { .add_message(pregel_src(BALL), MessageDirection::SrcToDst) // Collapse the per-edge messages into one sketch per destination // vertex before the update. - .add_aggregate_expr(hll_long_aggregate(pregel_default_msg())) + .add_aggregate_expr(hll_long_aggregate(pregel_default_msg(), self.lg_k)) .with_participation_column( "changed", lit(true), @@ -257,7 +257,7 @@ mod tests { Ok(map) } - /// HLL is near-exact at these small cardinalities (lg_k=12), so a 0.5 + /// HLL is near-exact at these small cardinalities (lg_k=8), so a 0.5 /// absolute tolerance both rounds to the true integer count and catches any /// participation/aggregation bug that drops a neighbour (a shift of >= 1). fn assert_size(map: &HashMap, id: i64, expected: i64) { diff --git a/src/expressions/hll.rs b/src/expressions/hll.rs index 7218bee..6682ab0 100644 --- a/src/expressions/hll.rs +++ b/src/expressions/hll.rs @@ -17,9 +17,11 @@ //! //! Sketches are carried as `Binary` columns, but the UDFs accept `BinaryView` //! as well, because parquet spills — the foundation of the out-of-core Pregel -//! engine — round-trip `Binary` as `BinaryView`. Note that `hll_long_aggregate` -//! is pinned to `lg_k = 12` (`DEFAULT_LG_K`); keep it paired with sketches -//! built at the same `lg_k`, or it will silently downsample. +//! engine — round-trip `Binary` as `BinaryView`. Every `hll_*` constructor +//! takes an explicit `lg_k`; keep `hll_long_aggregate` paired with sketches +//! built at the same `lg_k`, or the merged sketch silently adopts a different +//! precision — and a different memory footprint (an Hll8 array-mode image is +//! 2^lg_k bytes, so mismatched `lg_k` directly changes the algorithm's budget). //! //! These are the building blocks for HyperANF / HyperBALL and approximate //! closeness centrality (see `algorithm::centrality::hyperanf`). @@ -43,7 +45,6 @@ use crate::expressions::common::as_binary_like; use crate::expressions::common::downcast_int64; const DEFAULT_HLL_TYPE: HllType = HllType::Hll8; -const DEFAULT_LG_K: u8 = 12; #[derive(Debug, PartialEq, Eq, Hash)] pub(crate) struct HllLong { @@ -420,8 +421,13 @@ impl AggregateUDFImpl for HllAggregate { } } -pub(crate) fn hll_long_aggregate(col: Expr) -> Expr { - AggregateUDF::from(HllAggregate::new(DEFAULT_LG_K)).call(vec![col]) +/// Aggregate a column of serialized sketches into one sketch per group. +/// +/// `lg_k` must match the `lg_k` the input sketches were built with, so the +/// merged sketch keeps the caller's chosen precision (and register-array +/// footprint) instead of silently growing to a hard-coded default. +pub(crate) fn hll_long_aggregate(col: Expr, lg_k: u8) -> Expr { + AggregateUDF::from(HllAggregate::new(lg_k)).call(vec![col]) } #[cfg(test)] @@ -1048,7 +1054,7 @@ mod tests { let out = df .aggregate( vec![col("g")], - vec![hll_long_aggregate(hll_long(col("id"), 12)).alias("s")], + vec![hll_long_aggregate(hll_long(col("id"), 12), 12).alias("s")], )? .select(vec![hll_long_estimate(col("s")).alias("est")])? .collect() @@ -1076,7 +1082,7 @@ mod tests { let out = df .aggregate( vec![col("g")], - vec![hll_long_aggregate(hll_long(col("id"), 12)).alias("s")], + vec![hll_long_aggregate(hll_long(col("id"), 12), 12).alias("s")], )? .select(vec![col("g"), hll_long_estimate(col("s")).alias("est")])? .sort(vec![col("g").sort(true, true)])? @@ -1143,7 +1149,7 @@ mod tests { .unwrap() .aggregate( vec![col("g")], - vec![hll_long_aggregate(col("s")).alias("m")], + vec![hll_long_aggregate(col("s"), 12).alias("m")], ) .unwrap() .select(vec![hll_long_estimate(col("m")).alias("est")]) @@ -1204,7 +1210,7 @@ mod tests { .unwrap() .aggregate( vec![col("g")], - vec![hll_long_aggregate(hll_long(col("id"), 12)).alias("s")], + vec![hll_long_aggregate(hll_long(col("id"), 12), 12).alias("s")], ) .unwrap() .select(vec![hll_long_estimate(col("s")).alias("est")])