diff --git a/Cargo.lock b/Cargo.lock index 7be9de61..e3aa5f0c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,20 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "ahash" +version = "0.8.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" +dependencies = [ + "cfg-if", + "const-random", + "getrandom 0.3.4", + "once_cell", + "version_check", + "zerocopy", +] + [[package]] name = "aho-corasick" version = "1.1.5" @@ -11,12 +25,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "android-tzdata" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e999941b234f3131b00bc13c22d06e8c5ff726d1b6318ac7eb276997bbb4fef0" - [[package]] name = "android_system_properties" version = "0.1.6" @@ -82,6 +90,183 @@ version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" +[[package]] +name = "arrow" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6cfdd0833e32a9874d2b55089333ad310c0be208aafa277385ce2461dec90be3" +dependencies = [ + "arrow-arith", + "arrow-array", + "arrow-buffer", + "arrow-cast", + "arrow-data", + "arrow-ord", + "arrow-row", + "arrow-schema", + "arrow-select", + "arrow-string", +] + +[[package]] +name = "arrow-arith" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0a41203398f0eaa6f7ec8e62c0da742a21abf282c148fc157f6c35c90e29981a" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "chrono", + "num-traits", +] + +[[package]] +name = "arrow-array" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae33dad492b7df00a217563a7b0ef2874df68a0deea1b1a3acf628152f7f7a69" +dependencies = [ + "ahash", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "chrono", + "half", + "hashbrown", + "num-complex", + "num-integer", + "num-traits", +] + +[[package]] +name = "arrow-buffer" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9552f96391c005e6ab449fa941420935e7e062489b12b8b1b08879b2163f5b5" +dependencies = [ + "bytes", + "half", + "num-bigint", + "num-traits", +] + +[[package]] +name = "arrow-cast" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a8a327c9649f30d8406995f27642b68df354713cca3baaaf100f076f18d5f34" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-ord", + "arrow-schema", + "arrow-select", + "atoi", + "base64", + "chrono", + "half", + "lexical-core", + "num-traits", + "ryu", +] + +[[package]] +name = "arrow-data" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b24852db04738907e06c04ea61e42fe7fda962a34513022dc0d0e754fb7976b" +dependencies = [ + "arrow-buffer", + "arrow-schema", + "half", + "num-integer", + "num-traits", +] + +[[package]] +name = "arrow-ord" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63a083ec750f5c043f02946b4baf05fcdbb55f4560a3277055caca5cc99f3eb0" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "arrow-select", +] + +[[package]] +name = "arrow-row" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "514ba0ef0d4c5896202dae736251ce415abb43a950bed570fb7981b8716c0e4c" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "half", +] + +[[package]] +name = "arrow-schema" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "21ca356ad6425cecb6eb7b28e4f659f1ee7880fbb1a16127de7dd62901efee9e" + +[[package]] +name = "arrow-select" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c58da39eb3d8350ad4a549e5c2bc49284dac554016c69829310350f1731b0aad" +dependencies = [ + "ahash", + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "num-traits", +] + +[[package]] +name = "arrow-string" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6789b388467525e3271326b6b4915666ecfdf5142aef09779445c954b67543c" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "arrow-select", + "memchr", + "num-traits", + "regex", + "regex-syntax", +] + +[[package]] +name = "asap-executor" +version = "0.1.0" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f9dd41e095997fe2c34c1a1278f9f20c50a814ff#f9dd41e095997fe2c34c1a1278f9f20c50a814ff" +dependencies = [ + "ahash", + "arrow", + "asap-types", + "asap_sketchlib", + "chrono", + "futures", + "regex", + "serde", + "serde_json", + "thiserror 2.0.20", + "tracing", +] + [[package]] name = "asap-frontend-common" version = "0.1.0" @@ -177,6 +362,15 @@ dependencies = [ "syn 3.0.5", ] +[[package]] +name = "atoi" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f28d99ec8bfea296261ca1af174f24225171fea9664ba9003cbebee704810528" +dependencies = [ + "num-traits", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -265,6 +459,15 @@ version = "2.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + [[package]] name = "bumpalo" version = "3.20.3" @@ -326,22 +529,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.3.1", "rand_core 0.10.1", ] [[package]] name = "chrono" -version = "0.4.39" +version = "0.4.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7e36cc9d416881d2e24f9a963be5fb1cd90966419ac844274161d10488b3e825" +checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327" dependencies = [ - "android-tzdata", "iana-time-zone", "js-sys", "num-traits", "wasm-bindgen", - "windows-targets", + "windows-link", ] [[package]] @@ -390,6 +592,26 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" +[[package]] +name = "const-random" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87e00182fe74b066627d63b85fd550ac2998d4b0bd86bfed477a0ae4c7c71359" +dependencies = [ + "const-random-macro", +] + +[[package]] +name = "const-random-macro" +version = "0.1.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9d839f2a20b0aee515dc581a6172f2321f96cab76c1a38a4c584a194955390e" +dependencies = [ + "getrandom 0.2.17", + "once_cell", + "tiny-keccak", +] + [[package]] name = "control_plane" version = "0.1.0" @@ -413,6 +635,15 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + [[package]] name = "cpufeatures" version = "0.3.1" @@ -422,11 +653,29 @@ dependencies = [ "libc", ] +[[package]] +name = "crunchy" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" + +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + [[package]] name = "data_plane" version = "0.1.0" dependencies = [ "anyhow", + "asap-executor", + "asap-types", "axum", "clap", "control_plane", @@ -435,6 +684,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "sha2", "snap", "thiserror 1.0.69", "tokio", @@ -449,6 +699,16 @@ version = "0.5.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", +] + [[package]] name = "displaydoc" version = "0.2.7" @@ -513,6 +773,21 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "futures" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.34" @@ -520,6 +795,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -528,6 +804,40 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" +[[package]] +name = "futures-executor" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" + +[[package]] +name = "futures-macro" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.5", +] + +[[package]] +name = "futures-sink" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" + [[package]] name = "futures-task" version = "0.3.34" @@ -540,12 +850,27 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ + "futures-channel", "futures-core", + "futures-io", + "futures-macro", + "futures-sink", "futures-task", + "memchr", "pin-project-lite", "slab", ] +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + [[package]] name = "getopts" version = "0.2.24" @@ -594,6 +919,18 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "half" +version = "2.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ea2d84b969582b4b1864a92dc5d27cd2b77b622a8d79306834f1be5ba20d84b" +dependencies = [ + "cfg-if", + "crunchy", + "num-traits", + "zerocopy", +] + [[package]] name = "hashbrown" version = "0.17.1" @@ -893,12 +1230,75 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" +[[package]] +name = "lexical-core" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d8d125a277f807e55a77304455eb7b1cb52f2b18c143b60e766c120bd64a594" +dependencies = [ + "lexical-parse-float", + "lexical-parse-integer", + "lexical-util", + "lexical-write-float", + "lexical-write-integer", +] + +[[package]] +name = "lexical-parse-float" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52a9f232fbd6f550bc0137dcb5f99ab674071ac2d690ac69704593cb4abbea56" +dependencies = [ + "lexical-parse-integer", + "lexical-util", +] + +[[package]] +name = "lexical-parse-integer" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a7a039f8fb9c19c996cd7b2fcce303c1b2874fe1aca544edc85c4a5f8489b34" +dependencies = [ + "lexical-util", +] + +[[package]] +name = "lexical-util" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2604dd126bb14f13fb5d1bd6a66155079cb9fa655b37f875b3a742c705dbed17" + +[[package]] +name = "lexical-write-float" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "50c438c87c013188d415fbabbb1dceb44249ab81664efbd31b14ae55dabb6361" +dependencies = [ + "lexical-util", + "lexical-write-integer", +] + +[[package]] +name = "lexical-write-integer" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "409851a618475d2d5796377cad353802345cba92c867d9fbcde9cf4eac4e14df" +dependencies = [ + "lexical-util", +] + [[package]] name = "libc" version = "0.2.189" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" +[[package]] +name = "libm" +version = "0.2.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" + [[package]] name = "litemap" version = "0.8.3" @@ -1027,12 +1427,40 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "num-bigint" +version = "0.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c89e69e7e0f03bea5ef08013795c25018e101932225a656383bd384495ecc367" +dependencies = [ + "num-integer", + "num-traits", +] + +[[package]] +name = "num-complex" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "73f88a1307638156682bada9d7604135552957b7818057dcef22705b4d509495" +dependencies = [ + "num-traits", +] + [[package]] name = "num-conv" version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" +[[package]] +name = "num-integer" +version = "0.1.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ce2d95d4b3734dc35aa2f45e1aa22cd416814592a4f9d9205e11affd5b8e10b" +dependencies = [ + "num-traits", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -1040,6 +1468,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" dependencies = [ "autocfg", + "libm", ] [[package]] @@ -1562,6 +1991,17 @@ dependencies = [ "serde", ] +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "digest", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -1774,6 +2214,15 @@ dependencies = [ "time-core", ] +[[package]] +name = "tiny-keccak" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c9d3793400a45f954c52e73d068316d76b6f4e36977e3fcebb13a2721e80237" +dependencies = [ + "crunchy", +] + [[package]] name = "tinystr" version = "0.8.4" @@ -1960,6 +2409,12 @@ dependencies = [ "rand 0.10.2", ] +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "unicode-ident" version = "1.0.24" @@ -2019,6 +2474,12 @@ dependencies = [ "time", ] +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + [[package]] name = "vob" version = "3.0.6" diff --git a/Cargo.toml b/Cargo.toml index 21be05e5..3469c8f0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,6 +13,7 @@ planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f9dd41e095997fe2c34c1a1278f9f20c50a814ff" } asap-logical-asap-optimization = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f9dd41e095997fe2c34c1a1278f9f20c50a814ff" } asap-plan-selection = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f9dd41e095997fe2c34c1a1278f9f20c50a814ff" } +asap-executor = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f9dd41e095997fe2c34c1a1278f9f20c50a814ff" } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 2aaf8975..0a5351e4 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -5,6 +5,9 @@ edition.workspace = true [dependencies] control_plane.workspace = true +planner-types.workspace = true +asap-executor.workspace = true +sha2 = "0.10" serde.workspace = true serde_json.workspace = true thiserror.workspace = true @@ -21,3 +24,5 @@ snap = "1.1" [dev-dependencies] tower = { version = "0.5", features = ["util"] } http-body-util = "0.1" +# The tests find the frontier the way the Planner does; the runtime does not. +asap-executor = { workspace = true, features = ["test-support"] } diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs index 8d296d22..e770ff91 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -7,6 +7,7 @@ //! forwarded to Prometheus. pub mod remote_write; +pub mod sds; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; diff --git a/data_plane/src/sds.rs b/data_plane/src/sds.rs new file mode 100644 index 00000000..05ca7478 --- /dev/null +++ b/data_plane/src/sds.rs @@ -0,0 +1,252 @@ +//! Stored summaries (SDS): what the data plane stores for a plan's +//! ingestion-time outputs, and how a query finds them again. +//! +//! The Planner's selected DAG marks each output computed at ingestion time +//! (`MatAtIngestion`) and read at query time: the executor's materialization +//! frontier. The control plane installs the DAG with each `PaneChain` +//! expanded into its panes, so each such output reads one tumbling pane of +//! raw samples: +//! +//! ```text +//! frontier node <- ... <- TimeRange(width) <- TimeShift(offset)* <- Scan +//! ``` +//! +//! At evaluation time `T` it covers `[T - offset - width, T - offset)`. +//! Outputs that differ only in `offset` compute the same thing over different +//! panes, so they share one [`SummaryDefinition`]: the data plane builds each +//! pane of a definition once, and every output reads the pane it needs. + +use std::collections::{BTreeMap, HashMap}; + +use planner_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPDAGNode}; +use planner_types::ir::schema::Schema; +use planner_types::ir::{NonASAPOp, Operator}; +use planner_types::physical::Materialization; +use sha2::{Digest, Sha256}; + +/// Identifies what a stored summary computes, independent of node ids and of +/// which pane an output reads. Equal for equal computations across plans. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct DefinitionId(pub String); + +/// One computation the data plane maintains pane by pane. +#[derive(Debug, Clone, PartialEq)] +pub struct SummaryDefinition { + pub id: DefinitionId, + /// Pane width: each stored summary covers `[end - pane_ms, end)`. + pub pane_ms: i64, + /// The frontier output that builds a pane; any output of the definition + /// computes the same, and this is the first in node order. + pub builder: u64, + /// The raw input of the build: the `TimeRange` node fed one pane's samples. + pub raw_input: u64, + pub output_schema: Schema, +} + +/// How one frontier output reads its definition: at evaluation time `T`, the +/// pane ending at `T - offset_ms`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PaneBinding { + pub offset_ms: i64, +} + +/// The stored summaries of one plan. +#[derive(Debug, Clone, PartialEq, Default)] +pub struct Sds { + pub definitions: BTreeMap, + /// Every frontier output, with its definition and pane. + pub outputs: BTreeMap, +} + +#[derive(Debug, thiserror::Error, PartialEq)] +pub enum SdsError { + #[error("node {0} is not in the DAG")] + UnknownNode(u64), + #[error("ingestion-time output {node} does not read one pane of one Scan: {reason}")] + UnsupportedShape { node: u64, reason: String }, + /// The data plane keeps no query-time result across evaluations; it + /// declares `query_time_results_retained: false`, so the Planner should + /// not select such a plan. + #[error("node {0} is kept across evaluations (MatAtQueryKeptAcrossEvaluations), which the data plane does not support")] + KeptAcrossEvaluations(u64), +} + +impl Sds { + /// The definitions behind `frontier`, the DAG outputs computed at + /// ingestion time and read at query time. + pub fn from_dag(dag: &PhysicalASAPDAG, frontier: &[u64]) -> Result { + let nodes: HashMap = + dag.nodes.iter().map(|n| (n.id as u64, n)).collect(); + let mut inputs: HashMap> = HashMap::new(); + for edge in &dag.edges { + inputs + .entry(edge.consumer as u64) + .or_default() + .push(edge.producer as u64); + } + if let Some(node) = dag.nodes.iter().find(|n| { + matches!( + n.materialization, + Materialization::MatAtQueryKeptAcrossEvaluations(_) + ) + }) { + return Err(SdsError::KeptAcrossEvaluations(node.id as u64)); + } + let mut sds = Sds::default(); + let mut hashes = HashMap::new(); + for &node in frontier { + let (raw_input, pane_ms, offset_ms) = pane_of(node, &nodes, &inputs)?; + let id = DefinitionId(hex(&identity(node, &nodes, &inputs, &mut hashes)?)); + let output_schema = nodes[&node].output_schema.clone(); + let definition = + sds.definitions + .entry(id.clone()) + .or_insert_with(|| SummaryDefinition { + id: id.clone(), + pane_ms, + builder: node, + raw_input, + output_schema, + }); + if node < definition.builder { + definition.builder = node; + definition.raw_input = raw_input; + } + sds.outputs.insert(node, (id, PaneBinding { offset_ms })); + } + Ok(sds) + } +} + +/// `(raw input, pane width, offset)` of a frontier output: follow its single +/// inputs down to the `TimeRange`, then sum the `TimeShift`s down to the `Scan`. +fn pane_of( + node: u64, + nodes: &HashMap, + inputs: &HashMap>, +) -> Result<(u64, i64, i64), SdsError> { + let unsupported = |reason: &str| SdsError::UnsupportedShape { + node, + reason: reason.into(), + }; + let only_input = |id: u64| match inputs.get(&id).map(Vec::as_slice) { + Some([input]) => Ok(*input), + _ => Err(unsupported( + "every operator above the Scan must have one input", + )), + }; + let mut current = node; + let (raw_input, pane_ms) = loop { + match relational(nodes, current)? { + Some(NonASAPOp::TimeRange { range, .. }) => { + break (current, range.as_millis() as i64); + } + Some(NonASAPOp::TimeShift { .. }) => { + return Err(unsupported("a TimeShift above the TimeRange")); + } + Some(NonASAPOp::Scan { .. }) => return Err(unsupported("no TimeRange")), + _ => current = only_input(current)?, + } + }; + let mut offset_ms = 0; + current = only_input(raw_input)?; + loop { + match relational(nodes, current)? { + Some(NonASAPOp::TimeShift { shift, .. }) if shift.at.is_none() => { + offset_ms += shift.offset_ms; + current = only_input(current)?; + } + Some(NonASAPOp::Scan { .. }) => return Ok((raw_input, pane_ms, offset_ms)), + _ => { + return Err(unsupported( + "only offset TimeShifts between TimeRange and Scan", + )) + } + } + } +} + +fn relational<'a>( + nodes: &HashMap, + id: u64, +) -> Result>, SdsError> { + match &nodes.get(&id).ok_or(SdsError::UnknownNode(id))?.payload { + Operator::NonASAP(operator) => Ok(Some(operator)), + Operator::ASAP(_) => Ok(None), + } +} + +/// SHA-256 over a node's payload and output schema, then its inputs' +/// identities in edge order. The payload's input ids are zeroed, so the +/// identity does not depend on node numbering. An offset `TimeShift` has its +/// input's identity: the offset says which pane an output reads +/// ([`PaneBinding`]), not what it computes, so the panes of one computation +/// share an identity whether or not their range is shifted. +fn identity( + node: u64, + nodes: &HashMap, + inputs: &HashMap>, + hashes: &mut HashMap, +) -> Result<[u8; 32], SdsError> { + if let Some(hash) = hashes.get(&node) { + return Ok(*hash); + } + let dag_node = nodes.get(&node).ok_or(SdsError::UnknownNode(node))?; + if let Operator::NonASAP(NonASAPOp::TimeShift { shift, .. }) = &dag_node.payload { + if let (None, Some([input])) = (shift.at, inputs.get(&node).map(Vec::as_slice)) { + let hash = identity(*input, nodes, inputs, hashes)?; + hashes.insert(node, hash); + return Ok(hash); + } + } + let payload = dag_node.payload.map_children(|_| 0); + let mut hasher = Sha256::new(); + hasher.update(serde_json::to_vec(&(&payload, &dag_node.output_schema)).expect("serializes")); + for &input in inputs.get(&node).into_iter().flatten() { + hasher.update(identity(input, nodes, inputs, hashes)?); + } + let hash: [u8; 32] = hasher.finalize().into(); + hashes.insert(node, hash); + Ok(hash) +} + +fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() +} + +/// Built panes, by definition and pane end. A pane is stored once, when it +/// is complete, and never changes. +#[derive(Debug, Default)] +pub struct SummaryStore { + panes: BTreeMap<(DefinitionId, i64), S>, +} + +impl SummaryStore { + pub fn new() -> Self { + Self { + panes: BTreeMap::new(), + } + } + + pub fn insert(&mut self, definition: &DefinitionId, end_ms: i64, state: S) { + self.panes.insert((definition.clone(), end_ms), state); + } + + pub fn get(&self, definition: &DefinitionId, end_ms: i64) -> Option<&S> { + self.panes.get(&(definition.clone(), end_ms)) + } + + /// Drops the panes of `definition` that end at or before `end_ms`. + pub fn evict_through(&mut self, definition: &DefinitionId, end_ms: i64) { + self.panes + .retain(|(id, end), _| id != definition || *end > end_ms); + } + + pub fn len(&self) -> usize { + self.panes.len() + } + + pub fn is_empty(&self) -> bool { + self.panes.is_empty() + } +} diff --git a/data_plane/tests/sds.rs b/data_plane/tests/sds.rs new file mode 100644 index 00000000..a25a2d07 --- /dev/null +++ b/data_plane/tests/sds.rs @@ -0,0 +1,151 @@ +use std::path::Path; + +use asap_executor::physical_planner::frontier_from_timing; +use control_plane::{plan_workload, DeploymentInputs, InstalledPlan}; +use data_plane::sds::{DefinitionId, Sds, SdsError, SummaryStore}; +use planner_types::ir::physical_export::PhysicalASAPDAG; +use planner_types::ir::{NonASAPOp, Operator}; +use planner_types::physical::{Materialization, Medium}; + +/// The selected plan of a #509 example workload in `examples/`, for a +/// deployment that keeps raw data or not. +fn plan(example: &str, raw_data_retained: bool) -> InstalledPlan { + let path = Path::new(env!("CARGO_MANIFEST_DIR")) + .join(format!("../examples/planner-layering-{example}.json")); + let workload = serde_json::from_str(&std::fs::read_to_string(path).unwrap()).unwrap(); + let mut inputs = DeploymentInputs::default(); + inputs.deployment_capabilities.raw_data_retained = raw_data_retained; + plan_workload(&workload, &inputs).unwrap() +} + +fn sds_of(dag: &PhysicalASAPDAG) -> Result { + Sds::from_dag(dag, &frontier_from_timing(dag).unwrap()) +} + +/// Example 4b's six ingestion-time KLL panes are one definition with a +/// 10-minute pane; each output reads the pane `i * 10min` before evaluation. +#[test] +fn example4b_panes_share_one_definition() { + let plan = plan("example4b", false); + let sds = sds_of(&plan.dag.dag).unwrap(); + assert_eq!(sds.definitions.len(), 1); + let definition = sds.definitions.values().next().unwrap(); + assert_eq!(definition.pane_ms, 600_000); + let mut offsets: Vec = sds + .outputs + .values() + .map(|(id, binding)| { + assert_eq!(id, &definition.id); + binding.offset_ms + }) + .collect(); + offsets.sort(); + assert_eq!(offsets, (0..6).map(|i| i * 600_000).collect::>()); + let raw = plan + .dag + .dag + .nodes + .iter() + .find(|n| n.id as u64 == definition.raw_input) + .unwrap(); + assert!(matches!( + raw.payload, + Operator::NonASAP(NonASAPOp::TimeRange { .. }) + )); +} + +/// A plan computed entirely at query time stores nothing. +#[test] +fn query_time_plans_have_no_definitions() { + let plan = plan("example1", true); + assert_eq!(sds_of(&plan.dag.dag).unwrap(), Sds::default()); +} + +/// Identity depends on what is computed, not on how nodes are numbered. +#[test] +fn identity_ignores_node_numbering() { + let dag = plan("example4b", false).dag.dag; + let mut renumbered = dag.clone(); + let renumber = |id: &usize| id + 100; + for node in &mut renumbered.nodes { + node.id = renumber(&node.id); + node.payload = node.payload.map_children(renumber); + } + for edge in &mut renumbered.edges { + edge.producer = renumber(&edge.producer); + edge.consumer = renumber(&edge.consumer); + } + renumbered.roots = renumbered.roots.iter().map(renumber).collect(); + let ids = |sds: Sds| sds.definitions.into_keys().collect::>(); + assert_eq!( + ids(sds_of(&renumbered).unwrap()), + ids(sds_of(&dag).unwrap()) + ); +} + +/// The same computation over another metric is another definition. +#[test] +fn identity_depends_on_the_source() { + let dag = plan("example4b", false).dag.dag; + let other: PhysicalASAPDAG = serde_json::from_str( + &serde_json::to_string(&dag) + .unwrap() + .replace("latency_ms", "other"), + ) + .unwrap(); + assert_ne!( + sds_of(&other).unwrap().definitions.into_keys().next(), + sds_of(&dag).unwrap().definitions.into_keys().next() + ); +} + +/// An ingestion-time output must read a pane of raw samples. +#[test] +fn rejects_outputs_without_a_pane() { + let dag = plan("example4b", false).dag.dag; + let scan = dag + .nodes + .iter() + .find(|n| matches!(n.payload, Operator::NonASAP(NonASAPOp::Scan { .. }))) + .unwrap(); + let scan = scan.id as u64; + assert_eq!( + Sds::from_dag(&dag, &[scan]), + Err(SdsError::UnsupportedShape { + node: scan, + reason: "no TimeRange".into() + }) + ); +} + +/// A plan that keeps a query-time result across evaluations is not +/// installed: the data plane declares it cannot keep one. +#[test] +fn rejects_results_kept_across_evaluations() { + let mut dag = plan("example4b", true).dag.dag; + let node = &mut dag.nodes[0]; + node.materialization = Materialization::MatAtQueryKeptAcrossEvaluations(Medium::Memory); + let id = node.id as u64; + assert_eq!( + Sds::from_dag(&dag, &[]), + Err(SdsError::KeptAcrossEvaluations(id)) + ); +} + +/// Panes are found by definition and end, and evicted once no query needs them. +#[test] +fn store_finds_and_evicts_panes() { + let (a, b) = (DefinitionId("a".into()), DefinitionId("b".into())); + let mut store = SummaryStore::new(); + for end in [60, 120, 180] { + store.insert(&a, end, format!("a{end}")); + } + store.insert(&b, 60, "b60".to_string()); + assert_eq!(store.get(&a, 120).map(String::as_str), Some("a120")); + assert_eq!(store.get(&a, 90), None); + store.evict_through(&a, 120); + assert_eq!(store.get(&a, 120), None); + assert_eq!(store.get(&a, 180).map(String::as_str), Some("a180")); + assert_eq!(store.get(&b, 60).map(String::as_str), Some("b60")); + assert_eq!(store.len(), 2); +} diff --git a/examples/planner-layering-example3b.json b/examples/planner-layering-example3b.json new file mode 100644 index 00000000..c943e415 --- /dev/null +++ b/examples/planner-layering-example3b.json @@ -0,0 +1,44 @@ +{ + "query_workload": { + "language": "promql", + "query_batch": null, + "repeating_queries": [ + { + "query": "quantile_over_time(0.99, latency_ms[5m])", + "demand": { + "fixed_interval": 60000 + }, + "requirements": { + "accuracy": { + "accuracy_metric": "rank", + "max_error": 0.01, + "failure_probability": 0.01, + "accuracy_scope": "whole_result" + }, + "max_latency_ms": 200.0 + }, + "predictability": { + "predictable": { + "known_at": null + } + } + } + ] + }, + "data_workload": { + "arrival": { + "ingesting": { + "rows_per_second": 66666.66666666667 + } + }, + "sources": { + "latency_ms": { + "series_count": 1000000, + "metric_type": null, + "sample_interval": 15000, + "stream_norms": null, + "max_distinct_per_evaluation": null + } + } + } +} diff --git a/examples/planner-layering-example4b.json b/examples/planner-layering-example4b.json new file mode 100644 index 00000000..ab13a77f --- /dev/null +++ b/examples/planner-layering-example4b.json @@ -0,0 +1,44 @@ +{ + "query_workload": { + "language": "promql", + "query_batch": null, + "repeating_queries": [ + { + "query": "quantile_over_time(0.99, latency_ms[60m])", + "demand": { + "fixed_interval": 600000 + }, + "requirements": { + "accuracy": { + "accuracy_metric": "rank", + "max_error": 0.01, + "failure_probability": 0.01, + "accuracy_scope": "whole_result" + }, + "max_latency_ms": 200.0 + }, + "predictability": { + "predictable": { + "known_at": null + } + } + } + ] + }, + "data_workload": { + "arrival": { + "ingesting": { + "rows_per_second": 1000.0 + } + }, + "sources": { + "latency_ms": { + "series_count": 1000, + "metric_type": null, + "sample_interval": 1000, + "stream_norms": null, + "max_distinct_per_evaluation": null + } + } + } +} diff --git a/scripts/ci/README.md b/scripts/ci/README.md index 65569239..fc215ecc 100644 --- a/scripts/ci/README.md +++ b/scripts/ci/README.md @@ -2,7 +2,7 @@ The backend keeps reproducible Planner revisions in `Cargo.toml` and `Cargo.lock`. `Sync ASAPPlanner main` checks upstream main daily (or on manual dispatch), updates -all four Planner dependencies together, and opens or refreshes +all five Planner dependencies together, and opens or refreshes `automation/sync-asapplanner-main` against backend main. GitHub scheduled runs can be delayed. The existing `ASAP_CI_REPO_TOKEN` secret must have read access to private dependency diff --git a/scripts/ci/test_sync_planner.py b/scripts/ci/test_sync_planner.py index 51d8ba58..7c30eff5 100644 --- a/scripts/ci/test_sync_planner.py +++ b/scripts/ci/test_sync_planner.py @@ -34,7 +34,7 @@ class PlannerSyncTests(unittest.TestCase): manifest = ''.join( f'{name} = {{ git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "{"b" * 40}" }}\n' for name in ['planner-types', 'asap-frontend-promql', 'asap-logical-asap-optimization', - 'asap-plan-selection'] + 'asap-plan-selection', 'asap-executor'] ) def run_sync(self, manifest, revision=None, cargo_status=0): @@ -59,12 +59,12 @@ def test_updates_all_dependencies_and_lock_resolution(self): root, result = self.run_sync(self.manifest + other) self.assertEqual(result.returncode, 0, result.stderr) updated = (root / 'Cargo.toml').read_text() - self.assertEqual(updated.count(f'rev = "{self.revision}"'), 4) + self.assertEqual(updated.count(f'rev = "{self.revision}"'), 5) self.assertTrue(updated.endswith(other)) self.assertEqual((root / 'cargo-args').read_text().splitlines(), [ 'update', '-p', 'asap-types@0.1.0', '-p', 'asap-frontend-promql', '-p', 'asap-logical-asap-optimization', '-p', 'asap-plan-selection', - '--precise', self.revision]) + '-p', 'asap-executor', '--precise', self.revision]) def test_repository_manifest_updates_only_locked_packages(self): """Every pin in the real Cargo.toml is rewritten and every `-p` names a locked package.""" diff --git a/scripts/sync-asapplanner-main.sh b/scripts/sync-asapplanner-main.sh index c3f306f4..094ec6cf 100755 --- a/scripts/sync-asapplanner-main.sh +++ b/scripts/sync-asapplanner-main.sh @@ -20,8 +20,8 @@ pattern = re.compile( r'(git = "https://github.com/ProjectASAP/ASAPPlanner", rev = ")[0-9a-f]{40}("\s*})' ) updated, replacements = pattern.subn(rf"\g<1>{revision}\2", text) -if replacements != 4: - raise SystemExit(f"expected four ASAPPlanner dependencies, found {replacements}") +if replacements != 5: + raise SystemExit(f"expected five ASAPPlanner dependencies, found {replacements}") path.write_text(updated) PY @@ -30,6 +30,7 @@ cargo update \ -p asap-frontend-promql \ -p asap-logical-asap-optimization \ -p asap-plan-selection \ + -p asap-executor \ --precise "$planner_rev" echo "synchronized ASAPPlanner dependencies to $planner_rev"