From c9e05c70d918272a543c7b8d4056e5078715a9c7 Mon Sep 17 00:00:00 2001 From: Ahmad Sattar Date: Mon, 12 Oct 2026 11:26:09 +0200 Subject: [PATCH 1/3] Add the `hash-integrations-engine` crate with the DuckDB `Workspace` --- .clippy.toml | 2 +- Cargo.lock | 374 ++++++++++- Cargo.toml | 3 + libs/@local/integrations/README.md | 1 + libs/@local/integrations/engine/Cargo.toml | 33 + libs/@local/integrations/engine/LICENSE.md | 607 ++++++++++++++++++ libs/@local/integrations/engine/README.md | 5 + .../engine/docs/dependency-diagram.mmd | 14 + .../engine/docs/task-dependencies.json | 40 ++ libs/@local/integrations/engine/src/lib.rs | 24 + libs/@local/integrations/engine/src/sql.rs | 51 ++ .../integrations/engine/src/variant_name.rs | 107 +++ .../engine/src/workspace/connection.rs | 262 ++++++++ .../engine/src/workspace/error.rs | 270 ++++++++ .../engine/src/workspace/files.rs | 27 + .../engine/src/workspace/interrupt.rs | 180 ++++++ .../integrations/engine/src/workspace/mod.rs | 200 ++++++ .../engine/src/workspace/reply.rs | 36 ++ .../integrations/engine/src/workspace/rows.rs | 88 +++ .../engine/src/workspace/select.rs | 92 +++ .../integrations/engine/tests/workspace.rs | 434 +++++++++++++ libs/@local/integrations/engine/turbo.json | 35 + 22 files changed, 2864 insertions(+), 21 deletions(-) create mode 100644 libs/@local/integrations/engine/Cargo.toml create mode 100644 libs/@local/integrations/engine/LICENSE.md create mode 100644 libs/@local/integrations/engine/README.md create mode 100644 libs/@local/integrations/engine/docs/dependency-diagram.mmd create mode 100644 libs/@local/integrations/engine/docs/task-dependencies.json create mode 100644 libs/@local/integrations/engine/src/lib.rs create mode 100644 libs/@local/integrations/engine/src/sql.rs create mode 100644 libs/@local/integrations/engine/src/variant_name.rs create mode 100644 libs/@local/integrations/engine/src/workspace/connection.rs create mode 100644 libs/@local/integrations/engine/src/workspace/error.rs create mode 100644 libs/@local/integrations/engine/src/workspace/files.rs create mode 100644 libs/@local/integrations/engine/src/workspace/interrupt.rs create mode 100644 libs/@local/integrations/engine/src/workspace/mod.rs create mode 100644 libs/@local/integrations/engine/src/workspace/reply.rs create mode 100644 libs/@local/integrations/engine/src/workspace/rows.rs create mode 100644 libs/@local/integrations/engine/src/workspace/select.rs create mode 100644 libs/@local/integrations/engine/tests/workspace.rs create mode 100644 libs/@local/integrations/engine/turbo.json diff --git a/.clippy.toml b/.clippy.toml index de0264e814d..29d8a47b3a1 100644 --- a/.clippy.toml +++ b/.clippy.toml @@ -3,5 +3,5 @@ allow-indexing-slicing-in-tests = true allow-print-in-tests = true allow-renamed-params-for = ["core::fmt::Debug", "core::fmt::Display", "core::fmt::LowerExp", "core::fmt::LowerHex", "core::fmt::UpperHex", "core::fmt::Pointer", "futures_sink::Sink", "serde::de::Visitor", ".."] avoid-breaking-exported-api = false -doc-valid-idents = ["BlockProtocol", "HaRPC", "HashQL", "MinIO", "OpenAI", "OpenAPI", "PostgreSQL", "OAuth2", ".."] +doc-valid-idents = ["BlockProtocol", "DuckDB", "HaRPC", "HashQL", "MinIO", "OpenAI", "OpenAPI", "PostgreSQL", "OAuth2", ".."] suppress-restriction-lint-in-const = true diff --git a/Cargo.lock b/Cargo.lock index ff321477369..08a2349266a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -65,6 +65,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" dependencies = [ "cfg-if", + "const-random", "getrandom 0.3.4", "once_cell", "serde", @@ -343,6 +344,9 @@ name = "arbitrary" version = "1.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3d036a3c4ab069c7b410a2ce876bd74808d2d0888a82667669f8e783a898bf1" +dependencies = [ + "derive_arbitrary", +] [[package]] name = "arc-swap" @@ -403,6 +407,169 @@ version = "0.7.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f02882884d3e1bc524fb12c79f107f6ad0e1cfd498c536ffb494301740995dfe" +[[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 0.17.1", + "num-complex 0.4.6", + "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 0.22.1", + "chrono", + "comfy-table", + "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" +dependencies = [ + "bitflags 2.13.0", +] + +[[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 = "as-slice" version = "0.2.1" @@ -520,7 +687,7 @@ dependencies = [ "futures-lite", "parking", "polling", - "rustix", + "rustix 1.1.4", "slab", "windows-sys 0.61.2", ] @@ -593,6 +760,15 @@ dependencies = [ "pin-project-lite", ] +[[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" version = "0.6.1" @@ -1886,7 +2062,7 @@ dependencies = [ "serde", "serde_json", "serde_rusqlite", - "strum", + "strum 0.28.0", "tempfile", "thiserror 2.0.18", ] @@ -2762,6 +2938,17 @@ dependencies = [ "memchr", ] +[[package]] +name = "comfy-table" +version = "7.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a65ebfec4fb190b6f90e944a817d60499ee0744e582530e2c9900a22e591d9a" +dependencies = [ + "crossterm 0.28.1", + "unicode-segmentation", + "unicode-width 0.2.2", +] + [[package]] name = "common-path" version = "1.0.0" @@ -3115,6 +3302,19 @@ version = "0.8.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" +[[package]] +name = "crossterm" +version = "0.28.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "829d955a0bb380ef178a640b91779e3987da38c9aea133b20614cfed8cdea9c6" +dependencies = [ + "bitflags 2.13.0", + "crossterm_winapi", + "parking_lot", + "rustix 0.38.44", + "winapi", +] + [[package]] name = "crossterm" version = "0.29.0" @@ -3127,7 +3327,7 @@ dependencies = [ "document-features", "mio", "parking_lot", - "rustix", + "rustix 1.1.4", "signal-hook 0.3.18", "signal-hook-mio", "winapi", @@ -4050,6 +4250,17 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "derive_arbitrary" +version = "1.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e567bd82dcff979e4b03460c307b3cdc9e96fde3d73bed1496d2bc75d9dd62a" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "derive_builder" version = "0.20.2" @@ -4275,6 +4486,23 @@ version = "0.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7454e41ff9012c00d53cf7f475c5e3afa3b91b7c90568495495e8d9bf47a1055" +[[package]] +name = "duckdb" +version = "1.10505.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "970e05eedd3f55c435194d9104f90a9b4a79a80d6e73251bc9ff43e178130c4e" +dependencies = [ + "arrow", + "cast", + "comfy-table", + "fallible-iterator 0.3.0", + "fallible-streaming-iterator", + "hashlink 0.10.0", + "libduckdb-sys", + "num-integer", + "strum 0.27.2", +] + [[package]] name = "dunce" version = "1.0.5" @@ -5018,6 +5246,7 @@ checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" dependencies = [ "crc32fast", "miniz_oxide", + "zlib-rs", ] [[package]] @@ -5251,7 +5480,7 @@ version = "0.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8640e34b88f7652208ce9e88b1a37a2ae95227d84abec377ccd3c5cfeb141ed4" dependencies = [ - "rustix", + "rustix 1.1.4", "windows-sys 0.59.0", ] @@ -6814,6 +7043,22 @@ dependencies = [ "type-system", ] +[[package]] +name = "hash-integrations-engine" +version = "0.0.0" +dependencies = [ + "derive_more", + "duckdb", + "error-stack", + "pretty_assertions", + "proptest", + "serde_json", + "simple-mermaid", + "tempfile", + "tokio", + "tracing", +] + [[package]] name = "hash-middleware" version = "0.0.0" @@ -8218,7 +8463,7 @@ dependencies = [ "regex", "serde", "serde_json", - "strum", + "strum 0.28.0", "unicode-general-category", "uuid-simd", ] @@ -8489,6 +8734,23 @@ version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +[[package]] +name = "libduckdb-sys" +version = "1.10505.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6cb514dab5e271e849235c1cb98bd65a2ae107fbd619a6740219319c54a71d95" +dependencies = [ + "cc", + "flate2", + "pkg-config", + "serde", + "serde_json", + "tar", + "ureq 3.3.0", + "vcpkg", + "zip 6.0.0", +] + [[package]] name = "libfuzzer-sys" version = "0.4.13" @@ -8962,6 +9224,12 @@ dependencies = [ "text-size", ] +[[package]] +name = "linux-raw-sys" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d26c52dbd32dccf2d10cac7725f8eae5296885fb5703b261f7d0a0739ec807ab" + [[package]] name = "linux-raw-sys" version = "0.12.1" @@ -11312,7 +11580,7 @@ dependencies = [ "concurrent-queue", "hermit-abi", "pin-project-lite", - "rustix", + "rustix 1.1.4", "windows-sys 0.61.2", ] @@ -12235,7 +12503,7 @@ dependencies = [ "lru", "palette", "serde", - "strum", + "strum 0.28.0", "thiserror 2.0.18", "unicode-segmentation", "unicode-truncate", @@ -12249,7 +12517,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "567584a3b0e6a8203c23de40b4861497266725eb5363dbfd18a1edd603cca9f0" dependencies = [ "cfg-if", - "crossterm", + "crossterm 0.29.0", "instability", "ratatui-core", ] @@ -12299,7 +12567,7 @@ dependencies = [ "line-clipping", "ratatui-core", "serde", - "strum", + "strum 0.28.0", "time", "unicode-segmentation", "unicode-width 0.2.2", @@ -13020,6 +13288,19 @@ dependencies = [ "nom 7.1.3", ] +[[package]] +name = "rustix" +version = "0.38.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fdb5bc1ae2baa591800df16c9ca78619bf65c0488b41b96ccec5d11220d8c154" +dependencies = [ + "bitflags 2.13.0", + "errno", + "libc", + "linux-raw-sys 0.4.15", + "windows-sys 0.59.0", +] + [[package]] name = "rustix" version = "1.1.4" @@ -13029,7 +13310,7 @@ dependencies = [ "bitflags 2.13.0", "errno", "libc", - "linux-raw-sys", + "linux-raw-sys 0.12.1", "windows-sys 0.61.2", ] @@ -14215,8 +14496,8 @@ dependencies = [ "smol_str", "sqruff-lib-core", "sqruff-lib-dialects", - "strum", - "strum_macros", + "strum 0.28.0", + "strum_macros 0.28.0", "walkdir", ] @@ -14236,8 +14517,8 @@ dependencies = [ "pretty_assertions", "regex-automata", "smol_str", - "strum", - "strum_macros", + "strum 0.28.0", + "strum_macros 0.28.0", "thiserror 2.0.18", ] @@ -14251,7 +14532,7 @@ dependencies = [ "itertools 0.14.0", "serde_yaml", "sqruff-lib-core", - "strum", + "strum 0.28.0", ] [[package]] @@ -14372,13 +14653,34 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "strum" +version = "0.27.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af23d6f6c1a224baef9d3f61e287d2761385a5b88fdab4eb4c6f11aeb54c4bcf" +dependencies = [ + "strum_macros 0.27.2", +] + [[package]] name = "strum" version = "0.28.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9628de9b8791db39ceda2b119bbe13134770b56c138ec1d3af810d045c04f9bd" dependencies = [ - "strum_macros", + "strum_macros 0.28.0", +] + +[[package]] +name = "strum_macros" +version = "0.27.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7695ce3845ea4b33927c055a39dc438a45b059f7c1b3d91d38d10355fb8cbca7" +dependencies = [ + "heck 0.5.0", + "proc-macro2", + "quote", + "syn 2.0.118", ] [[package]] @@ -14648,7 +14950,7 @@ dependencies = [ "fastrand", "getrandom 0.4.3", "once_cell", - "rustix", + "rustix 1.1.4", "windows-sys 0.61.2", ] @@ -14790,7 +15092,7 @@ checksum = "9048a889effe34a5cddee0af7f53285198b16dca3be510858d38dfdb3e62a04e" dependencies = [ "bitflags 2.13.0", "parking_lot", - "rustix", + "rustix 1.1.4", "signal-hook 0.3.18", "windows-sys 0.61.2", ] @@ -14801,7 +15103,7 @@ version = "0.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "230a1b821ccbd75b185820a1f1ff7b14d21da1e442e22c0863ea5f08771a8874" dependencies = [ - "rustix", + "rustix 1.1.4", "windows-sys 0.61.2", ] @@ -17495,7 +17797,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32e45ad4206f6d2479085147f02bc2ef834ac85886624a23575ae137c8aa8156" dependencies = [ "libc", - "rustix", + "rustix 1.1.4", ] [[package]] @@ -17730,6 +18032,20 @@ dependencies = [ "zstd 0.11.2+zstd.1.5.2", ] +[[package]] +name = "zip" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eb2a05c7c36fde6c09b08576c9f7fb4cda705990f73b58fe011abf7dfb24168b" +dependencies = [ + "arbitrary", + "crc32fast", + "flate2", + "indexmap 2.14.0", + "memchr", + "zopfli", +] + [[package]] name = "zip" version = "7.2.0" @@ -17742,12 +18058,30 @@ dependencies = [ "typed-path", ] +[[package]] +name = "zlib-rs" +version = "0.6.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b268e58e7c693d7c271f93ffc4ba3b380412554231c85bf61ca7af91042a4112" + [[package]] name = "zmij" version = "1.0.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" +[[package]] +name = "zopfli" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f05cd8797d63865425ff89b5c4a48804f35ba0ce8d125800027ad6017d2b5249" +dependencies = [ + "bumpalo", + "crc32fast", + "log", + "simd-adler32", +] + [[package]] name = "zstd" version = "0.11.2+zstd.1.5.2" diff --git a/Cargo.toml b/Cargo.toml index 7ad6f6cd81d..6fc365137d1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -45,6 +45,7 @@ members = [ "libs/@local/hashql/mir", "libs/@local/hashql/syntax-jexpr", "libs/@local/integrations/definition", + "libs/@local/integrations/engine", "libs/@local/middleware", "libs/@local/repo-chores/rust", "libs/@local/secret", @@ -109,6 +110,7 @@ hash-graph-type-fetcher.path = "libs/@local/graph/type-fetcher" hash-graph-types.path = "libs/@local/graph/types" hash-graph-validation.path = "libs/@local/graph/validation" hash-integrations-definition.path = "libs/@local/integrations/definition" +hash-integrations-engine.path = "libs/@local/integrations/engine" hash-middleware.path = "libs/@local/middleware" hash-repo-chores.path = "libs/@local/repo-chores/rust" hash-secret.path = "libs/@local/secret" @@ -180,6 +182,7 @@ deadpool = { version = "0.12.3", default-features = fa derive-where = { version = "1.6.0", default-features = false, features = ["nightly"] } derive_more = { version = "2.0.1", default-features = false } dotenv-flow = { version = "0.16.2", default-features = false } +duckdb = { version = "1.10505.0", default-features = false } either = { version = "1.15.0", default-features = false } email_address = { version = "0.2.9", default-features = false } ena = { version = "0.14.3", default-features = false } diff --git a/libs/@local/integrations/README.md b/libs/@local/integrations/README.md index 91c0cc4f8ea..efb873612cf 100644 --- a/libs/@local/integrations/README.md +++ b/libs/@local/integrations/README.md @@ -3,3 +3,4 @@ Library crates for [`apps/hash-integrations`](../../../apps/hash-integrations). Each crate has a `hash-integrations-` prefix: - `definition`: Holds the checked model of an integration definition. +- `engine`: Runs an integration's pipelines in DuckDB. diff --git a/libs/@local/integrations/engine/Cargo.toml b/libs/@local/integrations/engine/Cargo.toml new file mode 100644 index 00000000000..77d8d2fb781 --- /dev/null +++ b/libs/@local/integrations/engine/Cargo.toml @@ -0,0 +1,33 @@ +[package] +name = "hash-integrations-engine" +description = "Runs integration pipelines in DuckDB" +version.workspace = true +edition.workspace = true +license.workspace = true +publish.workspace = true +authors.workspace = true + +[dependencies] +# Public workspace dependencies +error-stack = { workspace = true, public = true, features = ["std"] } + +# Public third-party dependencies +duckdb = { workspace = true, public = true, features = ["bundled", "json"] } + +# Private workspace dependencies + +# Private third-party dependencies +derive_more = { workspace = true, features = ["display", "error"] } +serde_json = { workspace = true } +simple-mermaid = { workspace = true } +tokio = { workspace = true, features = ["sync"] } +tracing = { workspace = true, features = ["attributes"] } + +[dev-dependencies] +pretty_assertions = { workspace = true } +proptest = { workspace = true, features = ["attr-macro", "std"] } +tempfile = { workspace = true } +tokio = { workspace = true, features = ["macros", "rt", "time"] } + +[lints] +workspace = true diff --git a/libs/@local/integrations/engine/LICENSE.md b/libs/@local/integrations/engine/LICENSE.md new file mode 100644 index 00000000000..c7d627721e2 --- /dev/null +++ b/libs/@local/integrations/engine/LICENSE.md @@ -0,0 +1,607 @@ +GNU Affero General Public License +================================= + +_Version 3, 19 November 2007_ +_Copyright © 2007 Free Software Foundation, Inc. <>_ + +Everyone is permitted to copy and distribute verbatim copies +of this license document, but changing it is not allowed. + +## Preamble + +The GNU Affero General Public License is a free, copyleft license for +software and other kinds of works, specifically designed to ensure +cooperation with the community in the case of network server software. + +The licenses for most software and other practical works are designed +to take away your freedom to share and change the works. By contrast, +our General Public Licenses are intended to guarantee your freedom to +share and change all versions of a program--to make sure it remains free +software for all its users. + +When we speak of free software, we are referring to freedom, not +price. Our General Public Licenses are designed to make sure that you +have the freedom to distribute copies of free software (and charge for +them if you wish), that you receive source code or can get it if you +want it, that you can change the software or use pieces of it in new +free programs, and that you know you can do these things. + +Developers that use our General Public Licenses protect your rights +with two steps: **(1)** assert copyright on the software, and **(2)** offer +you this License which gives you legal permission to copy, distribute +and/or modify the software. + +A secondary benefit of defending all users' freedom is that +improvements made in alternate versions of the program, if they +receive widespread use, become available for other developers to +incorporate. Many developers of free software are heartened and +encouraged by the resulting cooperation. However, in the case of +software used on network servers, this result may fail to come about. +The GNU General Public License permits making a modified version and +letting the public access it on a server without ever releasing its +source code to the public. + +The GNU Affero General Public License is designed specifically to +ensure that, in such cases, the modified source code becomes available +to the community. It requires the operator of a network server to +provide the source code of the modified version running there to the +users of that server. Therefore, public use of a modified version, on +a publicly accessible server, gives the public access to the source +code of the modified version. + +An older license, called the Affero General Public License and +published by Affero, was designed to accomplish similar goals. This is +a different license, not a version of the Affero GPL, but Affero has +released a new version of the Affero GPL which permits relicensing under +this license. + +The precise terms and conditions for copying, distribution and +modification follow. + +## TERMS AND CONDITIONS + +### 0. Definitions + +“This License” refers to version 3 of the GNU Affero General Public License. + +“Copyright” also means copyright-like laws that apply to other kinds of +works, such as semiconductor masks. + +“The Program” refers to any copyrightable work licensed under this +License. Each licensee is addressed as “you”. “Licensees” and +“recipients” may be individuals or organizations. + +To “modify” a work means to copy from or adapt all or part of the work +in a fashion requiring copyright permission, other than the making of an +exact copy. The resulting work is called a “modified version” of the +earlier work or a work “based on” the earlier work. + +A “covered work” means either the unmodified Program or a work based +on the Program. + +To “propagate” a work means to do anything with it that, without +permission, would make you directly or secondarily liable for +infringement under applicable copyright law, except executing it on a +computer or modifying a private copy. Propagation includes copying, +distribution (with or without modification), making available to the +public, and in some countries other activities as well. + +To “convey” a work means any kind of propagation that enables other +parties to make or receive copies. Mere interaction with a user through +a computer network, with no transfer of a copy, is not conveying. + +An interactive user interface displays “Appropriate Legal Notices” +to the extent that it includes a convenient and prominently visible +feature that **(1)** displays an appropriate copyright notice, and **(2)** +tells the user that there is no warranty for the work (except to the +extent that warranties are provided), that licensees may convey the +work under this License, and how to view a copy of this License. If +the interface presents a list of user commands or options, such as a +menu, a prominent item in the list meets this criterion. + +### 1. Source Code + +The “source code” for a work means the preferred form of the work +for making modifications to it. “Object code” means any non-source +form of a work. + +A “Standard Interface” means an interface that either is an official +standard defined by a recognized standards body, or, in the case of +interfaces specified for a particular programming language, one that +is widely used among developers working in that language. + +The “System Libraries” of an executable work include anything, other +than the work as a whole, that **(a)** is included in the normal form of +packaging a Major Component, but which is not part of that Major +Component, and **(b)** serves only to enable use of the work with that +Major Component, or to implement a Standard Interface for which an +implementation is available to the public in source code form. A +“Major Component”, in this context, means a major essential component +(kernel, window system, and so on) of the specific operating system +(if any) on which the executable work runs, or a compiler used to +produce the work, or an object code interpreter used to run it. + +The “Corresponding Source” for a work in object code form means all +the source code needed to generate, install, and (for an executable +work) run the object code and to modify the work, including scripts to +control those activities. However, it does not include the work's +System Libraries, or general-purpose tools or generally available free +programs which are used unmodified in performing those activities but +which are not part of the work. For example, Corresponding Source +includes interface definition files associated with source files for +the work, and the source code for shared libraries and dynamically +linked subprograms that the work is specifically designed to require, +such as by intimate data communication or control flow between those +subprograms and other parts of the work. + +The Corresponding Source need not include anything that users +can regenerate automatically from other parts of the Corresponding +Source. + +The Corresponding Source for a work in source code form is that +same work. + +### 2. Basic Permissions + +All rights granted under this License are granted for the term of +copyright on the Program, and are irrevocable provided the stated +conditions are met. This License explicitly affirms your unlimited +permission to run the unmodified Program. The output from running a +covered work is covered by this License only if the output, given its +content, constitutes a covered work. This License acknowledges your +rights of fair use or other equivalent, as provided by copyright law. + +You may make, run and propagate covered works that you do not +convey, without conditions so long as your license otherwise remains +in force. You may convey covered works to others for the sole purpose +of having them make modifications exclusively for you, or provide you +with facilities for running those works, provided that you comply with +the terms of this License in conveying all material for which you do +not control copyright. Those thus making or running the covered works +for you must do so exclusively on your behalf, under your direction +and control, on terms that prohibit them from making any copies of +your copyrighted material outside their relationship with you. + +Conveying under any other circumstances is permitted solely under +the conditions stated below. Sublicensing is not allowed; section 10 +makes it unnecessary. + +### 3. Protecting Users' Legal Rights From Anti-Circumvention Law + +No covered work shall be deemed part of an effective technological +measure under any applicable law fulfilling obligations under article +11 of the WIPO copyright treaty adopted on 20 December 1996, or +similar laws prohibiting or restricting circumvention of such +measures. + +When you convey a covered work, you waive any legal power to forbid +circumvention of technological measures to the extent such circumvention +is effected by exercising rights under this License with respect to +the covered work, and you disclaim any intention to limit operation or +modification of the work as a means of enforcing, against the work's +users, your or third parties' legal rights to forbid circumvention of +technological measures. + +### 4. Conveying Verbatim Copies + +You may convey verbatim copies of the Program's source code as you +receive it, in any medium, provided that you conspicuously and +appropriately publish on each copy an appropriate copyright notice; +keep intact all notices stating that this License and any +non-permissive terms added in accord with section 7 apply to the code; +keep intact all notices of the absence of any warranty; and give all +recipients a copy of this License along with the Program. + +You may charge any price or no price for each copy that you convey, +and you may offer support or warranty protection for a fee. + +### 5. Conveying Modified Source Versions + +You may convey a work based on the Program, or the modifications to +produce it from the Program, in the form of source code under the +terms of section 4, provided that you also meet all of these conditions: + +* **a)** The work must carry prominent notices stating that you modified +it, and giving a relevant date. +* **b)** The work must carry prominent notices stating that it is +released under this License and any conditions added under section 7. +This requirement modifies the requirement in section 4 to +“keep intact all notices”. +* **c)** You must license the entire work, as a whole, under this +License to anyone who comes into possession of a copy. This +License will therefore apply, along with any applicable section 7 +additional terms, to the whole of the work, and all its parts, +regardless of how they are packaged. This License gives no +permission to license the work in any other way, but it does not +invalidate such permission if you have separately received it. +* **d)** If the work has interactive user interfaces, each must display +Appropriate Legal Notices; however, if the Program has interactive +interfaces that do not display Appropriate Legal Notices, your +work need not make them do so. + +A compilation of a covered work with other separate and independent +works, which are not by their nature extensions of the covered work, +and which are not combined with it such as to form a larger program, +in or on a volume of a storage or distribution medium, is called an +“aggregate” if the compilation and its resulting copyright are not +used to limit the access or legal rights of the compilation's users +beyond what the individual works permit. Inclusion of a covered work +in an aggregate does not cause this License to apply to the other +parts of the aggregate. + +### 6. Conveying Non-Source Forms + +You may convey a covered work in object code form under the terms +of sections 4 and 5, provided that you also convey the +machine-readable Corresponding Source under the terms of this License, +in one of these ways: + +* **a)** Convey the object code in, or embodied in, a physical product +(including a physical distribution medium), accompanied by the +Corresponding Source fixed on a durable physical medium +customarily used for software interchange. +* **b)** Convey the object code in, or embodied in, a physical product +(including a physical distribution medium), accompanied by a +written offer, valid for at least three years and valid for as +long as you offer spare parts or customer support for that product +model, to give anyone who possesses the object code either **(1)** a +copy of the Corresponding Source for all the software in the +product that is covered by this License, on a durable physical +medium customarily used for software interchange, for a price no +more than your reasonable cost of physically performing this +conveying of source, or **(2)** access to copy the +Corresponding Source from a network server at no charge. +* **c)** Convey individual copies of the object code with a copy of the +written offer to provide the Corresponding Source. This +alternative is allowed only occasionally and noncommercially, and +only if you received the object code with such an offer, in accord +with subsection 6b. +* **d)** Convey the object code by offering access from a designated +place (gratis or for a charge), and offer equivalent access to the +Corresponding Source in the same way through the same place at no +further charge. You need not require recipients to copy the +Corresponding Source along with the object code. If the place to +copy the object code is a network server, the Corresponding Source +may be on a different server (operated by you or a third party) +that supports equivalent copying facilities, provided you maintain +clear directions next to the object code saying where to find the +Corresponding Source. Regardless of what server hosts the +Corresponding Source, you remain obligated to ensure that it is +available for as long as needed to satisfy these requirements. +* **e)** Convey the object code using peer-to-peer transmission, provided +you inform other peers where the object code and Corresponding +Source of the work are being offered to the general public at no +charge under subsection 6d. + +A separable portion of the object code, whose source code is excluded +from the Corresponding Source as a System Library, need not be +included in conveying the object code work. + +A “User Product” is either **(1)** a “consumer product”, which means any +tangible personal property which is normally used for personal, family, +or household purposes, or **(2)** anything designed or sold for incorporation +into a dwelling. In determining whether a product is a consumer product, +doubtful cases shall be resolved in favor of coverage. For a particular +product received by a particular user, “normally used” refers to a +typical or common use of that class of product, regardless of the status +of the particular user or of the way in which the particular user +actually uses, or expects or is expected to use, the product. A product +is a consumer product regardless of whether the product has substantial +commercial, industrial or non-consumer uses, unless such uses represent +the only significant mode of use of the product. + +“Installation Information” for a User Product means any methods, +procedures, authorization keys, or other information required to install +and execute modified versions of a covered work in that User Product from +a modified version of its Corresponding Source. The information must +suffice to ensure that the continued functioning of the modified object +code is in no case prevented or interfered with solely because +modification has been made. + +If you convey an object code work under this section in, or with, or +specifically for use in, a User Product, and the conveying occurs as +part of a transaction in which the right of possession and use of the +User Product is transferred to the recipient in perpetuity or for a +fixed term (regardless of how the transaction is characterized), the +Corresponding Source conveyed under this section must be accompanied +by the Installation Information. But this requirement does not apply +if neither you nor any third party retains the ability to install +modified object code on the User Product (for example, the work has +been installed in ROM). + +The requirement to provide Installation Information does not include a +requirement to continue to provide support service, warranty, or updates +for a work that has been modified or installed by the recipient, or for +the User Product in which it has been modified or installed. Access to a +network may be denied when the modification itself materially and +adversely affects the operation of the network or violates the rules and +protocols for communication across the network. + +Corresponding Source conveyed, and Installation Information provided, +in accord with this section must be in a format that is publicly +documented (and with an implementation available to the public in +source code form), and must require no special password or key for +unpacking, reading or copying. + +### 7. Additional Terms + +“Additional permissions” are terms that supplement the terms of this +License by making exceptions from one or more of its conditions. +Additional permissions that are applicable to the entire Program shall +be treated as though they were included in this License, to the extent +that they are valid under applicable law. If additional permissions +apply only to part of the Program, that part may be used separately +under those permissions, but the entire Program remains governed by +this License without regard to the additional permissions. + +When you convey a copy of a covered work, you may at your option +remove any additional permissions from that copy, or from any part of +it. (Additional permissions may be written to require their own +removal in certain cases when you modify the work.) You may place +additional permissions on material, added by you to a covered work, +for which you have or can give appropriate copyright permission. + +Notwithstanding any other provision of this License, for material you +add to a covered work, you may (if authorized by the copyright holders of +that material) supplement the terms of this License with terms: + +* **a)** Disclaiming warranty or limiting liability differently from the +terms of sections 15 and 16 of this License; or +* **b)** Requiring preservation of specified reasonable legal notices or +author attributions in that material or in the Appropriate Legal +Notices displayed by works containing it; or +* **c)** Prohibiting misrepresentation of the origin of that material, or +requiring that modified versions of such material be marked in +reasonable ways as different from the original version; or +* **d)** Limiting the use for publicity purposes of names of licensors or +authors of the material; or +* **e)** Declining to grant rights under trademark law for use of some +trade names, trademarks, or service marks; or +* **f)** Requiring indemnification of licensors and authors of that +material by anyone who conveys the material (or modified versions of +it) with contractual assumptions of liability to the recipient, for +any liability that these contractual assumptions directly impose on +those licensors and authors. + +All other non-permissive additional terms are considered “further +restrictions” within the meaning of section 10. If the Program as you +received it, or any part of it, contains a notice stating that it is +governed by this License along with a term that is a further +restriction, you may remove that term. If a license document contains +a further restriction but permits relicensing or conveying under this +License, you may add to a covered work material governed by the terms +of that license document, provided that the further restriction does +not survive such relicensing or conveying. + +If you add terms to a covered work in accord with this section, you +must place, in the relevant source files, a statement of the +additional terms that apply to those files, or a notice indicating +where to find the applicable terms. + +Additional terms, permissive or non-permissive, may be stated in the +form of a separately written license, or stated as exceptions; +the above requirements apply either way. + +### 8. Termination + +You may not propagate or modify a covered work except as expressly +provided under this License. Any attempt otherwise to propagate or +modify it is void, and will automatically terminate your rights under +this License (including any patent licenses granted under the third +paragraph of section 11). + +However, if you cease all violation of this License, then your +license from a particular copyright holder is reinstated **(a)** +provisionally, unless and until the copyright holder explicitly and +finally terminates your license, and **(b)** permanently, if the copyright +holder fails to notify you of the violation by some reasonable means +prior to 60 days after the cessation. + +Moreover, your license from a particular copyright holder is +reinstated permanently if the copyright holder notifies you of the +violation by some reasonable means, this is the first time you have +received notice of violation of this License (for any work) from that +copyright holder, and you cure the violation prior to 30 days after +your receipt of the notice. + +Termination of your rights under this section does not terminate the +licenses of parties who have received copies or rights from you under +this License. If your rights have been terminated and not permanently +reinstated, you do not qualify to receive new licenses for the same +material under section 10. + +### 9. Acceptance Not Required for Having Copies + +You are not required to accept this License in order to receive or +run a copy of the Program. Ancillary propagation of a covered work +occurring solely as a consequence of using peer-to-peer transmission +to receive a copy likewise does not require acceptance. However, +nothing other than this License grants you permission to propagate or +modify any covered work. These actions infringe copyright if you do +not accept this License. Therefore, by modifying or propagating a +covered work, you indicate your acceptance of this License to do so. + +### 10. Automatic Licensing of Downstream Recipients + +Each time you convey a covered work, the recipient automatically +receives a license from the original licensors, to run, modify and +propagate that work, subject to this License. You are not responsible +for enforcing compliance by third parties with this License. + +An “entity transaction” is a transaction transferring control of an +organization, or substantially all assets of one, or subdividing an +organization, or merging organizations. If propagation of a covered +work results from an entity transaction, each party to that +transaction who receives a copy of the work also receives whatever +licenses to the work the party's predecessor in interest had or could +give under the previous paragraph, plus a right to possession of the +Corresponding Source of the work from the predecessor in interest, if +the predecessor has it or can get it with reasonable efforts. + +You may not impose any further restrictions on the exercise of the +rights granted or affirmed under this License. For example, you may +not impose a license fee, royalty, or other charge for exercise of +rights granted under this License, and you may not initiate litigation +(including a cross-claim or counterclaim in a lawsuit) alleging that +any patent claim is infringed by making, using, selling, offering for +sale, or importing the Program or any portion of it. + +### 11. Patents + +A “contributor” is a copyright holder who authorizes use under this +License of the Program or a work on which the Program is based. The +work thus licensed is called the contributor's “contributor version”. + +A contributor's “essential patent claims” are all patent claims +owned or controlled by the contributor, whether already acquired or +hereafter acquired, that would be infringed by some manner, permitted +by this License, of making, using, or selling its contributor version, +but do not include claims that would be infringed only as a +consequence of further modification of the contributor version. For +purposes of this definition, “control” includes the right to grant +patent sublicenses in a manner consistent with the requirements of +this License. + +Each contributor grants you a non-exclusive, worldwide, royalty-free +patent license under the contributor's essential patent claims, to +make, use, sell, offer for sale, import and otherwise run, modify and +propagate the contents of its contributor version. + +In the following three paragraphs, a “patent license” is any express +agreement or commitment, however denominated, not to enforce a patent +(such as an express permission to practice a patent or covenant not to +sue for patent infringement). To “grant” such a patent license to a +party means to make such an agreement or commitment not to enforce a +patent against the party. + +If you convey a covered work, knowingly relying on a patent license, +and the Corresponding Source of the work is not available for anyone +to copy, free of charge and under the terms of this License, through a +publicly available network server or other readily accessible means, +then you must either **(1)** cause the Corresponding Source to be so +available, or **(2)** arrange to deprive yourself of the benefit of the +patent license for this particular work, or **(3)** arrange, in a manner +consistent with the requirements of this License, to extend the patent +license to downstream recipients. “Knowingly relying” means you have +actual knowledge that, but for the patent license, your conveying the +covered work in a country, or your recipient's use of the covered work +in a country, would infringe one or more identifiable patents in that +country that you have reason to believe are valid. + +If, pursuant to or in connection with a single transaction or +arrangement, you convey, or propagate by procuring conveyance of, a +covered work, and grant a patent license to some of the parties +receiving the covered work authorizing them to use, propagate, modify +or convey a specific copy of the covered work, then the patent license +you grant is automatically extended to all recipients of the covered +work and works based on it. + +A patent license is “discriminatory” if it does not include within +the scope of its coverage, prohibits the exercise of, or is +conditioned on the non-exercise of one or more of the rights that are +specifically granted under this License. You may not convey a covered +work if you are a party to an arrangement with a third party that is +in the business of distributing software, under which you make payment +to the third party based on the extent of your activity of conveying +the work, and under which the third party grants, to any of the +parties who would receive the covered work from you, a discriminatory +patent license **(a)** in connection with copies of the covered work +conveyed by you (or copies made from those copies), or **(b)** primarily +for and in connection with specific products or compilations that +contain the covered work, unless you entered into that arrangement, +or that patent license was granted, prior to 28 March 2007. + +Nothing in this License shall be construed as excluding or limiting +any implied license or other defenses to infringement that may +otherwise be available to you under applicable patent law. + +### 12. No Surrender of Others' Freedom + +If conditions are imposed on you (whether by court order, agreement or +otherwise) that contradict the conditions of this License, they do not +excuse you from the conditions of this License. If you cannot convey a +covered work so as to satisfy simultaneously your obligations under this +License and any other pertinent obligations, then as a consequence you may +not convey it at all. For example, if you agree to terms that obligate you +to collect a royalty for further conveying from those to whom you convey +the Program, the only way you could satisfy both those terms and this +License would be to refrain entirely from conveying the Program. + +### 13. Remote Network Interaction; Use with the GNU General Public License + +Notwithstanding any other provision of this License, if you modify the +Program, your modified version must prominently offer all users +interacting with it remotely through a computer network (if your version +supports such interaction) an opportunity to receive the Corresponding +Source of your version by providing access to the Corresponding Source +from a network server at no charge, through some standard or customary +means of facilitating copying of software. This Corresponding Source +shall include the Corresponding Source for any work covered by version 3 +of the GNU General Public License that is incorporated pursuant to the +following paragraph. + +Notwithstanding any other provision of this License, you have +permission to link or combine any covered work with a work licensed +under version 3 of the GNU General Public License into a single +combined work, and to convey the resulting work. The terms of this +License will continue to apply to the part which is the covered work, +but the work with which it is combined will remain governed by version +3 of the GNU General Public License. + +### 14. Revised Versions of this License + +The Free Software Foundation may publish revised and/or new versions of +the GNU Affero General Public License from time to time. Such new versions +will be similar in spirit to the present version, but may differ in detail to +address new problems or concerns. + +Each version is given a distinguishing version number. If the +Program specifies that a certain numbered version of the GNU Affero General +Public License “or any later version” applies to it, you have the +option of following the terms and conditions either of that numbered +version or of any later version published by the Free Software +Foundation. If the Program does not specify a version number of the +GNU Affero General Public License, you may choose any version ever published +by the Free Software Foundation. + +If the Program specifies that a proxy can decide which future +versions of the GNU Affero General Public License can be used, that proxy's +public statement of acceptance of a version permanently authorizes you +to choose that version for the Program. + +Later license versions may give you additional or different +permissions. However, no additional obligations are imposed on any +author or copyright holder as a result of your choosing to follow a +later version. + +### 15. Disclaimer of Warranty + +THERE IS NO WARRANTY FOR THE PROGRAM, TO THE EXTENT PERMITTED BY +APPLICABLE LAW. EXCEPT WHEN OTHERWISE STATED IN WRITING THE COPYRIGHT +HOLDERS AND/OR OTHER PARTIES PROVIDE THE PROGRAM “AS IS” WITHOUT WARRANTY +OF ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING, BUT NOT LIMITED TO, +THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR +PURPOSE. THE ENTIRE RISK AS TO THE QUALITY AND PERFORMANCE OF THE PROGRAM +IS WITH YOU. SHOULD THE PROGRAM PROVE DEFECTIVE, YOU ASSUME THE COST OF +ALL NECESSARY SERVICING, REPAIR OR CORRECTION. + +### 16. Limitation of Liability + +IN NO EVENT UNLESS REQUIRED BY APPLICABLE LAW OR AGREED TO IN WRITING +WILL ANY COPYRIGHT HOLDER, OR ANY OTHER PARTY WHO MODIFIES AND/OR CONVEYS +THE PROGRAM AS PERMITTED ABOVE, BE LIABLE TO YOU FOR DAMAGES, INCLUDING ANY +GENERAL, SPECIAL, INCIDENTAL OR CONSEQUENTIAL DAMAGES ARISING OUT OF THE +USE OR INABILITY TO USE THE PROGRAM (INCLUDING BUT NOT LIMITED TO LOSS OF +DATA OR DATA BEING RENDERED INACCURATE OR LOSSES SUSTAINED BY YOU OR THIRD +PARTIES OR A FAILURE OF THE PROGRAM TO OPERATE WITH ANY OTHER PROGRAMS), +EVEN IF SUCH HOLDER OR OTHER PARTY HAS BEEN ADVISED OF THE POSSIBILITY OF +SUCH DAMAGES. + +### 17. Interpretation of Sections 15 and 16 + +If the disclaimer of warranty and limitation of liability provided +above cannot be given local legal effect according to their terms, +reviewing courts shall apply local law that most closely approximates +an absolute waiver of all civil liability in connection with the +Program, unless a warranty or assumption of liability accompanies a +copy of the Program in return for a fee. diff --git a/libs/@local/integrations/engine/README.md b/libs/@local/integrations/engine/README.md new file mode 100644 index 00000000000..7543b32fe3a --- /dev/null +++ b/libs/@local/integrations/engine/README.md @@ -0,0 +1,5 @@ +# HASH Integrations Engine + +Runs an integration's pipelines in DuckDB. + +Each run gets a `Workspace`, a DuckDB database file in which the run's queries execute. A query is a single `SELECT`, and cannot open files, load extensions or change settings. diff --git a/libs/@local/integrations/engine/docs/dependency-diagram.mmd b/libs/@local/integrations/engine/docs/dependency-diagram.mmd new file mode 100644 index 00000000000..34dc22ca164 --- /dev/null +++ b/libs/@local/integrations/engine/docs/dependency-diagram.mmd @@ -0,0 +1,14 @@ +graph TD + linkStyle default stroke-width:1.5px + classDef default stroke-width:1px + classDef root stroke-width:3px + classDef dev stroke-width:1px + classDef build stroke-width:1px + %% Legend + %% --> : Normal dependency + %% -.-> : Dev dependency + %% ---> : Build dependency + 0[hash-integrations-engine] + class 0 root + 1[error-stack] + 0 --> 1 diff --git a/libs/@local/integrations/engine/docs/task-dependencies.json b/libs/@local/integrations/engine/docs/task-dependencies.json new file mode 100644 index 00000000000..5a2d69a8b0d --- /dev/null +++ b/libs/@local/integrations/engine/docs/task-dependencies.json @@ -0,0 +1,40 @@ +{ + "package": "hash-integrations-engine", + "dependencies": [ + "error-stack" + ], + "tasks": { + "doc:dependency-diagram": { + "dependsOn": [ + "hash-repo-chores#build" + ], + "cache": false, + "affectedBy": [ + "error-stack-macros", + "hash-repo-chores", + "hash-telemetry" + ] + }, + "fix:clippy": { + "dependsOn": [] + }, + "lint:clippy": { + "dependsOn": [], + "env": [ + "GITHUB_EVENT_NAME" + ], + "affectedBy": [ + "error-stack-macros" + ] + }, + "test:unit": { + "dependsOn": [], + "env": [ + "TEST_COVERAGE" + ], + "affectedBy": [ + "error-stack-macros" + ] + } + } +} diff --git a/libs/@local/integrations/engine/src/lib.rs b/libs/@local/integrations/engine/src/lib.rs new file mode 100644 index 00000000000..4c857df8771 --- /dev/null +++ b/libs/@local/integrations/engine/src/lib.rs @@ -0,0 +1,24 @@ +//! # HASH Integrations Engine +//! +//! Runs an integration's pipelines in DuckDB. Each run has a [`Workspace`], a database file in +//! which the pipelines' queries run. +//! +//! ## Workspace dependencies +#![doc = simple_mermaid::mermaid!("../docs/dependency-diagram.mmd")] +#![feature( + const_array, + const_closures, + const_trait_impl, + type_info, + variant_count +)] + +extern crate alloc; + +mod sql; +mod variant_name; +mod workspace; + +pub use self::workspace::{ + CloseError, DataErrorType, OpenError, QueryErrorType, Rows, Setting, StatementError, Workspace, +}; diff --git a/libs/@local/integrations/engine/src/sql.rs b/libs/@local/integrations/engine/src/sql.rs new file mode 100644 index 00000000000..3df30ca6543 --- /dev/null +++ b/libs/@local/integrations/engine/src/sql.rs @@ -0,0 +1,51 @@ +use core::fmt::{self, Write as _}; + +/// A table or column name in a DuckDB statement. +/// +/// It displays as a quoted identifier with each `"` doubled, so the name can contain spaces, +/// quotes and `/`, and can be a keyword. DuckDB refuses an empty name and a name that contains +/// NUL. It compares identifiers without regard to case, quoted or not. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct Identifier<'name>(&'name str); + +impl<'name> Identifier<'name> { + pub(crate) const fn new(name: &'name str) -> Self { + Self(name) + } +} + +impl fmt::Display for Identifier<'_> { + fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt.write_char('"')?; + for (position, part) in self.0.split('"').enumerate() { + if position > 0 { + fmt.write_str("\"\"")?; + } + fmt.write_str(part)?; + } + fmt.write_char('"') + } +} + +#[cfg(test)] +mod tests { + use duckdb::Connection; + use proptest::{prop_assert_eq, property_test}; + + use super::Identifier; + + #[property_test] + fn identifier_round_trips(#[strategy = "(['\"]|[^\\x00])+"] name: String) { + let connection = Connection::open_in_memory().expect("should open an in-memory database"); + let mut statement = connection + .prepare(&format!("SELECT 1 AS {}", Identifier::new(&name))) + .expect("the identifier should parse"); + statement.execute([]).expect("the statement should run"); + + prop_assert_eq!( + statement.column_names(), + [name], + "DuckDB should name the column with the quoted identifier" + ); + } +} diff --git a/libs/@local/integrations/engine/src/variant_name.rs b/libs/@local/integrations/engine/src/variant_name.rs new file mode 100644 index 00000000000..eeb90b3459e --- /dev/null +++ b/libs/@local/integrations/engine/src/variant_name.rs @@ -0,0 +1,107 @@ +//! Variant names read from an enum's definition when the crate compiles. + +use core::{any::TypeId, fmt, mem}; + +/// The name of an enum variant, in `PascalCase` as the enum declares it. +#[derive(Debug, Clone, Copy)] +pub(crate) struct VariantName(&'static str); + +impl VariantName { + /// Returns the name of each variant of `T` in declaration order. + /// + /// `N` must be the number of variants of `T`, or the build fails. + pub(crate) const fn all() -> [Self; N] { + const { + assert!( + N == mem::variant_count::(), + "`N` should be the number of variants" + ); + + let id = TypeId::of::(); + let mut names = [Self(""); N]; + let mut index = 0; + while index < N { + names[index] = Self(id.variant(index).name()); + index += 1; + } + names + } + } + + /// Returns the name as lowercase words joined by `separator`, so that `OutOfRange` displays + /// as `out of range` with a space. + pub(crate) const fn words(self, separator: char) -> Words { + Words { + name: self.0, + separator, + } + } + + /// Returns whether `text` spells the words of this name, ignoring ASCII case and spaces, so + /// that `Out of Range` matches `OutOfRange`. + pub(crate) fn matches_words(self, text: &str) -> bool { + text.bytes() + .filter(|byte| *byte != b' ') + .map(|byte| byte.to_ascii_lowercase()) + .eq(self.0.bytes().map(|byte| byte.to_ascii_lowercase())) + } +} + +/// A variant name displayed as lowercase words joined by a separator. +pub(crate) struct Words { + name: &'static str, + separator: char, +} + +impl fmt::Display for Words { + fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { + for (index, letter) in self.name.char_indices() { + if index > 0 && letter.is_ascii_uppercase() { + fmt::Write::write_char(fmt, self.separator)?; + } + fmt::Write::write_char(fmt, letter.to_ascii_lowercase())?; + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::VariantName; + + #[expect(dead_code, reason = "only the variant names are read")] + enum Example { + OutOfRange, + Binder, + } + + #[test] + fn words_joins_lowercase_words() { + let [out_of_range, binder] = VariantName::all::(); + assert_eq!( + out_of_range.words('_').to_string(), + "out_of_range", + "each capital letter should start a new word" + ); + assert_eq!( + binder.words(' ').to_string(), + "binder", + "a one-word name should only be lowercased" + ); + } + + #[test] + fn matches_words_ignores_case_and_spaces() { + let [out_of_range, _] = VariantName::all::(); + for text in ["Out of Range", "OUT OF RANGE", "OutOfRange"] { + assert!( + out_of_range.matches_words(text), + "`{text}` should match `OutOfRange`" + ); + } + assert!( + !out_of_range.matches_words("Out of Ranges"), + "a different name should not match" + ); + } +} diff --git a/libs/@local/integrations/engine/src/workspace/connection.rs b/libs/@local/integrations/engine/src/workspace/connection.rs new file mode 100644 index 00000000000..29358cbda67 --- /dev/null +++ b/libs/@local/integrations/engine/src/workspace/connection.rs @@ -0,0 +1,262 @@ +use alloc::sync::Arc; +use core::{error::Error, panic::AssertUnwindSafe}; +use std::{ + fs::{File, OpenOptions, TryLockError}, + panic, + path::PathBuf, + sync::mpsc, +}; + +use duckdb::{Config, types::Value}; +use error_stack::{Report, ResultExt as _}; + +use crate::{ + sql::Identifier, + workspace::{ + CloseError, Command, OpenError, Rows, Setting, StatementError, + files::WorkspaceFiles, + interrupt::{Interrupt, InterruptGuard, RequestId}, + reply::Reply, + select::Select, + }, +}; + +/// The database's only connection, owned by the workspace thread. +/// +/// Fields are dropped in declaration order, so `interrupt` stops interrupts before `duckdb` closes +/// the database, and `lock` is released after it. +pub(super) struct Connection { + interrupt: InterruptGuard, + duckdb: duckdb::Connection, + /// An exclusive lock on the workspace's lock file, held until the database is closed. + lock: File, +} + +impl Connection { + /// Takes an exclusive lock on the lock file of `files`. + /// + /// DuckDB does not stop a second connection in the same process from opening a database, so + /// the workspace locks a file of its own first. + fn lock(files: &WorkspaceFiles) -> Result> { + let lock = OpenOptions::new() + .write(true) + .create(true) + .truncate(false) + .open(files.lock()) + .change_context_lazy(|| OpenError::Lock { + path: files.lock().to_owned(), + })?; + match lock.try_lock() { + Ok(()) => Ok(lock), + Err(TryLockError::WouldBlock) => Err(Report::new(OpenError::Locked { + path: files.lock().to_owned(), + })), + Err(TryLockError::Error(error)) => { + Err(Report::new(error).change_context(OpenError::Lock { + path: files.lock().to_owned(), + })) + } + } + } + + /// Opens the database at `path` with its settings locked. + pub(super) fn open(path: PathBuf) -> Result> { + let files = WorkspaceFiles::new(path); + let lock = Self::lock(&files)?; + + let mut config = Config::default(); + for (setting, value) in [ + (Setting::AllowCommunityExtensions, "false"), + (Setting::AutoinstallKnownExtensions, "false"), + (Setting::AutoloadKnownExtensions, "false"), + (Setting::EnableExternalAccess, "false"), + (Setting::LockConfiguration, "true"), + ] { + config = config + .with(setting.to_string(), value) + .change_context_lazy(|| OpenError::Configure { + setting, + value: value.into(), + })?; + } + let duckdb = duckdb::Connection::open_with_flags(files.database(), config) + .change_context_lazy(|| OpenError::Open { + path: files.database().to_owned(), + })?; + + Ok(Self { + interrupt: InterruptGuard::new(Arc::new(Interrupt::new(duckdb.interrupt_handle()))), + duckdb, + lock, + }) + } + + /// Returns the [`Interrupt`] that cancels the requests running on this connection. + pub(super) const fn interrupt(&self) -> &Arc { + self.interrupt.interrupt() + } + + /// Checks that DuckDB reads `sql` as a single `SELECT` statement whose table functions are + /// allowed. + fn select<'sql>(&self, sql: &'sql str) -> Result, Report> { + let serialized: String = self + .duckdb + .prepare_cached(Select::SERIALIZE) + .and_then(|mut statement| statement.query_row([sql], |row| row.get(0))) + .map_err(StatementError::from_duckdb)?; + Select::check(sql, &serialized) + } + + #[tracing::instrument(level = "debug", skip_all)] + fn query(&self, sql: &str) -> Result> { + let select = self.select(sql)?; + let mut statement = self + .duckdb + .prepare(select.as_str()) + .map_err(StatementError::from_duckdb)?; + let mut results = statement.query([]).map_err(StatementError::from_duckdb)?; + + if let Some(executed) = results.as_ref() { + for index in 0..executed.column_count() { + let data_type = executed.column_type(index); + if !Rows::readable(&data_type) { + let column = executed + .column_name(index) + .map_err(StatementError::from_duckdb)?; + return Err(Report::new(StatementError::UnsupportedType { + column: column.as_str().into(), + data_type: data_type.to_string().into_boxed_str(), + })); + } + } + } + + let mut rows = Vec::new(); + while let Some(row) = results.next().map_err(StatementError::from_duckdb)? { + let values = (0..row.as_ref().column_count()) + .map(|index| row.get::<_, Value>(index)) + .collect::, _>>() + .map_err(StatementError::from_duckdb)?; + rows.push(values); + } + drop(results); + + let columns = statement + .column_names() + .into_iter() + .map(String::into_boxed_str) + .collect(); + Ok(Rows::new(columns, rows)) + } + + #[tracing::instrument(level = "debug", skip(self, sql))] + fn create_table(&self, table: &str, sql: &str) -> Result<(), Report> { + let select = self.select(sql)?; + // `select` is one statement, so the whole text is one statement as well. + self.duckdb + .prepare(&format!( + "CREATE OR REPLACE TABLE {} AS {}", + Identifier::new(table), + select.as_str() + )) + .and_then(|mut statement| statement.execute([])) + .map_err(StatementError::from_duckdb)?; + Ok(()) + } + + /// Checkpoints the database, then closes it whether or not the checkpoint succeeded. + /// + /// DuckDB also checkpoints when it closes the database, but it cannot report a failure from + /// there. + fn close(self) -> Result<(), Report> { + let checkpoint = self + .duckdb + .execute_batch("CHECKPOINT") + .change_context(CloseError::Checkpoint); + drop(self.interrupt); + drop(self.duckdb); + drop(self.lock); + checkpoint + } + + /// Runs `request` and sends its result on `reply`. + /// + /// A request whose caller stopped waiting before it started is skipped. A request that its + /// caller cancelled was interrupted, so its result is dropped. + /// + /// The start is recorded before the reply is checked, so a caller that cancels the request + /// either closed the reply before the check or interrupts the request. + fn answer( + &self, + id: RequestId, + reply: Reply, + request: impl FnOnce(&Self) -> Result>, + ) where + C: Error + Send + Sync + 'static, + { + let interrupt = self.interrupt(); + interrupt.start(id); + if reply.is_closed() { + interrupt.finish(); + return; + } + let result = request(self); + if !interrupt.finish() { + reply.send(result); + } + } + + /// Runs `commands` in order until [`Command::Close`] or until the [`Workspace`] is dropped. + /// + /// [`Workspace`]: crate::Workspace + pub(super) fn run(self, commands: &mpsc::Receiver<(RequestId, Command)>) { + while let Ok((id, command)) = commands.recv() { + match command { + Command::Query { sql, reply } => { + self.answer(id, reply, |connection| connection.query(&sql)); + } + Command::CreateTable { table, sql, reply } => { + self.answer(id, reply, |connection| { + connection.create_table(&table, &sql) + }); + } + Command::Close { reply } => { + reply.send(self.close()); + return; + } + } + } + + if let Err(report) = self.close() { + tracing::error!(error = ?report, "could not close the dropped workspace"); + } + } + + /// Opens the database at `path`, sends the result on `ready`, and runs `commands`. + /// + /// A panic while the commands run is logged. The connection is dropped while the panic + /// unwinds, so nothing it holds is used after the panic. + pub(super) fn serve( + path: PathBuf, + ready: Reply, OpenError>, + commands: &mpsc::Receiver<(RequestId, Command)>, + ) { + let connection = match Self::open(path) { + Ok(connection) => connection, + Err(report) => { + ready.send(Err(report)); + return; + } + }; + ready.send(Ok(Arc::clone(connection.interrupt()))); + + if let Err(panic) = panic::catch_unwind(AssertUnwindSafe(|| connection.run(commands))) { + let message = panic + .downcast_ref::<&str>() + .copied() + .or_else(|| panic.downcast_ref::().map(String::as_str)) + .unwrap_or_default(); + tracing::error!(panic = message, "the workspace thread panicked"); + } + } +} diff --git a/libs/@local/integrations/engine/src/workspace/error.rs b/libs/@local/integrations/engine/src/workspace/error.rs new file mode 100644 index 00000000000..1100eb1f84f --- /dev/null +++ b/libs/@local/integrations/engine/src/workspace/error.rs @@ -0,0 +1,270 @@ +use core::{array, fmt, mem}; +use std::path::PathBuf; + +use error_stack::Report; + +use crate::variant_name::VariantName; + +/// A DuckDB setting the workspace applies when it opens the database. +/// +/// It displays as DuckDB's name for the setting, its variant name in snake case. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Setting { + AllowCommunityExtensions, + AutoinstallKnownExtensions, + AutoloadKnownExtensions, + EnableExternalAccess, + LockConfiguration, +} + +impl Setting { + const NAMES: [VariantName; mem::variant_count::()] = VariantName::all::(); + + const fn name(self) -> VariantName { + Self::NAMES[self as usize] + } +} + +impl fmt::Display for Setting { + fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(&self.name().words('_'), fmt) + } +} + +/// Reports why [`Workspace::open`](crate::Workspace::open) failed. +#[derive(Debug, derive_more::Display, derive_more::Error)] +pub enum OpenError { + #[display("could not lock `{}`", path.display())] + Lock { path: PathBuf }, + #[display("another workspace holds the lock `{}`", path.display())] + Locked { path: PathBuf }, + #[display("could not set `{setting}` to `{value}`")] + Configure { setting: Setting, value: Box }, + #[display("could not open the database at `{}`", path.display())] + Open { path: PathBuf }, + #[display("could not start the workspace thread")] + Spawn, + #[display("the workspace thread stopped before the database opened")] + Stopped, +} + +/// A DuckDB error type that means a query cannot run as written. +/// +/// It displays as its variant name in lowercase words, such as `mismatch type`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +pub enum QueryErrorType { + Binder, + Catalog, + Dependency, + ExtensionAutoloading, + MismatchType, + MissingExtension, + NotImplemented, + ParameterNotAllowed, + ParameterNotResolved, + Parser, + Permission, + Syntax, +} + +impl QueryErrorType { + /// Every query error type, in declaration order. + const ALL: [Self; mem::variant_count::()] = array::from_fn(const |index| { + #[expect( + clippy::cast_possible_truncation, + reason = "`repr(u8)` keeps the variant count within `u8`" + )] + let discriminant = index as u8; + // SAFETY: `repr(u8)` gives the variants the discriminants 0 up to the variant count in + // declaration order, and `index` is below the variant count. + #[expect(unsafe_code, reason = "builds each variant from its discriminant")] + unsafe { + mem::transmute::(discriminant) + } + }); + const NAMES: [VariantName; mem::variant_count::()] = VariantName::all::(); + + const fn name(self) -> VariantName { + Self::NAMES[self as usize] + } + + /// Returns the error type that DuckDB calls `name`, or `None` for another name. + fn from_name(name: &str) -> Option { + Self::ALL + .into_iter() + .find(|error_type| error_type.name().matches_words(name)) + } +} + +impl fmt::Display for QueryErrorType { + fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(&self.name().words(' '), fmt) + } +} + +/// A DuckDB error type that means a value in the rows caused the error. +/// +/// It displays as its variant name in lowercase words, such as `out of range`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +pub enum DataErrorType { + Constraint, + Conversion, + Decimal, + DivideByZero, + InvalidInput, + OutOfRange, +} + +impl DataErrorType { + /// Every data error type, in declaration order. + const ALL: [Self; mem::variant_count::()] = array::from_fn(const |index| { + #[expect( + clippy::cast_possible_truncation, + reason = "`repr(u8)` keeps the variant count within `u8`" + )] + let discriminant = index as u8; + // SAFETY: `repr(u8)` gives the variants the discriminants 0 up to the variant count in + // declaration order, and `index` is below the variant count. + #[expect(unsafe_code, reason = "builds each variant from its discriminant")] + unsafe { + mem::transmute::(discriminant) + } + }); + const NAMES: [VariantName; mem::variant_count::()] = VariantName::all::(); + + const fn name(self) -> VariantName { + Self::NAMES[self as usize] + } + + /// Returns the error type that DuckDB calls `name`, or `None` for another name. + fn from_name(name: &str) -> Option { + Self::ALL + .into_iter() + .find(|error_type| error_type.name().matches_words(name)) + } +} + +impl fmt::Display for DataErrorType { + fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(&self.name().words(' '), fmt) + } +} + +/// Reports why [`Workspace::query`](crate::Workspace::query) or +/// [`Workspace::create_table`](crate::Workspace::create_table) failed. +#[derive(Debug, derive_more::Display, derive_more::Error)] +pub enum StatementError { + #[display("the SQL is not a `SELECT` query: {reason}")] + NotSelect { reason: Box }, + #[display("the SQL holds {count} statements instead of one")] + StatementCount { count: usize }, + #[display("the query calls table function `{name}`, which is not allowed")] + TableFunction { name: Box }, + /// `data_type` is the column's type as the `duckdb` crate receives it, in Arrow's notation. + #[display("column `{column}` has type `{data_type}`, which the workspace cannot read")] + UnsupportedType { + column: Box, + data_type: Box, + }, + /// The query cannot run as written, for example because it reads a missing table. + #[display("DuckDB cannot run the query: {error_type} error")] + InvalidQuery { error_type: QueryErrorType }, + /// A value in the rows caused an error, such as text that a cast cannot convert. + /// + /// The DuckDB error this carries can quote the value. + #[display("a value in the rows caused a DuckDB {error_type} error")] + InvalidData { error_type: DataErrorType }, + /// DuckDB failed for another reason, such as an IO error or running out of memory, or a value + /// could not be read. + #[display("the statement failed")] + Failed, + #[display("the workspace thread stopped before the statement finished")] + Stopped, +} + +impl StatementError { + /// Returns a report of `error` as [`InvalidQuery`], [`InvalidData`] or [`Failed`], by its + /// DuckDB error type. + /// + /// DuckDB starts an error message with the error type, as in `Binder Error: …`. Other error + /// types, and errors without a message, are [`Failed`]. + /// + /// [`InvalidQuery`]: Self::InvalidQuery + /// [`InvalidData`]: Self::InvalidData + /// [`Failed`]: Self::Failed + pub(super) fn from_duckdb(error: duckdb::Error) -> Report { + let name = if let duckdb::Error::DuckDBFailure(_, Some(message)) = &error { + message.split_once(" Error: ").map(|(name, _)| name) + } else { + None + }; + let context = name + .and_then(|name| { + QueryErrorType::from_name(name) + .map(|error_type| Self::InvalidQuery { error_type }) + .or_else(|| { + DataErrorType::from_name(name) + .map(|error_type| Self::InvalidData { error_type }) + }) + }) + .unwrap_or(Self::Failed); + Report::new(error).change_context(context) + } +} + +/// Reports why [`Workspace::close`](crate::Workspace::close) failed. +#[derive(Debug, derive_more::Display, derive_more::Error)] +pub enum CloseError { + #[display("could not checkpoint the database before closing it")] + Checkpoint, + #[display("the workspace thread stopped before the database closed")] + Stopped, +} + +#[cfg(test)] +mod tests { + use duckdb::ffi; + use error_stack::Report; + + use super::{DataErrorType, QueryErrorType, StatementError}; + + fn from_message(message: &str) -> Report { + StatementError::from_duckdb(duckdb::Error::DuckDBFailure( + ffi::Error::new(ffi::DuckDBError), + Some(message.to_owned()), + )) + } + + #[test] + fn from_duckdb_reads_error_type() { + assert!( + matches!( + from_message("Mismatch Type Error: Cannot compare values").current_context(), + StatementError::InvalidQuery { + error_type: QueryErrorType::MismatchType + } + ), + "a query error type should be read from the message" + ); + assert!( + matches!( + from_message("Out of Range Error: Overflow in addition").current_context(), + StatementError::InvalidData { + error_type: DataErrorType::OutOfRange + } + ), + "a data error type should be read from the message" + ); + for message in ["IO Error: No space left on device", "Error: no error type"] { + assert!( + matches!( + from_message(message).current_context(), + StatementError::Failed + ), + "`{message}` should be reported as failed" + ); + } + } +} diff --git a/libs/@local/integrations/engine/src/workspace/files.rs b/libs/@local/integrations/engine/src/workspace/files.rs new file mode 100644 index 00000000000..a5efba4483e --- /dev/null +++ b/libs/@local/integrations/engine/src/workspace/files.rs @@ -0,0 +1,27 @@ +use std::path::{Path, PathBuf}; + +/// The files of a workspace, named after its database file. +pub(super) struct WorkspaceFiles { + database: PathBuf, + /// The file that the workspace locks while the database is open. + lock: PathBuf, +} + +impl WorkspaceFiles { + pub(super) fn new(database: PathBuf) -> Self { + let mut lock = database.as_os_str().to_owned(); + lock.push(".lock"); + Self { + database, + lock: PathBuf::from(lock), + } + } + + pub(super) fn database(&self) -> &Path { + &self.database + } + + pub(super) fn lock(&self) -> &Path { + &self.lock + } +} diff --git a/libs/@local/integrations/engine/src/workspace/interrupt.rs b/libs/@local/integrations/engine/src/workspace/interrupt.rs new file mode 100644 index 00000000000..29ffa609e9a --- /dev/null +++ b/libs/@local/integrations/engine/src/workspace/interrupt.rs @@ -0,0 +1,180 @@ +use alloc::sync::Arc; +use core::{ + fmt, + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; +use std::{ + sync::{Mutex, MutexGuard, PoisonError}, + thread, +}; + +use duckdb::InterruptHandle; +use tokio::sync::oneshot; + +/// Identifies a request to the workspace thread. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) struct RequestId(u64); + +/// The request the workspace thread is running, and the handle that interrupts it. +struct Running { + request: Option, + /// Whether [`Interrupt::cancel`] interrupted `request`. + interrupted: bool, + /// Interrupts the statement running on the connection, until the connection closes. + handle: Option>, +} + +/// Identifies the requests to the workspace thread, and interrupts a running request whose caller +/// stopped waiting for it. +pub(super) struct Interrupt { + running: Mutex, + next_request: AtomicU64, +} + +impl Interrupt { + /// How often a cancelled request is interrupted again until it finishes. + const REPEAT: Duration = Duration::from_millis(1); + + pub(super) const fn new(handle: Arc) -> Self { + Self { + running: Mutex::new(Running { + request: None, + interrupted: false, + handle: Some(handle), + }), + next_request: AtomicU64::new(0), + } + } + + fn running(&self) -> MutexGuard<'_, Running> { + self.running.lock().unwrap_or_else(PoisonError::into_inner) + } + + /// Returns an ID that no earlier request has. + pub(super) fn next_request(&self) -> RequestId { + // Each request needs only a distinct ID, so relaxed ordering is enough. + RequestId(self.next_request.fetch_add(1, Ordering::Relaxed)) + } + + /// Records that the workspace thread starts `request`. + pub(super) fn start(&self, request: RequestId) { + let mut running = self.running(); + running.request = Some(request); + running.interrupted = false; + } + + /// Records that the workspace thread finished its request, and returns whether + /// [`cancel`](Self::cancel) interrupted it. + pub(super) fn finish(&self) -> bool { + let mut running = self.running(); + running.request = None; + running.interrupted + } + + /// Interrupts `request` if the workspace thread is running it, and returns whether it did. + fn interrupt(&self, request: RequestId) -> bool { + let mut running = self.running(); + if running.request != Some(request) { + return false; + } + let Some(handle) = &running.handle else { + return false; + }; + handle.interrupt(); + running.interrupted = true; + true + } + + /// Interrupts `request` until it finishes, if the workspace thread is running it. + /// + /// DuckDB clears an interrupt when a statement starts, so an interrupt that arrives between two + /// statements of a request is lost. A thread repeats the interrupt until the request finishes. + fn cancel(self: &Arc, request: RequestId) { + if !self.interrupt(request) { + return; + } + let interrupt = Arc::clone(self); + let repeat = thread::Builder::new() + .name("integrations-workspace-cancel".to_owned()) + .spawn(move || { + while interrupt.interrupt(request) { + thread::sleep(Self::REPEAT); + } + }); + if let Err(error) = repeat { + tracing::error!( + ?error, + "could not start the thread that repeats an interrupt, so it may be lost" + ); + } + } +} + +impl fmt::Debug for Interrupt { + fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt.debug_struct("Interrupt") + .field("request", &self.running().request) + .finish_non_exhaustive() + } +} + +/// Receives the reply to a request, and cancels the request if it is dropped before the reply +/// arrives. +pub(super) struct Pending<'interrupt, T> { + interrupt: &'interrupt Arc, + request: RequestId, + reply: oneshot::Receiver, +} + +impl<'interrupt, T> Pending<'interrupt, T> { + pub(super) const fn new( + interrupt: &'interrupt Arc, + request: RequestId, + reply: oneshot::Receiver, + ) -> Self { + Self { + interrupt, + request, + reply, + } + } + + /// Waits for the reply. + pub(super) async fn reply(&mut self) -> Result { + (&mut self.reply).await + } +} + +impl Drop for Pending<'_, T> { + fn drop(&mut self) { + // The workspace thread records that it starts a request before it checks whether the + // reply is closed. Closing the reply before the cancel means the thread either sees the + // closed reply or is interrupted. + self.reply.close(); + self.interrupt.cancel(self.request); + } +} + +/// Holds the connection's [`Interrupt`], and stops interrupts when it is dropped. +/// +/// The `duckdb` crate closes a connection before it clears the connection's interrupt handle, so +/// an interrupt that arrives while the connection closes could reach a closed connection. The +/// guard is dropped before the connection closes, and waits for an interrupt in progress. +pub(super) struct InterruptGuard(Arc); + +impl InterruptGuard { + pub(super) const fn new(interrupt: Arc) -> Self { + Self(interrupt) + } + + pub(super) const fn interrupt(&self) -> &Arc { + &self.0 + } +} + +impl Drop for InterruptGuard { + fn drop(&mut self) { + self.0.running().handle = None; + } +} diff --git a/libs/@local/integrations/engine/src/workspace/mod.rs b/libs/@local/integrations/engine/src/workspace/mod.rs new file mode 100644 index 00000000000..0e1180a3199 --- /dev/null +++ b/libs/@local/integrations/engine/src/workspace/mod.rs @@ -0,0 +1,200 @@ +mod connection; +mod error; +mod files; +mod interrupt; +mod reply; +mod rows; +mod select; + +use alloc::sync::Arc; +use core::error::Error; +use std::{path::PathBuf, sync::mpsc, thread}; + +use error_stack::{Report, ResultExt as _}; + +pub use self::{ + error::{CloseError, DataErrorType, OpenError, QueryErrorType, Setting, StatementError}, + rows::Rows, +}; +use crate::workspace::{ + connection::Connection, + interrupt::{Interrupt, Pending, RequestId}, + reply::Reply, +}; + +/// A request to the workspace thread, with the channel for its reply. +enum Command { + Query { + sql: Box, + reply: Reply, + }, + CreateTable { + table: Box, + sql: Box, + reply: Reply<(), StatementError>, + }, + Close { + reply: Reply<(), CloseError>, + }, +} + +/// A DuckDB database file in which an integration's queries run. +/// +/// A dedicated thread owns the database's only connection and runs queries in the order they +/// are sent. A query is a single `SELECT` statement, and the only table functions it can call are +/// `range`, `generate_series` and `unnest`. Queries cannot open files, load extensions or change +/// settings. +/// +/// Only one workspace at a time can open a database. It holds a lock on `.lock` until +/// the database is closed. +/// +/// Dropping the future of a request cancels it: the thread skips the request if it has not +/// started, and interrupts it otherwise. A caller can limit how long a request runs with +/// `tokio::time::timeout`. +/// +/// Dropping the workspace returns at once. The futures of its requests are dropped before it, so +/// the thread skips or interrupts them, closes the database and releases the lock. The thread +/// logs a failed checkpoint. [`close`](Self::close) waits for the requests already sent to run, +/// and returns a failed checkpoint as an error. +#[derive(Debug)] +pub struct Workspace { + commands: mpsc::Sender<(RequestId, Command)>, + interrupt: Arc, +} + +impl Workspace { + /// Opens the database at `path`, creating it if it does not exist. + /// + /// # Errors + /// + /// - [`Lock`] if `.lock` cannot be created or locked + /// - [`Locked`] if another workspace has the database open + /// - [`Configure`] if DuckDB refuses a setting + /// - [`Open`] if DuckDB cannot open the database + /// - [`Spawn`] if the workspace thread cannot be started + /// - [`Stopped`] if the workspace thread stopped before it opened the database + /// + /// [`Lock`]: OpenError::Lock + /// [`Locked`]: OpenError::Locked + /// [`Configure`]: OpenError::Configure + /// [`Open`]: OpenError::Open + /// [`Spawn`]: OpenError::Spawn + /// [`Stopped`]: OpenError::Stopped + pub async fn open(path: PathBuf) -> Result> { + let (commands, receiver) = mpsc::channel(); + let (ready, opened) = Reply::channel(); + let span = tracing::info_span!("workspace", path = %path.display()); + + thread::Builder::new() + .name("integrations-workspace".to_owned()) + .spawn(move || { + let _entered = span.entered(); + Connection::serve(path, ready, &receiver); + }) + .change_context(OpenError::Spawn)?; + + let interrupt = opened.await.change_context(OpenError::Stopped)??; + Ok(Self { + commands, + interrupt, + }) + } + + /// Sends the command that `command` builds around its reply channel, and waits for the reply. + /// + /// The request is cancelled if the returned future is dropped before the reply arrives. + async fn request( + &self, + command: impl FnOnce(Reply) -> Command, + stopped: C, + ) -> Result> + where + C: Error + Send + Sync + 'static, + { + let request = self.interrupt.next_request(); + let (reply, result) = Reply::channel(); + if self.commands.send((request, command(reply))).is_err() { + return Err(Report::new(stopped)); + } + Pending::new(&self.interrupt, request, result) + .reply() + .await + .change_context(stopped)? + } + + /// Runs the query `sql` and returns all of its rows. + /// + /// # Errors + /// + /// - [`NotSelect`] if DuckDB does not read `sql` as a `SELECT` query + /// - [`StatementCount`] if `sql` is not exactly one statement + /// - [`TableFunction`] if the query calls a table function that is not allowed + /// - [`UnsupportedType`] if a column has a type whose values cannot be read + /// - [`InvalidQuery`] if DuckDB cannot run the query as written + /// - [`InvalidData`] if a value in the rows causes an error + /// - [`Failed`] if DuckDB fails for another reason or a value cannot be read + /// - [`Stopped`] if the workspace thread stopped + /// + /// [`NotSelect`]: StatementError::NotSelect + /// [`StatementCount`]: StatementError::StatementCount + /// [`TableFunction`]: StatementError::TableFunction + /// [`UnsupportedType`]: StatementError::UnsupportedType + /// [`InvalidQuery`]: StatementError::InvalidQuery + /// [`InvalidData`]: StatementError::InvalidData + /// [`Failed`]: StatementError::Failed + /// [`Stopped`]: StatementError::Stopped + pub async fn query(&self, sql: Box) -> Result> { + self.request( + |reply| Command::Query { sql, reply }, + StatementError::Stopped, + ) + .await + } + + /// Replaces the table named `table` with the rows of the query `sql`. + /// + /// # Errors + /// + /// - [`NotSelect`] if DuckDB does not read `sql` as a `SELECT` query + /// - [`StatementCount`] if `sql` is not exactly one statement + /// - [`TableFunction`] if the query calls a table function that is not allowed + /// - [`InvalidQuery`] if DuckDB cannot run the query as written + /// - [`InvalidData`] if a value in the rows causes an error + /// - [`Failed`] if DuckDB fails for another reason + /// - [`Stopped`] if the workspace thread stopped + /// + /// [`NotSelect`]: StatementError::NotSelect + /// [`StatementCount`]: StatementError::StatementCount + /// [`TableFunction`]: StatementError::TableFunction + /// [`InvalidQuery`]: StatementError::InvalidQuery + /// [`InvalidData`]: StatementError::InvalidData + /// [`Failed`]: StatementError::Failed + /// [`Stopped`]: StatementError::Stopped + pub async fn create_table( + &self, + table: Box, + sql: Box, + ) -> Result<(), Report> { + self.request( + |reply| Command::CreateTable { table, sql, reply }, + StatementError::Stopped, + ) + .await + } + + /// Checkpoints and closes the database after the requests already sent have run. + /// + /// The database is closed even if the checkpoint fails. + /// + /// # Errors + /// + /// - [`Checkpoint`] if DuckDB cannot write its log into the database file + /// - [`Stopped`] if the workspace thread stopped + /// + /// [`Checkpoint`]: CloseError::Checkpoint + /// [`Stopped`]: CloseError::Stopped + pub async fn close(self) -> Result<(), Report> { + self.request(|reply| Command::Close { reply }, CloseError::Stopped) + .await + } +} diff --git a/libs/@local/integrations/engine/src/workspace/reply.rs b/libs/@local/integrations/engine/src/workspace/reply.rs new file mode 100644 index 00000000000..49d6a1e9223 --- /dev/null +++ b/libs/@local/integrations/engine/src/workspace/reply.rs @@ -0,0 +1,36 @@ +use core::error::Error; + +use error_stack::Report; +use tokio::sync::oneshot; + +/// The channel on which the workspace thread answers one request. +pub(super) struct Reply(oneshot::Sender>>); + +impl Reply { + /// Returns a reply channel and the receiver that the caller waits on. + pub(super) fn channel() -> (Self, oneshot::Receiver>>) { + let (sender, receiver) = oneshot::channel(); + (Self(sender), receiver) + } + + /// Returns whether the caller stopped waiting for the reply. + pub(super) fn is_closed(&self) -> bool { + self.0.is_closed() + } + + /// Sends `result` to the caller. + /// + /// A caller that stopped waiting does not receive the result, so a failure is logged instead. + /// Only the report's current context is logged, because DuckDB errors can quote row values. + pub(super) fn send(self, result: Result>) + where + C: Error + Send + Sync + 'static, + { + if let Err(Err(report)) = self.0.send(result) { + tracing::error!( + error = %report.current_context(), + "workspace request failed after its caller stopped waiting" + ); + } + } +} diff --git a/libs/@local/integrations/engine/src/workspace/rows.rs b/libs/@local/integrations/engine/src/workspace/rows.rs new file mode 100644 index 00000000000..fcf58937a24 --- /dev/null +++ b/libs/@local/integrations/engine/src/workspace/rows.rs @@ -0,0 +1,88 @@ +use duckdb::{ + arrow::datatypes::{DataType, IntervalUnit, TimeUnit}, + types::Value, +}; + +/// The rows a query returned. +#[derive(Debug)] +pub struct Rows { + columns: Box<[Box]>, + rows: Vec>, +} + +impl Rows { + pub(super) const fn new(columns: Box<[Box]>, rows: Vec>) -> Self { + Self { columns, rows } + } + + /// Returns whether the `duckdb` crate reads values of `data_type` without panicking. + /// + /// The `duckdb` crate panics on DuckDB's `TIME_NS`, which arrives as `Time64(Nanosecond)`, also + /// inside a list, struct, map or union. The other types that return `false` are ones DuckDB + /// does not return. + pub(super) fn readable(data_type: &DataType) -> bool { + match data_type { + DataType::Utf8 + | DataType::LargeUtf8 + | DataType::Binary + | DataType::LargeBinary + | DataType::FixedSizeBinary(_) + | DataType::Boolean + | DataType::Int8 + | DataType::Int16 + | DataType::Int32 + | DataType::Int64 + | DataType::UInt8 + | DataType::UInt16 + | DataType::UInt32 + | DataType::UInt64 + | DataType::Float16 + | DataType::Float32 + | DataType::Float64 + | DataType::Decimal32(..) + | DataType::Decimal64(..) + | DataType::Decimal128(..) + | DataType::Timestamp(..) + | DataType::Date32 + | DataType::Time64(TimeUnit::Microsecond) + | DataType::Interval(IntervalUnit::MonthDayNano) => true, + DataType::List(item) | DataType::LargeList(item) | DataType::FixedSizeList(item, _) => { + Self::readable(item.data_type()) + } + DataType::Map(entry, _) => Self::readable(entry.data_type()), + DataType::Struct(fields) => { + fields.iter().all(|field| Self::readable(field.data_type())) + } + DataType::Union(fields, _) => fields + .iter() + .all(|(_, field)| Self::readable(field.data_type())), + DataType::Dictionary(key, value) => { + matches!(**key, DataType::UInt8 | DataType::UInt16 | DataType::UInt32) + && Self::readable(value) + } + DataType::Null + | DataType::Date64 + | DataType::Time32(_) + | DataType::Time64(_) + | DataType::Duration(_) + | DataType::Interval(_) + | DataType::BinaryView + | DataType::Utf8View + | DataType::ListView(_) + | DataType::LargeListView(_) + | DataType::Decimal256(..) + | DataType::RunEndEncoded(..) => false, + } + } + + /// Returns the column names, in the order the query returned them. + #[must_use] + pub const fn columns(&self) -> &[Box] { + &self.columns + } + + /// Returns each row's values, in the order of [`columns`](Self::columns). + pub fn iter(&self) -> impl ExactSizeIterator { + self.rows.iter().map(|row| &**row) + } +} diff --git a/libs/@local/integrations/engine/src/workspace/select.rs b/libs/@local/integrations/engine/src/workspace/select.rs new file mode 100644 index 00000000000..433b54eaf2a --- /dev/null +++ b/libs/@local/integrations/engine/src/workspace/select.rs @@ -0,0 +1,92 @@ +//! The check that SQL is a single `SELECT` statement whose table functions are allowed. +//! +//! DuckDB's `json_serialize_sql` reads the SQL with DuckDB's own parser. It refuses every +//! statement that is not a `SELECT`, and returns the parsed statements otherwise. The check walks +//! them for table functions, because a `SELECT` can call table functions that change settings, +//! checkpoint the database or run further SQL, such as `enable_logging`, `checkpoint` and `query`. + +use error_stack::{Report, ResultExt as _}; +use serde_json::Value; + +use crate::workspace::StatementError; + +/// The table functions a query can call. +const TABLE_FUNCTIONS: [&str; 3] = ["generate_series", "range", "unnest"]; + +/// SQL that DuckDB reads as a single `SELECT` statement whose table functions are allowed. +#[derive(Debug, Clone, Copy)] +pub(super) struct Select<'sql>(&'sql str); + +impl<'sql> Select<'sql> { + /// The query that returns DuckDB's reading of the SQL given as its parameter. + pub(super) const SERIALIZE: &'static str = "SELECT json_serialize_sql(?::VARCHAR)"; + + /// Checks `sql` against `serialized`, the result of [`SERIALIZE`](Self::SERIALIZE) for it. + /// + /// # Errors + /// + /// - [`NotSelect`] if DuckDB refused the SQL + /// - [`StatementCount`] if the SQL is not exactly one statement + /// - [`TableFunction`] if the statement calls a table function that is not allowed + /// - [`Failed`] if `serialized` is not JSON, or is nested too deeply to read + /// + /// [`NotSelect`]: StatementError::NotSelect + /// [`StatementCount`]: StatementError::StatementCount + /// [`TableFunction`]: StatementError::TableFunction + /// [`Failed`]: StatementError::Failed + pub(super) fn check(sql: &'sql str, serialized: &str) -> Result> { + let serialized: Value = + serde_json::from_str(serialized).change_context(StatementError::Failed)?; + if serialized.get("error") != Some(&Value::Bool(false)) { + return Err(Report::new(StatementError::NotSelect { + reason: serialized + .get("error_message") + .and_then(Value::as_str) + .unwrap_or_default() + .into(), + })); + } + + let statements = serialized + .get("statements") + .and_then(Value::as_array) + .map_or(&[][..], Vec::as_slice); + if statements.len() != 1 { + return Err(Report::new(StatementError::StatementCount { + count: statements.len(), + })); + } + if let Some(name) = statements.iter().find_map(Self::disallowed_table_function) { + return Err(Report::new(StatementError::TableFunction { name })); + } + Ok(Self(sql)) + } + + /// Returns the name of the first table function under `node` that is not allowed. + fn disallowed_table_function(node: &Value) -> Option> { + match node { + Value::Object(fields) => { + if fields.get("type").and_then(Value::as_str) == Some("TABLE_FUNCTION") { + let name = fields + .get("function") + .and_then(|function| function.get("function_name")) + .and_then(Value::as_str) + .unwrap_or_default(); + if !TABLE_FUNCTIONS + .iter() + .any(|allowed| allowed.eq_ignore_ascii_case(name)) + { + return Some(name.into()); + } + } + fields.values().find_map(Self::disallowed_table_function) + } + Value::Array(items) => items.iter().find_map(Self::disallowed_table_function), + Value::Null | Value::Bool(_) | Value::Number(_) | Value::String(_) => None, + } + } + + pub(super) const fn as_str(self) -> &'sql str { + self.0 + } +} diff --git a/libs/@local/integrations/engine/tests/workspace.rs b/libs/@local/integrations/engine/tests/workspace.rs new file mode 100644 index 00000000000..7144ea1238c --- /dev/null +++ b/libs/@local/integrations/engine/tests/workspace.rs @@ -0,0 +1,434 @@ +#![expect(missing_docs, reason = "integration tests")] + +use core::{future::poll_fn, pin::Pin, task::Poll, time::Duration}; +use std::path::PathBuf; + +use duckdb::types::{TimeUnit, Value}; +use error_stack::Report; +use hash_integrations_engine::{ + DataErrorType, OpenError, QueryErrorType, StatementError, Workspace, +}; +use pretty_assertions::assert_eq; +use tempfile::TempDir; + +fn database_path(directory: &TempDir) -> PathBuf { + directory.path().join("workspace.duckdb") +} + +async fn open(directory: &TempDir) -> Workspace { + Workspace::open(database_path(directory)) + .await + .expect("the workspace should open") +} + +async fn first_value(workspace: &Workspace, sql: &str) -> Result> { + let rows = workspace.query(sql.into()).await?; + let row = rows.iter().next().expect("the query should return a row"); + Ok(row.first().expect("the row should have a value").clone()) +} + +/// Polls `request` once, which sends it to the workspace thread, and checks that it still waits +/// for its reply. +async fn send(mut request: Pin<&mut impl Future>) { + assert!( + poll_fn(|context| Poll::Ready(request.as_mut().poll(context).is_pending())).await, + "the request should still be waiting for its reply" + ); +} + +#[tokio::test] +async fn query_returns_rows_in_column_order() { + let directory = TempDir::new().expect("should create a directory"); + let workspace = open(&directory).await; + + workspace + .create_table( + "flights".into(), + "SELECT * FROM (VALUES ('BA117', 214), ('LH400', 364)) AS flights (code, seats)".into(), + ) + .await + .expect("the table should be created"); + let rows = workspace + .query("SELECT seats, code FROM flights ORDER BY code".into()) + .await + .expect("the query should run"); + + assert_eq!( + rows.columns(), + [Box::from("seats"), Box::from("code")], + "the columns should be in the selected order" + ); + assert_eq!( + rows.iter().collect::>(), + [ + [Value::Int(214), Value::Text("BA117".to_owned())], + [Value::Int(364), Value::Text("LH400".to_owned())], + ], + "the rows should hold the table's values in the selected column order" + ); +} + +#[tokio::test] +async fn query_without_rows_keeps_columns() { + let directory = TempDir::new().expect("should create a directory"); + let workspace = open(&directory).await; + + let rows = workspace + .query("SELECT 1 AS id, 'a' AS name WHERE false".into()) + .await + .expect("the query should run"); + + assert_eq!( + rows.columns(), + [Box::from("id"), Box::from("name")], + "the columns should be named although no row matched" + ); + assert_eq!(rows.iter().len(), 0, "the query should return no rows"); +} + +#[tokio::test] +async fn create_table_replaces_table() { + let directory = TempDir::new().expect("should create a directory"); + let workspace = open(&directory).await; + + for sql in ["SELECT 'BA117' AS code", "SELECT 2 AS seats;"] { + workspace + .create_table("departure \"board\"".into(), sql.into()) + .await + .expect("the table should be created"); + } + + let rows = workspace + .query("SELECT * FROM \"departure \"\"board\"\"\"".into()) + .await + .expect("the query should run"); + assert_eq!( + rows.columns(), + [Box::from("seats")], + "the second query should replace the table" + ); +} + +#[tokio::test] +async fn query_failure_keeps_workspace_open() { + let directory = TempDir::new().expect("should create a directory"); + let workspace = open(&directory).await; + + let report = workspace + .query("SELECT * FROM missing".into()) + .await + .expect_err("a query of a missing table should fail"); + + assert!( + matches!( + report.current_context(), + StatementError::InvalidQuery { + error_type: QueryErrorType::Catalog + } + ), + "the missing table should be reported, not {report:?}" + ); + assert!( + report.downcast_ref::().is_some(), + "the DuckDB error should stay in the report, not {report:?}" + ); + assert_eq!( + first_value(&workspace, "SELECT 1") + .await + .expect("the next query should run"), + Value::Int(1), + "the next query should return its value" + ); +} + +#[tokio::test] +async fn query_reports_invalid_data() { + let directory = TempDir::new().expect("should create a directory"); + let workspace = open(&directory).await; + + let report = workspace + .query("SELECT 'BA117'::INTEGER".into()) + .await + .expect_err("a cast of text that is not a number should fail"); + + assert!( + matches!( + report.current_context(), + StatementError::InvalidData { + error_type: DataErrorType::Conversion + } + ), + "the value should be reported, not {report:?}" + ); +} + +#[tokio::test] +async fn query_refuses_other_statements() { + let directory = TempDir::new().expect("should create a directory"); + let workspace = open(&directory).await; + + for sql in [ + "CREATE TABLE flights (code TEXT)", + "CALL enable_logging()", + "SELECT 1; DROP TABLE flights", + "SELEC 1", + ] { + let report = workspace + .query(sql.into()) + .await + .expect_err("only a `SELECT` should run"); + assert!( + matches!( + report.current_context(), + StatementError::NotSelect { reason } if !reason.is_empty() + ), + "`{sql}` should be refused with DuckDB's reason, not by {report:?}" + ); + } +} + +#[tokio::test] +async fn create_table_refuses_other_statement_counts() { + let directory = TempDir::new().expect("should create a directory"); + let workspace = open(&directory).await; + + for (sql, count) in [("SELECT 1; SELECT 2", 2), ("", 0), ("-- nothing", 0)] { + let report = workspace + .create_table("flights".into(), sql.into()) + .await + .expect_err("SQL that is not one statement should be refused"); + assert!( + matches!( + report.current_context(), + StatementError::StatementCount { count: found } if *found == count + ), + "`{sql}` should be refused as {count} statements, not by {report:?}" + ); + } + workspace + .create_table("flights".into(), "SELECT 1; -- one statement".into()) + .await + .expect("a trailing `;` and comment should be accepted"); +} + +#[tokio::test] +async fn query_refuses_disallowed_table_functions() { + let directory = TempDir::new().expect("should create a directory"); + let workspace = open(&directory).await; + + for (sql, name) in [ + ("SELECT * FROM enable_logging()", "enable_logging"), + ("SELECT * FROM checkpoint()", "checkpoint"), + ("SELECT * FROM query('SELECT 1')", "query"), + ("SELECT * FROM read_csv('flights.csv')", "read_csv"), + ( + "SELECT 1 WHERE EXISTS (SELECT * FROM force_checkpoint())", + "force_checkpoint", + ), + ( + "WITH logs AS (SELECT * FROM truncate_duckdb_logs()) SELECT 1", + "truncate_duckdb_logs", + ), + ] { + let report = workspace + .query(sql.into()) + .await + .expect_err("the table function should be refused"); + assert!( + matches!( + report.current_context(), + StatementError::TableFunction { name: found } if &**found == name + ), + "`{sql}` should be refused for calling `{name}`, not by {report:?}" + ); + } + + assert_eq!( + first_value( + &workspace, + "SELECT count(*) FROM RANGE(2), generate_series(1, 2), unnest([1, 2])" + ) + .await + .expect("the allowed table functions should run"), + Value::BigInt(8), + "the query should count every combination" + ); +} + +#[tokio::test] +async fn open_locks_settings() { + let directory = TempDir::new().expect("should create a directory"); + let workspace = open(&directory).await; + + let rows = workspace + .query( + "SELECT current_setting('allow_community_extensions'), \ + current_setting('autoinstall_known_extensions'), \ + current_setting('autoload_known_extensions'), \ + current_setting('enable_external_access'), current_setting('lock_configuration')" + .into(), + ) + .await + .expect("the settings should be readable"); + + assert_eq!( + rows.iter().next(), + Some( + [ + Value::Boolean(false), + Value::Boolean(false), + Value::Boolean(false), + Value::Boolean(false), + Value::Boolean(true), + ] + .as_slice() + ), + "the workspace should disable extensions and file access, and lock the settings" + ); +} + +#[tokio::test] +async fn query_reads_supported_types() { + let directory = TempDir::new().expect("should create a directory"); + let workspace = open(&directory).await; + + let rows = workspace + .query( + "SELECT TIMESTAMP '2024-01-01 12:00', 1.5::DECIMAL(4, 1), 1::HUGEINT, DATE \ + '2024-01-01', '12:00'::TIME, INTERVAL 1 DAY, '\\x01'::BLOB, 'a'::ENUM('a', 'b'), \ + [1], {'a': 1}, MAP {'a': 1}, 1::UNION(n INTEGER, s TEXT)" + .into(), + ) + .await + .expect("every column type should be readable"); + + assert_eq!( + rows.iter().next().map(<[Value]>::len), + Some(12), + "the row should hold every value" + ); +} + +#[tokio::test] +async fn query_refuses_unsupported_types() { + let directory = TempDir::new().expect("should create a directory"); + let workspace = open(&directory).await; + + for sql in [ + "SELECT '12:00'::TIME_NS AS departs", + "SELECT ['12:00'::TIME_NS] AS departs", + "SELECT {'at': '12:00'::TIME_NS} AS departs", + "SELECT MAP {'at': '12:00'::TIME_NS} AS departs", + ] { + let report = workspace + .query(sql.into()) + .await + .expect_err("the column type should be refused"); + assert!( + matches!( + report.current_context(), + StatementError::UnsupportedType { column, .. } if &**column == "departs" + ), + "`{sql}` should be refused for its column type, not by {report:?}" + ); + } + assert_eq!( + first_value(&workspace, "SELECT '12:00'::TIME") + .await + .expect("the workspace should stay open"), + Value::Time64(TimeUnit::Microsecond, 43_200_000_000), + "a supported time should be read" + ); +} + +#[tokio::test] +async fn open_refuses_second_workspace() { + let directory = TempDir::new().expect("should create a directory"); + let workspace = open(&directory).await; + + let report = Workspace::open(database_path(&directory)) + .await + .expect_err("a second workspace should not open the database"); + assert!( + matches!(report.current_context(), OpenError::Locked { .. }), + "the lock should be reported, not {report:?}" + ); + + workspace.close().await.expect("the workspace should close"); + open(&directory).await; +} + +#[tokio::test] +async fn dropping_request_interrupts_query() { + let directory = TempDir::new().expect("should create a directory"); + let workspace = open(&directory).await; + + let mut query = Box::pin(workspace.query("SELECT count(*) FROM range(1000000000000)".into())); + send(query.as_mut()).await; + tokio::time::sleep(Duration::from_millis(100)).await; + drop(query); + + assert_eq!( + tokio::time::timeout(Duration::from_secs(10), first_value(&workspace, "SELECT 1")) + .await + .expect("the next query should run once the dropped one is interrupted") + .expect("the next query should succeed"), + Value::Int(1), + "the next query should return its value" + ); +} + +#[tokio::test] +async fn dropping_queued_request_skips_it() { + let directory = TempDir::new().expect("should create a directory"); + let workspace = open(&directory).await; + + let mut running = Box::pin(workspace.query("SELECT count(*) FROM range(1000000000000)".into())); + let mut queued = Box::pin(workspace.create_table("skipped".into(), "SELECT 1 AS id".into())); + send(running.as_mut()).await; + send(queued.as_mut()).await; + drop(queued); + drop(running); + + assert_eq!( + tokio::time::timeout( + Duration::from_secs(10), + first_value( + &workspace, + "SELECT count(*) FROM information_schema.tables WHERE table_name = 'skipped'" + ) + ) + .await + .expect("the next query should run once the dropped one is interrupted") + .expect("the next query should succeed"), + Value::BigInt(0), + "the dropped request should not create its table" + ); +} + +#[tokio::test] +async fn dropping_workspace_interrupts_running_query() { + let directory = TempDir::new().expect("should create a directory"); + let workspace = open(&directory).await; + + let mut query = Box::pin(workspace.query("SELECT count(*) FROM range(1000000000000)".into())); + send(query.as_mut()).await; + tokio::time::sleep(Duration::from_millis(100)).await; + drop(query); + drop(workspace); + + // The thread releases the lock once the interrupted query returns and the database closes. + let reopened = tokio::time::timeout(Duration::from_secs(10), async { + loop { + match Workspace::open(database_path(&directory)).await { + Err(report) if matches!(report.current_context(), OpenError::Locked { .. }) => { + tokio::time::sleep(Duration::from_millis(10)).await; + } + result => break result, + } + } + }) + .await + .expect("the lock should be released once the dropped workspace interrupts its query"); + reopened.expect("the database should open once the lock is released"); +} diff --git a/libs/@local/integrations/engine/turbo.json b/libs/@local/integrations/engine/turbo.json new file mode 100644 index 00000000000..aa014a8eab0 --- /dev/null +++ b/libs/@local/integrations/engine/turbo.json @@ -0,0 +1,35 @@ +{ + "extends": ["//"], + "tasks": { + "doc:dependency-diagram": { + "command": [ + "cargo", + "run", + "-p", + "hash-repo-chores", + "--", + "dependency-diagram", + "--output", + "docs/dependency-diagram.mmd", + "--root", + "hash-integrations-engine", + "--root-deps-and-dependents", + "--link-mode", + "non-roots", + "--include-dev-deps", + "--include-build-deps", + "--logging-console-level", + "info" + ] + }, + "fix:clippy": { + "command": ["just", "clippy", "--fix"] + }, + "lint:clippy": { + "command": ["just", "clippy"] + }, + "test:unit": { + "command": ["mise", "run", "test:unit", "hash-integrations-engine"] + } + } +} From 893b8e1d38583b53042141ed30a4d6c926a23cb1 Mon Sep 17 00:00:00 2001 From: Fedor Sakharov Date: Mon, 19 Oct 2026 14:27:04 +0200 Subject: [PATCH 2/3] Bound workspace requests and simplify row mapping --- .../engine/src/workspace/connection.rs | 27 +++++++++---------- .../integrations/engine/src/workspace/mod.rs | 10 ++++--- .../integrations/engine/tests/workspace.rs | 12 ++++++--- 3 files changed, 27 insertions(+), 22 deletions(-) diff --git a/libs/@local/integrations/engine/src/workspace/connection.rs b/libs/@local/integrations/engine/src/workspace/connection.rs index 29358cbda67..e824118df95 100644 --- a/libs/@local/integrations/engine/src/workspace/connection.rs +++ b/libs/@local/integrations/engine/src/workspace/connection.rs @@ -4,11 +4,11 @@ use std::{ fs::{File, OpenOptions, TryLockError}, panic, path::PathBuf, - sync::mpsc, }; use duckdb::{Config, types::Value}; use error_stack::{Report, ResultExt as _}; +use tokio::sync::mpsc; use crate::{ sql::Identifier, @@ -114,7 +114,7 @@ impl Connection { .duckdb .prepare(select.as_str()) .map_err(StatementError::from_duckdb)?; - let mut results = statement.query([]).map_err(StatementError::from_duckdb)?; + let results = statement.query([]).map_err(StatementError::from_duckdb)?; if let Some(executed) = results.as_ref() { for index in 0..executed.column_count() { @@ -131,15 +131,14 @@ impl Connection { } } - let mut rows = Vec::new(); - while let Some(row) = results.next().map_err(StatementError::from_duckdb)? { - let values = (0..row.as_ref().column_count()) - .map(|index| row.get::<_, Value>(index)) - .collect::, _>>() - .map_err(StatementError::from_duckdb)?; - rows.push(values); - } - drop(results); + let rows = results + .mapped(|row| { + (0..row.as_ref().column_count()) + .map(|index| row.get::<_, Value>(index)) + .collect::, _>>() + }) + .collect::, _>>() + .map_err(StatementError::from_duckdb)?; let columns = statement .column_names() @@ -209,8 +208,8 @@ impl Connection { /// Runs `commands` in order until [`Command::Close`] or until the [`Workspace`] is dropped. /// /// [`Workspace`]: crate::Workspace - pub(super) fn run(self, commands: &mpsc::Receiver<(RequestId, Command)>) { - while let Ok((id, command)) = commands.recv() { + pub(super) fn run(self, commands: &mut mpsc::Receiver<(RequestId, Command)>) { + while let Some((id, command)) = commands.blocking_recv() { match command { Command::Query { sql, reply } => { self.answer(id, reply, |connection| connection.query(&sql)); @@ -239,7 +238,7 @@ impl Connection { pub(super) fn serve( path: PathBuf, ready: Reply, OpenError>, - commands: &mpsc::Receiver<(RequestId, Command)>, + commands: &mut mpsc::Receiver<(RequestId, Command)>, ) { let connection = match Self::open(path) { Ok(connection) => connection, diff --git a/libs/@local/integrations/engine/src/workspace/mod.rs b/libs/@local/integrations/engine/src/workspace/mod.rs index 0e1180a3199..b45fa4a56f7 100644 --- a/libs/@local/integrations/engine/src/workspace/mod.rs +++ b/libs/@local/integrations/engine/src/workspace/mod.rs @@ -8,9 +8,10 @@ mod select; use alloc::sync::Arc; use core::error::Error; -use std::{path::PathBuf, sync::mpsc, thread}; +use std::{path::PathBuf, thread}; use error_stack::{Report, ResultExt as _}; +use tokio::sync::mpsc; pub use self::{ error::{CloseError, DataErrorType, OpenError, QueryErrorType, Setting, StatementError}, @@ -44,6 +45,7 @@ enum Command { /// are sent. A query is a single `SELECT` statement, and the only table functions it can call are /// `range`, `generate_series` and `unnest`. Queries cannot open files, load extensions or change /// settings. +/// At most one request waits in the queue; further callers wait for space before sending theirs. /// /// Only one workspace at a time can open a database. It holds a lock on `.lock` until /// the database is closed. @@ -81,7 +83,7 @@ impl Workspace { /// [`Spawn`]: OpenError::Spawn /// [`Stopped`]: OpenError::Stopped pub async fn open(path: PathBuf) -> Result> { - let (commands, receiver) = mpsc::channel(); + let (commands, mut receiver) = mpsc::channel(1); let (ready, opened) = Reply::channel(); let span = tracing::info_span!("workspace", path = %path.display()); @@ -89,7 +91,7 @@ impl Workspace { .name("integrations-workspace".to_owned()) .spawn(move || { let _entered = span.entered(); - Connection::serve(path, ready, &receiver); + Connection::serve(path, ready, &mut receiver); }) .change_context(OpenError::Spawn)?; @@ -113,7 +115,7 @@ impl Workspace { { let request = self.interrupt.next_request(); let (reply, result) = Reply::channel(); - if self.commands.send((request, command(reply))).is_err() { + if self.commands.send((request, command(reply))).await.is_err() { return Err(Report::new(stopped)); } Pending::new(&self.interrupt, request, result) diff --git a/libs/@local/integrations/engine/tests/workspace.rs b/libs/@local/integrations/engine/tests/workspace.rs index 7144ea1238c..3aeec9f23f7 100644 --- a/libs/@local/integrations/engine/tests/workspace.rs +++ b/libs/@local/integrations/engine/tests/workspace.rs @@ -27,8 +27,7 @@ async fn first_value(workspace: &Workspace, sql: &str) -> Result) { assert!( poll_fn(|context| Poll::Ready(request.as_mut().poll(context).is_pending())).await, @@ -385,8 +384,12 @@ async fn dropping_queued_request_skips_it() { let mut running = Box::pin(workspace.query("SELECT count(*) FROM range(1000000000000)".into())); let mut queued = Box::pin(workspace.create_table("skipped".into(), "SELECT 1 AS id".into())); + let mut waiting = + Box::pin(workspace.create_table("skipped_too".into(), "SELECT 2 AS id".into())); send(running.as_mut()).await; send(queued.as_mut()).await; + send(waiting.as_mut()).await; + drop(waiting); drop(queued); drop(running); @@ -395,14 +398,15 @@ async fn dropping_queued_request_skips_it() { Duration::from_secs(10), first_value( &workspace, - "SELECT count(*) FROM information_schema.tables WHERE table_name = 'skipped'" + "SELECT count(*) FROM information_schema.tables WHERE table_name IN ('skipped', \ + 'skipped_too')" ) ) .await .expect("the next query should run once the dropped one is interrupted") .expect("the next query should succeed"), Value::BigInt(0), - "the dropped request should not create its table" + "the dropped requests should not create their tables" ); } From 44614005d24a32feffb1431a41ea3ae0229e1dba Mon Sep 17 00:00:00 2001 From: Fedor Sakharov Date: Mon, 19 Oct 2026 14:42:16 +0200 Subject: [PATCH 3/3] Make the workspace request queue capacity configurable --- libs/@local/integrations/engine/src/lib.rs | 1 + .../integrations/engine/src/workspace/mod.rs | 27 +++++++++++++++---- .../integrations/engine/tests/workspace.rs | 19 ++++++++----- 3 files changed, 36 insertions(+), 11 deletions(-) diff --git a/libs/@local/integrations/engine/src/lib.rs b/libs/@local/integrations/engine/src/lib.rs index 4c857df8771..f9ab7aa3ad3 100644 --- a/libs/@local/integrations/engine/src/lib.rs +++ b/libs/@local/integrations/engine/src/lib.rs @@ -21,4 +21,5 @@ mod workspace; pub use self::workspace::{ CloseError, DataErrorType, OpenError, QueryErrorType, Rows, Setting, StatementError, Workspace, + WorkspaceConfig, }; diff --git a/libs/@local/integrations/engine/src/workspace/mod.rs b/libs/@local/integrations/engine/src/workspace/mod.rs index b45fa4a56f7..1563f454be2 100644 --- a/libs/@local/integrations/engine/src/workspace/mod.rs +++ b/libs/@local/integrations/engine/src/workspace/mod.rs @@ -7,7 +7,7 @@ mod rows; mod select; use alloc::sync::Arc; -use core::error::Error; +use core::{error::Error, num::NonZeroUsize}; use std::{path::PathBuf, thread}; use error_stack::{Report, ResultExt as _}; @@ -23,6 +23,23 @@ use crate::workspace::{ reply::Reply, }; +/// Settings for a [`Workspace`]. +#[derive(Debug, Clone, Copy)] +pub struct WorkspaceConfig { + /// The number of requests that can wait while another runs. Defaults to 64. + /// Callers wait for queue space when it is full. + pub queue_capacity: NonZeroUsize, +} + +impl Default for WorkspaceConfig { + fn default() -> Self { + Self { + queue_capacity: NonZeroUsize::new(64) + .expect("the default queue capacity should be nonzero"), + } + } +} + /// A request to the workspace thread, with the channel for its reply. enum Command { Query { @@ -45,7 +62,7 @@ enum Command { /// are sent. A query is a single `SELECT` statement, and the only table functions it can call are /// `range`, `generate_series` and `unnest`. Queries cannot open files, load extensions or change /// settings. -/// At most one request waits in the queue; further callers wait for space before sending theirs. +/// [`WorkspaceConfig::queue_capacity`] bounds the queue; callers wait for space when it is full. /// /// Only one workspace at a time can open a database. It holds a lock on `.lock` until /// the database is closed. @@ -65,7 +82,7 @@ pub struct Workspace { } impl Workspace { - /// Opens the database at `path`, creating it if it does not exist. + /// Opens the database at `path` with `config`, creating it if it does not exist. /// /// # Errors /// @@ -82,8 +99,8 @@ impl Workspace { /// [`Open`]: OpenError::Open /// [`Spawn`]: OpenError::Spawn /// [`Stopped`]: OpenError::Stopped - pub async fn open(path: PathBuf) -> Result> { - let (commands, mut receiver) = mpsc::channel(1); + pub async fn open(path: PathBuf, config: WorkspaceConfig) -> Result> { + let (commands, mut receiver) = mpsc::channel(config.queue_capacity.get()); let (ready, opened) = Reply::channel(); let span = tracing::info_span!("workspace", path = %path.display()); diff --git a/libs/@local/integrations/engine/tests/workspace.rs b/libs/@local/integrations/engine/tests/workspace.rs index 3aeec9f23f7..ee0560386f5 100644 --- a/libs/@local/integrations/engine/tests/workspace.rs +++ b/libs/@local/integrations/engine/tests/workspace.rs @@ -1,12 +1,12 @@ #![expect(missing_docs, reason = "integration tests")] -use core::{future::poll_fn, pin::Pin, task::Poll, time::Duration}; +use core::{future::poll_fn, num::NonZeroUsize, pin::Pin, task::Poll, time::Duration}; use std::path::PathBuf; use duckdb::types::{TimeUnit, Value}; use error_stack::Report; use hash_integrations_engine::{ - DataErrorType, OpenError, QueryErrorType, StatementError, Workspace, + DataErrorType, OpenError, QueryErrorType, StatementError, Workspace, WorkspaceConfig, }; use pretty_assertions::assert_eq; use tempfile::TempDir; @@ -16,7 +16,7 @@ fn database_path(directory: &TempDir) -> PathBuf { } async fn open(directory: &TempDir) -> Workspace { - Workspace::open(database_path(directory)) + Workspace::open(database_path(directory), WorkspaceConfig::default()) .await .expect("the workspace should open") } @@ -345,7 +345,7 @@ async fn open_refuses_second_workspace() { let directory = TempDir::new().expect("should create a directory"); let workspace = open(&directory).await; - let report = Workspace::open(database_path(&directory)) + let report = Workspace::open(database_path(&directory), WorkspaceConfig::default()) .await .expect_err("a second workspace should not open the database"); assert!( @@ -380,7 +380,14 @@ async fn dropping_request_interrupts_query() { #[tokio::test] async fn dropping_queued_request_skips_it() { let directory = TempDir::new().expect("should create a directory"); - let workspace = open(&directory).await; + let workspace = Workspace::open( + database_path(&directory), + WorkspaceConfig { + queue_capacity: NonZeroUsize::MIN, + }, + ) + .await + .expect("the workspace should open"); let mut running = Box::pin(workspace.query("SELECT count(*) FROM range(1000000000000)".into())); let mut queued = Box::pin(workspace.create_table("skipped".into(), "SELECT 1 AS id".into())); @@ -424,7 +431,7 @@ async fn dropping_workspace_interrupts_running_query() { // The thread releases the lock once the interrupted query returns and the database closes. let reopened = tokio::time::timeout(Duration::from_secs(10), async { loop { - match Workspace::open(database_path(&directory)).await { + match Workspace::open(database_path(&directory), WorkspaceConfig::default()).await { Err(report) if matches!(report.current_context(), OpenError::Locked { .. }) => { tokio::time::sleep(Duration::from_millis(10)).await; }