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..f9ab7aa3ad3
--- /dev/null
+++ b/libs/@local/integrations/engine/src/lib.rs
@@ -0,0 +1,25 @@
+//! # 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,
+ WorkspaceConfig,
+};
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..e824118df95
--- /dev/null
+++ b/libs/@local/integrations/engine/src/workspace/connection.rs
@@ -0,0 +1,261 @@
+use alloc::sync::Arc;
+use core::{error::Error, panic::AssertUnwindSafe};
+use std::{
+ fs::{File, OpenOptions, TryLockError},
+ panic,
+ path::PathBuf,
+};
+
+use duckdb::{Config, types::Value};
+use error_stack::{Report, ResultExt as _};
+use tokio::sync::mpsc;
+
+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 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 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()
+ .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: &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));
+ }
+ 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: &mut 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..1563f454be2
--- /dev/null
+++ b/libs/@local/integrations/engine/src/workspace/mod.rs
@@ -0,0 +1,219 @@
+mod connection;
+mod error;
+mod files;
+mod interrupt;
+mod reply;
+mod rows;
+mod select;
+
+use alloc::sync::Arc;
+use core::{error::Error, num::NonZeroUsize};
+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},
+ rows::Rows,
+};
+use crate::workspace::{
+ connection::Connection,
+ interrupt::{Interrupt, Pending, RequestId},
+ 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 {
+ 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.
+/// [`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.
+///
+/// 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` with `config`, 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, 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());
+
+ thread::Builder::new()
+ .name("integrations-workspace".to_owned())
+ .spawn(move || {
+ let _entered = span.entered();
+ Connection::serve(path, ready, &mut 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))).await.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..ee0560386f5
--- /dev/null
+++ b/libs/@local/integrations/engine/tests/workspace.rs
@@ -0,0 +1,445 @@
+#![expect(missing_docs, reason = "integration tests")]
+
+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, WorkspaceConfig,
+};
+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), WorkspaceConfig::default())
+ .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 and checks that it is still pending.
+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), WorkspaceConfig::default())
+ .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 = 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()));
+ 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);
+
+ assert_eq!(
+ tokio::time::timeout(
+ Duration::from_secs(10),
+ first_value(
+ &workspace,
+ "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 requests should not create their tables"
+ );
+}
+
+#[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), WorkspaceConfig::default()).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"]
+ }
+ }
+}