From 70248de192c38ffd2d5b7678650a1caa7deb9050 Mon Sep 17 00:00:00 2001
From: vato007 <contact@michaelpivato.dev>
Date: Thu, 23 Jan 2025 07:48:59 +1030
Subject: [PATCH] Add most of the dynamic node host implementations

---
 Cargo.lock                         | 419 ++++++++++++++++++-----------
 Cargo.toml                         |   8 +-
 src/graph/dynamic/csv_reader.rs    | 120 +++++++--
 src/graph/dynamic/csv_readers.rs   |  21 +-
 src/graph/dynamic/csv_row.rs       |  12 +-
 src/graph/dynamic/csv_writer.rs    |  10 +-
 src/graph/dynamic/dynamic_state.rs |  14 +-
 src/graph/dynamic/mod.rs           |   2 +-
 src/graph/dynamic/read_map.rs      |  13 +-
 src/graph/dynamic/write_map.rs     |  18 +-
 wit/dynamic_node.wit               |   6 +-
 11 files changed, 427 insertions(+), 216 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 296681a..005a9b4 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -177,7 +177,7 @@ checksum = "d57d4cec3c647232e1094dc013546c0b33ce785d8aeb251e1f20dfaf8a9a13fe"
 dependencies = [
  "futures-util",
  "native-tls",
- "thiserror",
+ "thiserror 1.0.63",
  "url",
 ]
 
@@ -238,9 +238,15 @@ dependencies = [
 
 [[package]]
 name = "atoi_simd"
-version = "0.15.6"
+version = "0.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9ae037714f313c1353189ead58ef9eec30a8e8dc101b2622d461418fd59e28a9"
+checksum = "4790f9e8961209112beb783d85449b508673cf4a6a419c8449b210743ac4dbe9"
+
+[[package]]
+name = "atomic-waker"
+version = "1.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"
 
 [[package]]
 name = "autocfg"
@@ -336,7 +342,7 @@ dependencies = [
  "serde_json",
  "serde_repr",
  "serde_urlencoded",
- "thiserror",
+ "thiserror 1.0.63",
  "tokio",
  "tokio-util",
  "tower-service",
@@ -357,9 +363,9 @@ dependencies = [
 
 [[package]]
 name = "brotli"
-version = "6.0.0"
+version = "7.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "74f7971dbd9326d58187408ab83117d8ac1bb9c17b085fdacd1cf2f598719b6b"
+checksum = "cc97b8f16f944bba54f0433f07e30be199b6dc2bd25937444bbad560bcea29bd"
 dependencies = [
  "alloc-no-stdlib",
  "alloc-stdlib",
@@ -459,9 +465,9 @@ dependencies = [
 
 [[package]]
 name = "chrono-tz"
-version = "0.8.6"
+version = "0.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d59ae0466b83e838b81a54256c39d5d7c20b9d7daa10510a242d9b75abd5936e"
+checksum = "9c6ac4f2c0bf0f44e9161aec9675e1050aa4a530663c4a9e37e108fa948bca9f"
 dependencies = [
  "chrono",
  "chrono-tz-build",
@@ -470,12 +476,11 @@ dependencies = [
 
 [[package]]
 name = "chrono-tz-build"
-version = "0.2.1"
+version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "433e39f13c9a060046954e0592a8d0a4bcb1040125cbf91cb8ee58964cfb350f"
+checksum = "e94fea34d77a245229e7746bd2beb786cd2a896f306ff491fb8cecb3074b10a7"
 dependencies = [
  "parse-zoneinfo",
- "phf",
  "phf_codegen",
 ]
 
@@ -607,7 +612,7 @@ dependencies = [
  "env_logger",
  "futures",
  "futures-io",
- "itertools 0.13.0",
+ "itertools 0.14.0",
  "log",
  "nalgebra",
  "num_cpus",
@@ -647,18 +652,18 @@ dependencies = [
 
 [[package]]
 name = "cranelift-bforest"
-version = "0.115.0"
+version = "0.116.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ac89549be94911dd0e839b4a7db99e9ed29c17517e1c026f61066884c168aa3c"
+checksum = "e15d04a0ce86cb36ead88ad68cf693ffd6cda47052b9e0ac114bc47fd9cd23c4"
 dependencies = [
  "cranelift-entity",
 ]
 
 [[package]]
 name = "cranelift-bitset"
-version = "0.115.0"
+version = "0.116.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b9bd49369f76c77e34e641af85d0956869237832c118964d08bf5f51f210875a"
+checksum = "7c6e3969a7ce267259ce244b7867c5d3bc9e65b0a87e81039588dfdeaede9f34"
 dependencies = [
  "serde",
  "serde_derive",
@@ -666,9 +671,9 @@ dependencies = [
 
 [[package]]
 name = "cranelift-codegen"
-version = "0.115.0"
+version = "0.116.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fd96ce9cf8efebd7f5ab8ced5a0ce44250280bbae9f593d74a6d7effc3582a35"
+checksum = "2c22032c4cb42558371cf516bb47f26cdad1819d3475c133e93c49f50ebf304e"
 dependencies = [
  "bumpalo",
  "cranelift-bforest",
@@ -690,33 +695,33 @@ dependencies = [
 
 [[package]]
 name = "cranelift-codegen-meta"
-version = "0.115.0"
+version = "0.116.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5a68e358827afe4bfb6239fcbf6fbd5ac56206ece8a99c8f5f9bbd518773281a"
+checksum = "c904bc71c61b27fc57827f4a1379f29de64fe95653b620a3db77d59655eee0b8"
 dependencies = [
  "cranelift-codegen-shared",
 ]
 
 [[package]]
 name = "cranelift-codegen-shared"
-version = "0.115.0"
+version = "0.116.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e184c9767afbe73d50c55ec29abcf4c32f9baf0d9d22b86d58c4d55e06dee181"
+checksum = "40180f5497572f644ce88c255480981ae2ec1d7bb4d8e0c0136a13b87a2f2ceb"
 
 [[package]]
 name = "cranelift-control"
-version = "0.115.0"
+version = "0.116.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5cc7664f2a66f053e33f149e952bb5971d138e3af637f5097727ed6dc0ed95dd"
+checksum = "26d132c6d0bd8a489563472afc171759da0707804a65ece7ceb15a8c6d7dd5ef"
 dependencies = [
  "arbitrary",
 ]
 
 [[package]]
 name = "cranelift-entity"
-version = "0.115.0"
+version = "0.116.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "118597e3a9cf86c3556fa579a7a23b955fa18231651a52a77a2475d305a9cf84"
+checksum = "4b2d0d9618275474fbf679dd018ac6e009acbd6ae6850f6a67be33fb3b00b323"
 dependencies = [
  "cranelift-bitset",
  "serde",
@@ -725,9 +730,9 @@ dependencies = [
 
 [[package]]
 name = "cranelift-frontend"
-version = "0.115.0"
+version = "0.116.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7638ea1efb069a0aa18d8ee67401b6b0d19f6bfe5de5e9ede348bfc80bb0d8c7"
+checksum = "4fac41e16729107393174b0c9e3730fb072866100e1e64e80a1a963b2e484d57"
 dependencies = [
  "cranelift-codegen",
  "log",
@@ -737,15 +742,15 @@ dependencies = [
 
 [[package]]
 name = "cranelift-isle"
-version = "0.115.0"
+version = "0.116.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "15c53e1152a0b01c4ed2b1e0535602b8e86458777dd9d18b28732b16325c7dc0"
+checksum = "1ca20d576e5070044d0a72a9effc2deacf4d6aa650403189d8ea50126483944d"
 
 [[package]]
 name = "cranelift-native"
-version = "0.115.0"
+version = "0.116.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7b7d8f895444fa52dd7bdd0bed11bf007a7fb43af65a6deac8fcc4094c6372f7"
+checksum = "b8dee82f3f1f2c4cba9177f1cc5e350fe98764379bcd29340caa7b01f85076c7"
 dependencies = [
  "cranelift-codegen",
  "libc",
@@ -1135,10 +1140,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a"
 
 [[package]]
-name = "fast-float"
-version = "0.2.0"
+name = "fast-float2"
+version = "0.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "95765f67b4b18863968b4a1bd5bb576f732b29a4a28c7cd84c09fa3e2875f33c"
+checksum = "f8eb564c5c7423d25c886fb561d1e4ee69f72354d16918afa32c08811f6b6a55"
 
 [[package]]
 name = "fastrand"
@@ -1418,7 +1423,10 @@ version = "0.15.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bf151400ff0baff5465007dd2f3e717f3fe502074ca563069ce3a6629d07b289"
 dependencies = [
+ "allocator-api2",
+ "equivalent",
  "foldhash",
+ "rayon",
  "serde",
 ]
 
@@ -1698,9 +1706,9 @@ dependencies = [
 
 [[package]]
 name = "itertools"
-version = "0.13.0"
+version = "0.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186"
+checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285"
 dependencies = [
  "either",
 ]
@@ -1772,9 +1780,9 @@ checksum = "884e2677b40cc8c339eaefcb701c32ef1fd2493d71118dc0ca4b6a736c93bd67"
 
 [[package]]
 name = "libc"
-version = "0.2.155"
+version = "0.2.169"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "97b3888a4aecf77e811145cadf6eef5901f4782c53886191b2f693f24761847c"
+checksum = "b5aba8db14291edd000dfcc4d620c7ebfb122c613afb886ca8803fa4e128a20a"
 
 [[package]]
 name = "libm"
@@ -1898,9 +1906,9 @@ dependencies = [
 
 [[package]]
 name = "memmap2"
-version = "0.7.1"
+version = "0.9.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f49388d20533534cd19360ad3d6a7dadc885944aa802ba3995040c5ec11288c6"
+checksum = "fd3f7eed9d3848f8b98834af67102b720745c4ec028fcd0aa0239277e7de374f"
 dependencies = [
  "libc",
 ]
@@ -2210,16 +2218,6 @@ dependencies = [
  "windows-targets 0.52.6",
 ]
 
-[[package]]
-name = "parquet-format-safe"
-version = "0.2.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1131c54b167dd4e4799ce762e1ab01549ebb94d5bdd13e6ec1b467491c378e1f"
-dependencies = [
- "async-trait",
- "futures",
-]
-
 [[package]]
 name = "parse-display"
 version = "0.9.1"
@@ -2363,9 +2361,9 @@ dependencies = [
 
 [[package]]
 name = "polars"
-version = "0.43.1"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0e248cf2f0069277f8fe80d413cfb9240c7dd1cfa382b5674c1b4afa57222747"
+checksum = "0c0af18ae021b0396c42f39396146332957ebc4d4d25d931b4fe73509948f348"
 dependencies = [
  "getrandom 0.2.15",
  "polars-arrow",
@@ -2383,24 +2381,20 @@ dependencies = [
 
 [[package]]
 name = "polars-arrow"
-version = "0.43.1"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2981d5b2f34c84069a39fceca0d36dffeb97db8cadba101e7ea6605c8d42294d"
+checksum = "d1fd3c64d50b7f5f328e1566cab9979d4bc1ba2ff22114b301ed2ee0e518dbca"
 dependencies = [
  "ahash",
  "atoi",
- "atoi_simd",
  "bytemuck",
  "chrono",
  "chrono-tz",
  "dyn-clone",
  "either",
  "ethnum",
- "fast-float",
- "futures",
  "getrandom 0.2.15",
- "hashbrown 0.14.5",
- "itoa",
+ "hashbrown 0.15.2",
  "itoap",
  "lz4",
  "multiversion",
@@ -2410,10 +2404,10 @@ dependencies = [
  "polars-error",
  "polars-schema",
  "polars-utils",
- "ryu",
  "simdutf8",
  "streaming-iterator",
  "strength_reduce",
+ "strum_macros",
  "version_check",
  "zstd",
 ]
@@ -2430,25 +2424,31 @@ dependencies = [
 
 [[package]]
 name = "polars-compute"
-version = "0.43.1"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5a97b2a5c9b880ab7e52553c40a336fdb6e3244bf896b4d4917700defe8085d5"
+checksum = "e60822c245a870113df5a88fb184039501eda0a56bcd0c3f866406ff659df340"
 dependencies = [
+ "atoi_simd",
  "bytemuck",
+ "chrono",
  "either",
+ "fast-float2",
+ "itoa",
+ "itoap",
  "num-traits",
  "polars-arrow",
  "polars-error",
  "polars-utils",
+ "ryu",
  "strength_reduce",
  "version_check",
 ]
 
 [[package]]
 name = "polars-core"
-version = "0.43.1"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d5bc2cadcca904a9dc4d2c2b437c346712806e9a678bf17c7e94ebf622faae76"
+checksum = "4794a9e38ef2faf7e47a6f736c7f156c6fbb66cd529f82593b2d48348e422c8d"
 dependencies = [
  "ahash",
  "bitflags 2.6.0",
@@ -2458,7 +2458,9 @@ dependencies = [
  "comfy-table",
  "either",
  "hashbrown 0.14.5",
+ "hashbrown 0.15.2",
  "indexmap 2.7.0",
+ "itoa",
  "num-traits",
  "once_cell",
  "polars-arrow",
@@ -2471,31 +2473,34 @@ dependencies = [
  "rand_distr",
  "rayon",
  "regex",
- "thiserror",
+ "strum_macros",
+ "thiserror 2.0.11",
  "version_check",
  "xxhash-rust",
 ]
 
 [[package]]
 name = "polars-error"
-version = "0.43.1"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "56b0a8eb9b1e56a4640de6887d613cb4de73c4e09d491f3b779855d4c3bcb9ba"
+checksum = "100093a164bf6c001487ea528b7504f4be1a6881bcffe279bd6133e8f4b4e4f7"
 dependencies = [
  "polars-arrow-format",
  "regex",
  "simdutf8",
- "thiserror",
+ "thiserror 2.0.11",
 ]
 
 [[package]]
 name = "polars-expr"
-version = "0.43.1"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "34e9c0e8c7ba93aac64051b92dc68eac5a0e9543cf44ca784467db2c035821fe"
+checksum = "ad56c5ea4d6e0546fbc3fa35918a537b76587600a5118770ed331136249d50d8"
 dependencies = [
  "ahash",
  "bitflags 2.6.0",
+ "hashbrown 0.15.2",
+ "num-traits",
  "once_cell",
  "polars-arrow",
  "polars-compute",
@@ -2503,26 +2508,28 @@ dependencies = [
  "polars-io",
  "polars-ops",
  "polars-plan",
+ "polars-row",
  "polars-time",
  "polars-utils",
+ "rand 0.8.5",
  "rayon",
 ]
 
 [[package]]
 name = "polars-io"
-version = "0.43.1"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "454ebbebe1cb8cb4768adca44b8fc9431abc3c91d5927f6824e73f916bced911"
+checksum = "95d774d5971d2092f0588e89d2f0be524dff35ea368272c0810ba54a860e4411"
 dependencies = [
  "ahash",
  "async-trait",
  "atoi_simd",
  "bytes",
  "chrono",
- "fast-float",
+ "fast-float2",
  "futures",
  "glob",
- "hashbrown 0.14.5",
+ "hashbrown 0.15.2",
  "home",
  "itoa",
  "memchr",
@@ -2547,9 +2554,9 @@ dependencies = [
 
 [[package]]
 name = "polars-lazy"
-version = "0.43.1"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7e61c062e833d2376de0a4cf745504449215cbf499cea293cb592e674ffb39ca"
+checksum = "fa457bfa96f45cf14c33507eaa3ebcec6a8d52e7f7fc60cd23f338631369d417"
 dependencies = [
  "ahash",
  "bitflags 2.6.0",
@@ -2563,6 +2570,7 @@ dependencies = [
  "polars-ops",
  "polars-pipe",
  "polars-plan",
+ "polars-stream",
  "polars-time",
  "polars-utils",
  "rayon",
@@ -2571,9 +2579,9 @@ dependencies = [
 
 [[package]]
 name = "polars-mem-engine"
-version = "0.43.1"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c0643812829cc990e1533a5bf48c21a1b3eaa46aabf2967b0f53f99097cbc74c"
+checksum = "f73aa56fc0a4c1e9d56b4a4485800f4780ca214030d32d0150eccc44f71d6dab"
 dependencies = [
  "memmap2",
  "polars-arrow",
@@ -2590,9 +2598,9 @@ dependencies = [
 
 [[package]]
 name = "polars-ops"
-version = "0.43.1"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5ac14a136d87bea798f3db51d5987556deb2293da34bfc8b105ebffa05f6e810"
+checksum = "b267480495ffe382dab63318e3c6bf4073bb82971c8b80294d079293fece458b"
 dependencies = [
  "ahash",
  "argminmax",
@@ -2601,7 +2609,7 @@ dependencies = [
  "chrono",
  "chrono-tz",
  "either",
- "hashbrown 0.14.5",
+ "hashbrown 0.15.2",
  "hex",
  "indexmap 2.7.0",
  "memchr",
@@ -2614,15 +2622,17 @@ dependencies = [
  "polars-utils",
  "rayon",
  "regex",
+ "regex-syntax",
+ "strum_macros",
  "unicode-reverse",
  "version_check",
 ]
 
 [[package]]
 name = "polars-parquet"
-version = "0.43.1"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "491f5af321169259d5b1294c9fe8ed89faaeac34b4dec4abcedc0d1b3d11013a"
+checksum = "20237f232b1a74b1fae6b5c9bea8c440f2e5d3b5506601b038f0a7a34b84b710"
 dependencies = [
  "ahash",
  "async-stream",
@@ -2632,13 +2642,13 @@ dependencies = [
  "ethnum",
  "flate2",
  "futures",
- "hashbrown 0.14.5",
+ "hashbrown 0.15.2",
  "lz4",
  "num-traits",
- "parquet-format-safe",
  "polars-arrow",
  "polars-compute",
  "polars-error",
+ "polars-parquet-format",
  "polars-utils",
  "simdutf8",
  "snap",
@@ -2647,16 +2657,26 @@ dependencies = [
 ]
 
 [[package]]
-name = "polars-pipe"
-version = "0.43.1"
+name = "polars-parquet-format"
+version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "29215c31f599295cc0f803c42fc812cc518db6d5ed4d6c7cc03daf3976a0add5"
+checksum = "c025243dcfe8dbc57e94d9f82eb3bef10b565ab180d5b99bed87fd8aea319ce1"
+dependencies = [
+ "async-trait",
+ "futures",
+]
+
+[[package]]
+name = "polars-pipe"
+version = "0.45.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "82e3066f4fea8e55e72eba54ffe20ebdf08f63b9691aba8ea1135c3aeb9c2c7e"
 dependencies = [
  "crossbeam-channel",
  "crossbeam-queue",
  "enum_dispatch",
  "futures",
- "hashbrown 0.14.5",
+ "hashbrown 0.15.2",
  "num-traits",
  "polars-arrow",
  "polars-compute",
@@ -2674,9 +2694,9 @@ dependencies = [
 
 [[package]]
 name = "polars-plan"
-version = "0.43.1"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e3f728df4bc643492a2057a0a125c7e550cbcfe35b391444653ad294be9ab190"
+checksum = "99a3832887671df1eb326df52cbfcc47789d3d58454c1084a154b48b240175e2"
 dependencies = [
  "ahash",
  "bitflags 2.6.0",
@@ -2685,11 +2705,13 @@ dependencies = [
  "chrono",
  "chrono-tz",
  "either",
- "hashbrown 0.14.5",
+ "hashbrown 0.15.2",
  "memmap2",
+ "num-traits",
  "once_cell",
  "percent-encoding",
  "polars-arrow",
+ "polars-compute",
  "polars-core",
  "polars-io",
  "polars-ops",
@@ -2705,21 +2727,23 @@ dependencies = [
 
 [[package]]
 name = "polars-row"
-version = "0.43.1"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4eb931f0929ca7498b3ed5056357d2d364cad42cce95383a7e3983dbceb4bed1"
+checksum = "8e36350fb8a90238e02c8ece0f0c4c24f3374197e9c08c1c22cc8b9c526e6c25"
 dependencies = [
+ "bitflags 2.6.0",
  "bytemuck",
  "polars-arrow",
+ "polars-compute",
  "polars-error",
  "polars-utils",
 ]
 
 [[package]]
 name = "polars-schema"
-version = "0.43.1"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9c7e1234b942d3244024ecbac9c7f5a48a52a815f8ca4b9d075fbba16afb1a39"
+checksum = "8c6aa4913cffc522cea3ccbc0cafb350bec18fed0a1ef8d417ac88ea320d7749"
 dependencies = [
  "indexmap 2.7.0",
  "polars-error",
@@ -2729,9 +2753,9 @@ dependencies = [
 
 [[package]]
 name = "polars-sql"
-version = "0.43.1"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ce52bfd2ef1e2e18ac26d7d7ea3f9132b199cff06d975156703fa5badcfae187"
+checksum = "c62a2247028629b1db384437a9f2792488f0ddb539ec16fb46a5e2bceeba6dbc"
 dependencies = [
  "hex",
  "once_cell",
@@ -2750,10 +2774,40 @@ dependencies = [
 ]
 
 [[package]]
-name = "polars-time"
-version = "0.43.1"
+name = "polars-stream"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9925ab75e1d859ae2283ca09d7683198b0b9ff5afecd03f2c9180f3e36e35056"
+checksum = "b8cd9da4b063146c3ab7c08678a52eb9d466ade4f4c8617605a5a3ea063002c6"
+dependencies = [
+ "atomic-waker",
+ "crossbeam-deque",
+ "crossbeam-utils",
+ "futures",
+ "memmap2",
+ "parking_lot",
+ "pin-project-lite",
+ "polars-core",
+ "polars-error",
+ "polars-expr",
+ "polars-io",
+ "polars-mem-engine",
+ "polars-ops",
+ "polars-parquet",
+ "polars-plan",
+ "polars-utils",
+ "rand 0.8.5",
+ "rayon",
+ "recursive",
+ "slotmap",
+ "tokio",
+ "version_check",
+]
+
+[[package]]
+name = "polars-time"
+version = "0.45.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "12f005c3441eed1a96464305f73e197813cbae7894ff6712726a1182e31f52b4"
 dependencies = [
  "atoi",
  "bytemuck",
@@ -2762,30 +2816,33 @@ dependencies = [
  "now",
  "once_cell",
  "polars-arrow",
+ "polars-compute",
  "polars-core",
  "polars-error",
  "polars-ops",
  "polars-utils",
  "regex",
+ "strum_macros",
 ]
 
 [[package]]
 name = "polars-utils"
-version = "0.43.1"
+version = "0.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b44846e1fc6ae1dfdc7f65a37af7d270d0a6a17a58fff76716561f5b887a8ad7"
+checksum = "e0fc010eea42ad113b641aa53106e4d6e474650c73573d959a546eed0ce6d479"
 dependencies = [
  "ahash",
  "bytemuck",
  "bytes",
  "compact_str",
- "hashbrown 0.14.5",
+ "hashbrown 0.15.2",
  "indexmap 2.7.0",
  "libc",
  "memmap2",
  "num-traits",
  "once_cell",
  "polars-error",
+ "rand 0.8.5",
  "raw-cpuid",
  "rayon",
  "stacker",
@@ -2847,13 +2904,14 @@ dependencies = [
 
 [[package]]
 name = "pulley-interpreter"
-version = "28.0.0"
+version = "29.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "403a1a95f4c18a45c86c7bff13df00347afd0abcbf2e54af273c837339ffcf77"
+checksum = "62d95f8575df49a2708398182f49a888cf9dc30210fb1fd2df87c889edcee75d"
 dependencies = [
  "cranelift-bitset",
  "log",
  "sptr",
+ "wasmtime-math",
 ]
 
 [[package]]
@@ -3036,7 +3094,7 @@ checksum = "ba009ff324d1fc1b900bd1fdb31564febe58a8ccc8a6fdbb93b543d33b13ca43"
 dependencies = [
  "getrandom 0.2.15",
  "libredox",
- "thiserror",
+ "thiserror 1.0.63",
 ]
 
 [[package]]
@@ -3078,9 +3136,9 @@ dependencies = [
 
 [[package]]
 name = "regex-syntax"
-version = "0.8.4"
+version = "0.8.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7a66a03ae7c801facd77a29370b4faec201768915ac14a721ba36f20bc9c209b"
+checksum = "2b15c43186be67a4fd63bee50d0303afffcef381492ebe2c5d87f324e1b8815c"
 
 [[package]]
 name = "ring"
@@ -3530,6 +3588,15 @@ dependencies = [
  "autocfg",
 ]
 
+[[package]]
+name = "slotmap"
+version = "1.0.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dbff4acf519f630b3a3ddcfaea6c06b42174d9a44bc70c620e9ed1649d58b82a"
+dependencies = [
+ "version_check",
+]
+
 [[package]]
 name = "smallvec"
 version = "1.13.2"
@@ -3592,9 +3659,9 @@ dependencies = [
 
 [[package]]
 name = "sqlparser"
-version = "0.49.0"
+version = "0.52.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a4a404d0e14905361b918cb8afdb73605e25c1d5029312bd9785142dcb3aa49e"
+checksum = "9a875d8cd437cc8a97e9aeaeea352ec9a19aea99c23e9effb17757291de80b08"
 dependencies = [
  "log",
 ]
@@ -3646,7 +3713,7 @@ dependencies = [
  "sha2",
  "smallvec",
  "sqlformat",
- "thiserror",
+ "thiserror 1.0.63",
  "tokio",
  "tokio-stream",
  "tracing",
@@ -3730,7 +3797,7 @@ dependencies = [
  "smallvec",
  "sqlx-core",
  "stringprep",
- "thiserror",
+ "thiserror 1.0.63",
  "tracing",
  "whoami",
 ]
@@ -3768,7 +3835,7 @@ dependencies = [
  "smallvec",
  "sqlx-core",
  "stringprep",
- "thiserror",
+ "thiserror 1.0.63",
  "tracing",
  "whoami",
 ]
@@ -3931,9 +3998,9 @@ dependencies = [
 
 [[package]]
 name = "sysinfo"
-version = "0.31.4"
+version = "0.32.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "355dbe4f8799b304b05e1b0f05fc59b2a18d36645cf169607da45bde2f69a1be"
+checksum = "4c33cd241af0f2e9e3b5c32163b873b29956890b5342e6745b917ce9d490f4af"
 dependencies = [
  "core-foundation-sys",
  "libc",
@@ -3950,9 +4017,9 @@ checksum = "c1bbb9f3c5c463a01705937a24fdabc5047929ac764b2d5b9cf681c1f5041ed5"
 
 [[package]]
 name = "target-lexicon"
-version = "0.12.16"
+version = "0.13.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "61c41af27dd6d1e27b1b16b489db798443478cef1f06a660c96db617ba5de3b1"
+checksum = "dc12939a1c9b9d391e0b7135f72fd30508b73450753e28341fed159317582a77"
 
 [[package]]
 name = "tempfile"
@@ -3996,7 +4063,7 @@ dependencies = [
  "serde",
  "serde_json",
  "serde_with",
- "thiserror",
+ "thiserror 1.0.63",
  "tokio",
  "tokio-stream",
  "tokio-tar",
@@ -4010,7 +4077,16 @@ version = "1.0.63"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c0342370b38b6a11b6cc11d6a805569958d54cfa061a29969c3b5ce2ea405724"
 dependencies = [
- "thiserror-impl",
+ "thiserror-impl 1.0.63",
+]
+
+[[package]]
+name = "thiserror"
+version = "2.0.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d452f284b73e6d76dd36758a0c8684b1d5be31f92b89d07fd5822175732206fc"
+dependencies = [
+ "thiserror-impl 2.0.11",
 ]
 
 [[package]]
@@ -4024,6 +4100,17 @@ dependencies = [
  "syn 2.0.87",
 ]
 
+[[package]]
+name = "thiserror-impl"
+version = "2.0.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "26afc1baea8a989337eeb52b6e72a039780ce45c3edfcc9c5b9d112feeb173c2"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.87",
+]
+
 [[package]]
 name = "tiberius"
 version = "0.12.3"
@@ -4044,7 +4131,7 @@ dependencies = [
  "once_cell",
  "pin-project-lite",
  "pretty-hex",
- "thiserror",
+ "thiserror 1.0.63",
  "tokio",
  "tracing",
  "uuid",
@@ -4248,6 +4335,17 @@ dependencies = [
  "once_cell",
 ]
 
+[[package]]
+name = "trait-variant"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "70977707304198400eb4835a78f6a9f928bf41bba420deb8fdb175cd965d77a7"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.87",
+]
+
 [[package]]
 name = "try-lock"
 version = "0.2.5"
@@ -4509,9 +4607,9 @@ dependencies = [
 
 [[package]]
 name = "wasmtime"
-version = "28.0.0"
+version = "29.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f639ecae347b9a2227e453a7b7671e84370a0b61f47a15e0390fe9b7725e47b3"
+checksum = "11976a250672556d1c4c04c6d5d7656ac9192ac9edc42a4587d6c21460010e69"
 dependencies = [
  "addr2line 0.24.2",
  "anyhow",
@@ -4527,7 +4625,6 @@ dependencies = [
  "indexmap 2.7.0",
  "ittapi",
  "libc",
- "libm",
  "log",
  "mach2",
  "memfd",
@@ -4546,6 +4643,7 @@ dependencies = [
  "smallvec",
  "sptr",
  "target-lexicon",
+ "trait-variant",
  "wasm-encoder 0.221.2",
  "wasmparser 0.221.2",
  "wasmtime-asm-macros",
@@ -4557,6 +4655,7 @@ dependencies = [
  "wasmtime-fiber",
  "wasmtime-jit-debug",
  "wasmtime-jit-icache-coherence",
+ "wasmtime-math",
  "wasmtime-slab",
  "wasmtime-versioned-export-macros",
  "wasmtime-winch",
@@ -4566,18 +4665,18 @@ dependencies = [
 
 [[package]]
 name = "wasmtime-asm-macros"
-version = "28.0.0"
+version = "29.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "882a18800471cfc063c8b3ccf75723784acc3fd534009ac09421f2fac2fcdcec"
+checksum = "1f178b0d125201fbe9f75beaf849bd3e511891f9e45ba216a5b620802ccf64f2"
 dependencies = [
  "cfg-if",
 ]
 
 [[package]]
 name = "wasmtime-cache"
-version = "28.0.0"
+version = "29.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "368d974999abe6095341da9b9e2c0908a6272e796001e06b7022ad60b2d19710"
+checksum = "8b1161c8f62880deea07358bc40cceddc019f1c81d46007bc390710b2fe24ffc"
 dependencies = [
  "anyhow",
  "base64 0.21.7",
@@ -4595,9 +4694,9 @@ dependencies = [
 
 [[package]]
 name = "wasmtime-component-macro"
-version = "28.0.0"
+version = "29.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "eb5c0a77c9e1927c3d471f53cc13767c3d3438e5d5ffd394e3eb31c86445fd60"
+checksum = "d74de6592ed945d0a602f71243982a304d5d02f1e501b638addf57f42d57dfaf"
 dependencies = [
  "anyhow",
  "proc-macro2",
@@ -4610,15 +4709,15 @@ dependencies = [
 
 [[package]]
 name = "wasmtime-component-util"
-version = "28.0.0"
+version = "29.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "43702ca98bf5162eca0573db691ed9ecd36d716f8c6688410fe26ec16b6f9bcb"
+checksum = "707dc7b3c112ab5a366b30cfe2fb5b2f8e6a0f682f16df96a5ec582bfe6f056e"
 
 [[package]]
 name = "wasmtime-cranelift"
-version = "28.0.0"
+version = "29.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "20070aa5b75080a8932ec328419faf841df2bc6ceb16b55b0df2b952098392a2"
+checksum = "366be722674d4bf153290fbcbc4d7d16895cc82fb3e869f8d550ff768f9e9e87"
 dependencies = [
  "anyhow",
  "cfg-if",
@@ -4633,7 +4732,7 @@ dependencies = [
  "object",
  "smallvec",
  "target-lexicon",
- "thiserror",
+ "thiserror 1.0.63",
  "wasmparser 0.221.2",
  "wasmtime-environ",
  "wasmtime-versioned-export-macros",
@@ -4641,9 +4740,9 @@ dependencies = [
 
 [[package]]
 name = "wasmtime-environ"
-version = "28.0.0"
+version = "29.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2604ddb24879d4dc1dedcb7081d7a8e017259bce916fdae097a97db52cbaab80"
+checksum = "cdadc1af7097347aa276a4f008929810f726b5b46946971c660b6d421e9994ad"
 dependencies = [
  "anyhow",
  "cpp_demangle",
@@ -4668,9 +4767,9 @@ dependencies = [
 
 [[package]]
 name = "wasmtime-fiber"
-version = "28.0.0"
+version = "29.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "98593412d2b167ebe2b59d4a17a184978a72f976b53b3a0ec05629451079ac1d"
+checksum = "ccba90d4119f081bca91190485650730a617be1fff5228f8c4757ce133d21117"
 dependencies = [
  "anyhow",
  "cc",
@@ -4683,9 +4782,9 @@ dependencies = [
 
 [[package]]
 name = "wasmtime-jit-debug"
-version = "28.0.0"
+version = "29.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2caed0122664573c2bbcde649515f9e1bc783b14f2ba74b999720cf0225e234d"
+checksum = "3e7b61488a5ee00c35c8c22de707c36c0aecacf419a3be803a6a2ba5e860f56a"
 dependencies = [
  "object",
  "rustix",
@@ -4694,9 +4793,9 @@ dependencies = [
 
 [[package]]
 name = "wasmtime-jit-icache-coherence"
-version = "28.0.0"
+version = "29.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d40d7722b9e1fbeae135715710a8a2570b1e6cf72b74dd653962d89831c6c70d"
+checksum = "ec5e8552e01692e6c2e5293171704fed8abdec79d1a6995a0870ab190e5747d1"
 dependencies = [
  "anyhow",
  "cfg-if",
@@ -4705,16 +4804,25 @@ dependencies = [
 ]
 
 [[package]]
-name = "wasmtime-slab"
-version = "28.0.0"
+name = "wasmtime-math"
+version = "29.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8579c335220b4ece9aa490a0e8b46de78cd342b195ab21ff981d095e14b52383"
+checksum = "29210ec2aa25e00f4d54605cedaf080f39ec01a872c5bd520ad04c67af1dde17"
+dependencies = [
+ "libm",
+]
+
+[[package]]
+name = "wasmtime-slab"
+version = "29.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fcb5821a96fa04ac14bc7b158bb3d5cd7729a053db5a74dad396cd513a5e5ccf"
 
 [[package]]
 name = "wasmtime-versioned-export-macros"
-version = "28.0.0"
+version = "29.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d7de0a56fb0a69b185968f2d7a9ba54750920a806470dff7ad8de91ac06d277e"
+checksum = "86ff86db216dc0240462de40c8290887a613dddf9685508eb39479037ba97b5b"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -4723,9 +4831,9 @@ dependencies = [
 
 [[package]]
 name = "wasmtime-winch"
-version = "28.0.0"
+version = "29.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "abd309943c443f5590d12f9aba9ba63c481091c955a0a14de0c2a9e0e3aaeca9"
+checksum = "fdbabfb8f20502d5e1d81092b9ead3682ae59988487aafcd7567387b7a43cf8f"
 dependencies = [
  "anyhow",
  "cranelift-codegen",
@@ -4740,9 +4848,9 @@ dependencies = [
 
 [[package]]
 name = "wasmtime-wit-bindgen"
-version = "28.0.0"
+version = "29.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "969f83022dac3435d6469edb582ceed04cfe32aa44dc3ef16e5cb55574633df8"
+checksum = "8358319c2dd1e4db79e3c1c5d3a5af84956615343f9f89f4e4996a36816e06e6"
 dependencies = [
  "anyhow",
  "heck",
@@ -4844,9 +4952,9 @@ dependencies = [
 
 [[package]]
 name = "winch-codegen"
-version = "28.0.0"
+version = "29.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9110decc2983ed94de904804dcd979ba59cbabc78a94fec6b1d8468ec513d0f6"
+checksum = "2f849ef2c5f46cb0a20af4b4487aaa239846e52e2c03f13fa3c784684552859c"
 dependencies = [
  "anyhow",
  "cranelift-codegen",
@@ -4854,6 +4962,7 @@ dependencies = [
  "regalloc2",
  "smallvec",
  "target-lexicon",
+ "thiserror 1.0.63",
  "wasmparser 0.221.2",
  "wasmtime-cranelift",
  "wasmtime-environ",
diff --git a/Cargo.toml b/Cargo.toml
index ea04c70..3c55a0c 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -17,7 +17,7 @@ serde = { version = "1", features = ["derive"] }
 clap = { version = "4", features = ["derive"] }
 anyhow = "1"
 
-itertools = "0.13.0"
+itertools = "0.14.0"
 chrono = { version = "0.4.39", features = ["default", "serde"] }
 
 rayon = "1.10.0"
@@ -25,8 +25,8 @@ tokio = { version = "1.42.0", features = ["full"] }
 sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "any"] }
 rmp-serde = "1.1"
 tempfile = "3.7"
-polars = { version = "0.43.1", features = ["lazy", "performant", "parquet", "streaming", "cse", "dtype-datetime"] }
-polars-sql = "0.43.1"
+polars = { version = "0.45.1", features = ["lazy", "performant", "parquet", "streaming", "cse", "dtype-datetime"] }
+polars-sql = "0.45.1"
 serde_json = "1.0.122"
 num_cpus = "1.16.0"
 schemars = { version = "0.8.21", features = ["chrono"] }
@@ -38,7 +38,7 @@ futures = "0.3.31"
 tokio-util = { version = "0.7.13", features = ["compat"] }
 async-trait = "0.1.83"
 testcontainers = "0.23.1"
-wasmtime = "28.0.0"
+wasmtime = "29.0.1"
 
 # More info on targets: https://doc.rust-lang.org/cargo/reference/cargo-targets.html#configuring-a-target
 [lib]
diff --git a/src/graph/dynamic/csv_reader.rs b/src/graph/dynamic/csv_reader.rs
index 2a1d44a..9389454 100644
--- a/src/graph/dynamic/csv_reader.rs
+++ b/src/graph/dynamic/csv_reader.rs
@@ -1,36 +1,124 @@
 use super::{csv_row::CsvRow, dynamic_state::{vato007::ingey::types::HostCsvReader, DynamicState, ReadMapData}};
+use csv::{Reader, StringRecord};
+use polars::datatypes::AnyValue;
+use polars::prelude::{col, lit, LazyCsvReader, LazyFileListReader};
+use std::collections::{BTreeMap, HashMap};
+use std::fs::File;
+use std::io::Read;
 
 pub struct CsvReader {
     path: String,
+    reader: Reader<File>,
+}
+
+impl CsvReader {
+    pub fn new(path: String) -> Self {
+        let reader = Reader::from_path(&path).expect("Failed to create csv reader");
+        CsvReader {
+            path,
+            reader,
+        }
+    }
 }
 
 impl HostCsvReader for DynamicState {
-    fn columns(&mut self,self_:wasmtime::component::Resource<CsvReader>,) -> wasmtime::component::__internal::Vec<wasmtime::component::__internal::String> {
-        todo!()
+    fn columns(&mut self, self_: wasmtime::component::Resource<CsvReader>) -> Vec<String> {
+        let resource = self.resources.get_mut(&self_).expect("Failed to find resource");
+        if resource.reader.has_headers() {
+            resource.reader.headers().expect("Reader says it has headers but doesn't").iter().map(|element| element.to_owned()).collect()
+        } else {
+            vec![]
+        }
     }
 
-    fn next(&mut self,self_:wasmtime::component::Resource<CsvReader>,) -> wasmtime::component::Resource<CsvRow> {
-        todo!()
+    // TODO: These next methods needs to be cleaned up badly
+    fn next(&mut self, self_: wasmtime::component::Resource<CsvReader>) -> Result<wasmtime::component::Resource<CsvRow>, String> {
+        let resource = self.resources.get_mut(&self_).expect("Failed to find resource");
+        let mut buf = StringRecord::new();
+        let result = resource.reader.read_record(&mut buf);
+        match result {
+            Ok(read) => {
+                if read {
+                    let mut record_map = BTreeMap::new();
+                    if resource.reader.has_headers() {
+                        resource.reader.headers().expect("Reader says it has headers but doesn't").iter().enumerate().for_each(|(i, name)| {
+                            record_map.insert(name.to_owned(), buf.get(i).unwrap().to_owned());
+                        });
+                    }
+                    let result = self.resources.push(CsvRow { values: record_map }).expect("");
+                    Ok(result)
+                } else {
+                    Err("No more records available to read".to_owned())
+                }
+            }
+            Err(err) => {
+                Err(err.to_string())
+            }
+        }
     }
 
-    fn has_next(&mut self,self_:wasmtime::component::Resource<CsvReader>,) -> bool {
-        todo!()
+
+    fn next_into_map(&mut self, self_: wasmtime::component::Resource<CsvReader>) -> wasmtime::component::Resource<ReadMapData> {
+        let resource = self.resources.get_mut(&self_).expect("Failed to find resource");
+        let mut buf = StringRecord::new();
+        let result = resource.reader.read_record(&mut buf);
+        let record_map = match result {
+            Ok(read) => {
+                if read {
+                    let mut record_map = HashMap::new();
+                    if resource.reader.has_headers() {
+                        resource.reader.headers().expect("Reader says it has headers but doesn't").iter().enumerate().for_each(|(i, name)| {
+                            record_map.insert(name.to_owned(), buf.get(i).unwrap().to_owned());
+                        });
+                    }
+                    record_map
+                } else {
+                    HashMap::new()
+                }
+            }
+            Err(err) => {
+                HashMap::new()
+            }
+        };
+        self.resources.push(ReadMapData { data: record_map }).expect("")
     }
 
+    fn has_next(&mut self, self_: wasmtime::component::Resource<CsvReader>) -> bool {
+        let resource = self.resources.get_mut(&self_).expect("Failed to find resource");
+        !resource.reader.is_done()
+    }
+
+    // TODO: Clean this up as well
     #[doc = " Get a row by values in one or more columns"]
-fn query(&mut self,self_:wasmtime::component::Resource<CsvReader>,values:wasmtime::component::__internal::Vec<(wasmtime::component::__internal::String,wasmtime::component::__internal::String,)>,) -> wasmtime::component::Resource<CsvRow> {
-        todo!()
+    fn query(&mut self, self_: wasmtime::component::Resource<CsvReader>, values: Vec<(String, String,)>) -> wasmtime::component::Resource<CsvRow> {
+        let resource = self.resources.get_mut(&self_).expect("Failed to find resource");
+        let mut df = LazyCsvReader::new(&resource.path).finish().expect("Failed to open file");
+        for (key, value) in values {
+            df = df.filter(col(key).eq(lit(value)));
+        }
+        let result = df.collect().expect("Failed to filter file");
+        let mut record_map = BTreeMap::new();
+
+        if let Ok(row) = result.get_row(0) {
+            for field in result.fields() {
+                let column_index = result.get_column_index(field.name()).unwrap();
+                let value: &AnyValue = row.0.get(column_index).unwrap();
+                record_map.insert(field.name.to_string(), value.to_string());
+            }
+        }
+        self.resources.push(CsvRow { values: record_map }).expect("Failed to create csv row")
     }
 
-    fn read_into_string(&mut self,self_:wasmtime::component::Resource<CsvReader>,) -> wasmtime::component::__internal::String {
-        todo!()
+    fn read_into_string(&mut self, self_: wasmtime::component::Resource<CsvReader>) -> String {
+        let resource = self.resources.get(&self_).expect("Failed to find resource");
+        let mut file = File::open(&resource.path).unwrap();
+        let mut contents = String::new();
+        file.read_to_string(&mut contents).unwrap_or(0);
+        contents
     }
 
-    fn read_into_map(&mut self,self_:wasmtime::component::Resource<CsvReader>,) -> wasmtime::component::Resource<ReadMapData> {
-        todo!()
-    }
-
-    fn drop(&mut self,rep:wasmtime::component::Resource<CsvReader>) -> wasmtime::Result<()> {
-        todo!()
+    fn drop(&mut self, rep: wasmtime::component::Resource<CsvReader>) -> wasmtime::Result<()> {
+        self.resources.delete(rep)?;
+        Ok(())
     }
 }
\ No newline at end of file
diff --git a/src/graph/dynamic/csv_readers.rs b/src/graph/dynamic/csv_readers.rs
index ac9aecb..8bc45e5 100644
--- a/src/graph/dynamic/csv_readers.rs
+++ b/src/graph/dynamic/csv_readers.rs
@@ -1,15 +1,26 @@
 use super::{csv_reader::CsvReader, dynamic_state::{vato007::ingey::types::HostCsvReaders, DynamicState}};
+use std::collections::HashMap;
+use wasmtime::component::Resource;
 
 pub struct CsvReadersData {
-
+    // Map name of reader to path
+    readers: HashMap<String, String>,
 }
 
 impl HostCsvReaders for DynamicState {
-    fn get_reader(&mut self,self_:wasmtime::component::Resource<CsvReadersData>,name:wasmtime::component::__internal::String,) -> Option<wasmtime::component::Resource<CsvReader>> {
-        todo!()
+    fn get_reader(&mut self, self_: Resource<CsvReadersData>, name: String) -> Option<Resource<CsvReader>> {
+        let resource = self.resources.get(&self_).expect("Failed to find own resource");
+        let file_path = resource.readers.get(&name);
+        if let Some(path) = file_path.cloned() {
+            let csv_reader = CsvReader::new(path);
+            self.resources.push(csv_reader).ok()
+        } else {
+            None
+        }
     }
 
-    fn drop(&mut self,rep:wasmtime::component::Resource<CsvReadersData>) -> wasmtime::Result<()> {
-        todo!()
+    fn drop(&mut self, rep: Resource<CsvReadersData>) -> wasmtime::Result<()> {
+        self.resources.delete(rep)?;
+        Ok(())
     }
 }
\ No newline at end of file
diff --git a/src/graph/dynamic/csv_row.rs b/src/graph/dynamic/csv_row.rs
index 0a3b04d..c013df5 100644
--- a/src/graph/dynamic/csv_row.rs
+++ b/src/graph/dynamic/csv_row.rs
@@ -3,31 +3,31 @@ use std::collections::BTreeMap;
 use super::dynamic_state::{vato007::ingey::types::HostCsvRow, DynamicState};
 
 pub struct CsvRow {
-    values: BTreeMap<String, String>,
+    pub values: BTreeMap<String, String>,
 }
 
 impl HostCsvRow for DynamicState {
-    fn columns(&mut self,self_:wasmtime::component::Resource<CsvRow>,) -> wasmtime::component::__internal::Vec<wasmtime::component::__internal::String> {
+    fn columns(&mut self, self_: wasmtime::component::Resource<CsvRow>) -> Vec<String> {
         let resource = self.resources.get(&self_).expect("Failed to find the required resource");
         resource.values.keys().cloned().collect()
     }
 
-    fn values(&mut self,self_:wasmtime::component::Resource<CsvRow>,) -> wasmtime::component::__internal::Vec<wasmtime::component::__internal::String> {
+    fn values(&mut self, self_: wasmtime::component::Resource<CsvRow>) -> Vec<String> {
         let resource = self.resources.get(&self_).expect("Failed to find the required resource");
         resource.values.values().cloned().collect()
     }
 
-    fn entries(&mut self,self_:wasmtime::component::Resource<CsvRow>,) -> wasmtime::component::__internal::Vec<(wasmtime::component::__internal::String,wasmtime::component::__internal::String,)> {
+    fn entries(&mut self, self_: wasmtime::component::Resource<CsvRow>) -> Vec<(String, String,)> {
         let resource = self.resources.get(&self_).expect("Failed to find the required resource");
         resource.values.keys().map(|key| (key.clone(), resource.values.get(key).unwrap().clone())).collect()
     }
 
-    fn value(&mut self,self_:wasmtime::component::Resource<CsvRow>,name:wasmtime::component::__internal::String,) -> Option<wasmtime::component::__internal::String> {
+    fn value(&mut self, self_: wasmtime::component::Resource<CsvRow>, name: String) -> Option<String> {
         let resource = self.resources.get(&self_).expect("Failed to find the required resource");
         resource.values.get(&name).cloned()
     }
 
-    fn drop(&mut self,rep:wasmtime::component::Resource<CsvRow>) -> wasmtime::Result<()> {
+    fn drop(&mut self, rep: wasmtime::component::Resource<CsvRow>) -> wasmtime::Result<()> {
         self.resources.delete(rep)?;
         Ok(())
     }
diff --git a/src/graph/dynamic/csv_writer.rs b/src/graph/dynamic/csv_writer.rs
index 37b023f..8b09e78 100644
--- a/src/graph/dynamic/csv_writer.rs
+++ b/src/graph/dynamic/csv_writer.rs
@@ -1,20 +1,18 @@
 use super::{dynamic_state::{vato007::ingey::types::HostCsvWriter, DynamicState}, write_map::WriteMap};
 
-pub struct CsvWriterData {
-
-}
+pub struct CsvWriterData {}
 
 
 impl HostCsvWriter for DynamicState {
-    fn write_map(&mut self,self_:wasmtime::component::Resource<CsvWriterData>,row:wasmtime::component::Resource<WriteMap>,) -> () {
+    fn write_map(&mut self, self_: wasmtime::component::Resource<CsvWriterData>, row: wasmtime::component::Resource<WriteMap>) -> () {
         todo!()
     }
 
-    fn write_row(&mut self,self_:wasmtime::component::Resource<CsvWriterData>,row:wasmtime::component::__internal::Vec<(wasmtime::component::__internal::String,wasmtime::component::__internal::String,)>,) -> () {
+    fn write_row(&mut self, self_: wasmtime::component::Resource<CsvWriterData>, row: Vec<(String, String,)>) -> () {
         todo!()
     }
 
-    fn drop(&mut self,rep:wasmtime::component::Resource<CsvWriterData>) -> wasmtime::Result<()> {
+    fn drop(&mut self, rep: wasmtime::component::Resource<CsvWriterData>) -> wasmtime::Result<()> {
         todo!()
     }
 }
\ No newline at end of file
diff --git a/src/graph/dynamic/dynamic_state.rs b/src/graph/dynamic/dynamic_state.rs
index 8cf965a..5585ba9 100644
--- a/src/graph/dynamic/dynamic_state.rs
+++ b/src/graph/dynamic/dynamic_state.rs
@@ -1,11 +1,11 @@
-use vato007::ingey::types::Host;
-use wasmtime::component::{bindgen, ResourceTable};
-pub use super::csv_row::CsvRow;
 pub use super::csv_reader::CsvReader;
-pub use super::write_map::WriteMap;
 pub use super::csv_readers::CsvReadersData;
+pub use super::csv_row::CsvRow;
 pub use super::csv_writer::CsvWriterData;
 pub use super::read_map::ReadMapData;
+pub use super::write_map::WriteMap;
+use vato007::ingey::types::Host;
+use wasmtime::component::{bindgen, ResourceTable};
 
 bindgen!({
     with: {
@@ -24,10 +24,8 @@ pub struct DynamicState {
 
 impl DynamicState {
     pub fn new() -> DynamicState {
-        DynamicState {resources: ResourceTable::new()}
+        DynamicState { resources: ResourceTable::new() }
     }
 }
 
-impl Host for DynamicState {
-
-}
\ No newline at end of file
+impl Host for DynamicState {}
\ No newline at end of file
diff --git a/src/graph/dynamic/mod.rs b/src/graph/dynamic/mod.rs
index e3cdbe2..fdfacc3 100644
--- a/src/graph/dynamic/mod.rs
+++ b/src/graph/dynamic/mod.rs
@@ -38,7 +38,7 @@ impl RunnableNode for DynamicNodeRunner {
         Dynamic::add_to_linker(&mut linker, |state: &mut DynamicState| state)?;
         let mut store = Store::new(
             &engine,
-            DynamicState::new()
+            DynamicState::new(),
         );
         let bindings = Dynamic::instantiate(&mut store, &component, &linker)?;
         // TODO: Instantiate readers
diff --git a/src/graph/dynamic/read_map.rs b/src/graph/dynamic/read_map.rs
index b433dee..28e4582 100644
--- a/src/graph/dynamic/read_map.rs
+++ b/src/graph/dynamic/read_map.rs
@@ -1,15 +1,18 @@
+use std::collections::HashMap;
+
 use super::dynamic_state::{vato007::ingey::types::HostReadMap, DynamicState};
 
 pub struct ReadMapData {
-
+    pub data: HashMap<String, String>,
 }
 
 impl HostReadMap for DynamicState {
-    fn get(&mut self,self_:wasmtime::component::Resource<ReadMapData>,key:wasmtime::component::__internal::String,) -> wasmtime::component::__internal::String {
-        todo!()
+    fn get(&mut self, self_: wasmtime::component::Resource<ReadMapData>, key: String) -> Option<String> {
+        self.resources.get(&self_).ok().map(|data| data.data.get(&key).cloned()).flatten()
     }
 
-    fn drop(&mut self,rep:wasmtime::component::Resource<ReadMapData>) -> wasmtime::Result<()> {
-        todo!()
+    fn drop(&mut self, rep: wasmtime::component::Resource<ReadMapData>) -> wasmtime::Result<()> {
+        self.resources.delete(rep)?;
+        Ok(())
     }
 }
\ No newline at end of file
diff --git a/src/graph/dynamic/write_map.rs b/src/graph/dynamic/write_map.rs
index 6e48214..8f9d779 100644
--- a/src/graph/dynamic/write_map.rs
+++ b/src/graph/dynamic/write_map.rs
@@ -1,19 +1,23 @@
 use super::dynamic_state::{vato007::ingey::types::HostWriteMap, DynamicState};
+use itertools::Itertools;
+use std::collections::HashMap;
 
 pub struct WriteMap {
-   
+    pub data: HashMap<String, String>,
 }
 
 impl HostWriteMap for DynamicState {
-    fn keys(&mut self,self_:wasmtime::component::Resource<WriteMap>,) -> wasmtime::component::__internal::Vec<wasmtime::component::__internal::String> {
-        todo!()
+    fn keys(&mut self, self_: wasmtime::component::Resource<WriteMap>) -> Vec<String> {
+        self.resources.get(&self_).map(|data| data.data.keys()).expect("Failed to find resource").cloned().collect_vec()
     }
 
-    fn put(&mut self,self_:wasmtime::component::Resource<WriteMap>,name:wasmtime::component::__internal::String,value:wasmtime::component::__internal::String,) -> () {
-        todo!()
+    fn put(&mut self, self_: wasmtime::component::Resource<WriteMap>, name: String, value: String) -> () {
+        let resource = self.resources.get_mut(&self_).expect("Failed to find resource");
+        resource.data.insert(name, value);
     }
 
-    fn drop(&mut self,rep:wasmtime::component::Resource<WriteMap>) -> wasmtime::Result<()> {
-        todo!()
+    fn drop(&mut self, rep: wasmtime::component::Resource<WriteMap>) -> wasmtime::Result<()> {
+        self.resources.delete(rep)?;
+        Ok(())
     }
 }
\ No newline at end of file
diff --git a/wit/dynamic_node.wit b/wit/dynamic_node.wit
index 58dfb6a..f806ea0 100644
--- a/wit/dynamic_node.wit
+++ b/wit/dynamic_node.wit
@@ -11,14 +11,14 @@ interface types {
     resource csv-reader {
         columns: func() -> list<string>;
 
-        next: func() -> csv-row;
+        next: func() -> result<csv-row, string>;
+        next-into-map: func() -> read-map;
         has-next: func() -> bool;
 
         // Get a row by values in one or more columns
         query: func(values: list<tuple<string, string>>) -> csv-row;
 
         read-into-string: func() -> string;
-        read-into-map: func() -> read-map;
     }
 
     resource write-map {
@@ -36,7 +36,7 @@ interface types {
     }
 
     resource read-map {
-        get: func(key: string) -> string;
+        get: func(key: string) -> option<string>;
     }
 }