From 9354b939caa4dd7bd222ccc0fa36e6a2f518661b Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 9 Jun 2026 04:03:02 +0800 Subject: [PATCH] refactor(zip): simplify archive extraction path (#3290) * refactor(zip): simplify archive extraction path * build(deps): align datafusion and http updates * fix(test): restore pre-commit compatibility --- Cargo.lock | 401 ++- Cargo.toml | 8 +- .../src/data_source/table_source.rs | 5 - .../src/dispatcher/parquet_table.rs | 9 +- crates/s3select-query/src/metadata/mod.rs | 10 +- crates/zip/Cargo.toml | 15 +- crates/zip/README.md | 56 +- crates/zip/benches/zip_benchmark.rs | 416 +++ crates/zip/src/lib.rs | 2226 +++++++++++------ 9 files changed, 2105 insertions(+), 1041 deletions(-) create mode 100644 crates/zip/benches/zip_benchmark.rs diff --git a/Cargo.lock b/Cargo.lock index 95bc4555d..798bbfa31 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -841,7 +841,7 @@ dependencies = [ "bytes", "fastrand", "hex", - "http 1.4.1", + "http 1.4.2", "sha1 0.10.6", "time", "tokio", @@ -904,7 +904,7 @@ dependencies = [ "bytes-utils", "fastrand", "http 0.2.12", - "http 1.4.1", + "http 1.4.2", "http-body 0.4.6", "http-body 1.0.1", "percent-encoding", @@ -939,7 +939,7 @@ dependencies = [ "hex", "hmac 0.13.0", "http 0.2.12", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "lru 0.16.4", "percent-encoding", @@ -969,7 +969,7 @@ dependencies = [ "bytes", "fastrand", "http 0.2.12", - "http 1.4.1", + "http 1.4.2", "regex-lite", "tracing", ] @@ -994,7 +994,7 @@ dependencies = [ "bytes", "fastrand", "http 0.2.12", - "http 1.4.1", + "http 1.4.2", "regex-lite", "tracing", ] @@ -1020,7 +1020,7 @@ dependencies = [ "aws-types", "fastrand", "http 0.2.12", - "http 1.4.1", + "http 1.4.2", "regex-lite", "tracing", ] @@ -1042,7 +1042,7 @@ dependencies = [ "hex", "hmac 0.13.0", "http 0.2.12", - "http 1.4.1", + "http 1.4.2", "p256 0.13.2", "percent-encoding", "sha2 0.11.0", @@ -1074,7 +1074,7 @@ dependencies = [ "bytes", "crc-fast", "hex", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "http-body-util", "md-5 0.11.0", @@ -1108,7 +1108,7 @@ dependencies = [ "bytes-utils", "futures-core", "futures-util", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "http-body-util", "percent-encoding", @@ -1127,7 +1127,7 @@ dependencies = [ "aws-smithy-runtime-api", "aws-smithy-types", "h2", - "http 1.4.1", + "http 1.4.2", "hyper", "hyper-rustls", "hyper-util", @@ -1187,7 +1187,7 @@ dependencies = [ "bytes", "fastrand", "http 0.2.12", - "http 1.4.1", + "http 1.4.2", "http-body 0.4.6", "http-body 1.0.1", "http-body-util", @@ -1208,7 +1208,7 @@ dependencies = [ "aws-smithy-types", "bytes", "http 0.2.12", - "http 1.4.1", + "http 1.4.2", "pin-project-lite", "tokio", "tracing", @@ -1234,7 +1234,7 @@ checksum = "7442cb268338f0eb8278140a107c046756aa01093d8ef5e99628d34ae09c94f5" dependencies = [ "aws-smithy-runtime-api", "aws-smithy-types", - "http 1.4.1", + "http 1.4.2", ] [[package]] @@ -1248,7 +1248,7 @@ dependencies = [ "bytes-utils", "futures-core", "http 0.2.12", - "http 1.4.1", + "http 1.4.2", "http-body 0.4.6", "http-body 1.0.1", "http-body-util", @@ -1297,7 +1297,7 @@ dependencies = [ "bytes", "form_urlencoded", "futures-util", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "http-body-util", "hyper", @@ -1328,7 +1328,7 @@ checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" dependencies = [ "bytes", "futures-core", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "http-body-util", "mime", @@ -2577,14 +2577,13 @@ checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8" [[package]] name = "datafusion" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "93db0e623840612f7f2cd757f7e8a8922064192363732c88692e0870016e141b" +checksum = "997a31e15872606a49478e670c58302094c97cb96abb0a7d60720f8e92170040" dependencies = [ "arrow", "arrow-schema", "async-trait", - "bytes", "bzip2", "chrono", "datafusion-catalog", @@ -2614,14 +2613,13 @@ dependencies = [ "datafusion-sql", "flate2", "futures", + "indexmap 2.14.0", "itertools 0.14.0", "liblzma", "log", "object_store", "parking_lot 0.12.5", "parquet", - "rand 0.9.4", - "regex", "sqlparser", "tempfile", "tokio", @@ -2632,9 +2630,9 @@ dependencies = [ [[package]] name = "datafusion-catalog" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "37cefde60b26a7f4ff61e9d2ff2833322f91df2b568d7238afe67bde5bdffb66" +checksum = "f7dd61161508f8f5fa1107774ea687bd753c22d83a32eebf963549f89de14139" dependencies = [ "arrow", "async-trait", @@ -2657,9 +2655,9 @@ dependencies = [ [[package]] name = "datafusion-catalog-listing" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "17e112307715d6a7a331111a4c2330ff54bc237183511c319e3708a4cff431fb" +checksum = "897c70f871277f9ce99aa38347be0d679bbe3e617156c4d2a8378cec8a2a0891" dependencies = [ "arrow", "async-trait", @@ -2680,34 +2678,35 @@ dependencies = [ [[package]] name = "datafusion-common" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d72a11ca44a95e1081870d3abb80c717496e8a7acb467a1d3e932bb636af5cc2" +checksum = "121c9ded5d87d9172319e006f2afdb9928d72dbacd6a90a458d8acb1e3b43a65" dependencies = [ - "ahash 0.8.12", "arrow", "arrow-ipc", + "arrow-schema", "chrono", + "foldhash 0.2.0", "half", - "hashbrown 0.16.1", + "hashbrown 0.17.1", "indexmap 2.14.0", "itertools 0.14.0", "libc", "log", "object_store", "parquet", - "paste", "recursive", "sqlparser", "tokio", + "uuid", "web-time", ] [[package]] name = "datafusion-common-runtime" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "89f4afaed29670ec4fd6053643adc749fe3f4bc9d1ce1b8c5679b22c67d12def" +checksum = "981b9dae74f78ee3d9f714fb49b01919eab975461b56149510c3ba9ea11287d1" dependencies = [ "futures", "log", @@ -2716,9 +2715,9 @@ dependencies = [ [[package]] name = "datafusion-datasource" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e9fb386e1691355355a96419978a0022b7947b44d4a24a6ea99f00b6b485cbb6" +checksum = "ffd7d295b2ec7c00d8a56562f41ed41062cf0af75549ed891c12a0a09eddfefe" dependencies = [ "arrow", "async-compression", @@ -2742,6 +2741,7 @@ dependencies = [ "liblzma", "log", "object_store", + "parking_lot 0.12.5", "rand 0.9.4", "tokio", "tokio-util", @@ -2751,9 +2751,9 @@ dependencies = [ [[package]] name = "datafusion-datasource-arrow" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ffa6c52cfed0734c5f93754d1c0175f558175248bf686c944fb05c373e5fc096" +checksum = "552b0b3f342f7ec41b3fbd70f6339dc82a30cfd0349e7f280e7852528085349f" dependencies = [ "arrow", "arrow-ipc", @@ -2775,9 +2775,9 @@ dependencies = [ [[package]] name = "datafusion-datasource-csv" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "503f29e0582c1fc189578d665ff57d9300da1f80c282777d7eb67bb79fb8cdca" +checksum = "68850aa426b897e879c8b87e512ea8124f1d0a2869a4e51808ddaaddf1bc0ada" dependencies = [ "arrow", "async-trait", @@ -2798,9 +2798,9 @@ dependencies = [ [[package]] name = "datafusion-datasource-json" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e33804749abc8d0c8cb7473228483cb8070e524c6f6086ee1b85a64debe2b3d2" +checksum = "402f93242ae08ef99139ee2c528a49d087efe88d5c7b2c3ff5480855a40ce54f" dependencies = [ "arrow", "async-trait", @@ -2815,16 +2815,15 @@ dependencies = [ "datafusion-session", "futures", "object_store", - "serde_json", "tokio", "tokio-stream", ] [[package]] name = "datafusion-datasource-parquet" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32a8e0365e0e08e8ff94d912f0ababcf9065a1a304018ba90b1fc83c855b4997" +checksum = "ffd2499c1bee0eeccf6a57156105700eeeb17bc701899ac719183c4e74231450" dependencies = [ "arrow", "async-trait", @@ -2834,6 +2833,7 @@ dependencies = [ "datafusion-datasource", "datafusion-execution", "datafusion-expr", + "datafusion-functions", "datafusion-functions-aggregate-common", "datafusion-physical-expr", "datafusion-physical-expr-adapter", @@ -2852,20 +2852,19 @@ dependencies = [ [[package]] name = "datafusion-doc" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8de6ac0df1662b9148ad3c987978b32cbec7c772f199b1d53520c8fa764a87ee" +checksum = "cb9e7e5d11130c48c8bd4e80c79a9772dd28ce6dc330baca9246205d245b9e2e" [[package]] name = "datafusion-execution" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c03c7fbdaefcca4ef6ffe425a5fc2325763bfb426599bb0bf4536466efabe709" +checksum = "37a8643ab852eb68864e1b72ae789e8066282dce48eea6347ffb0aee33d1ccc0" dependencies = [ "arrow", "arrow-buffer", "async-trait", - "chrono", "dashmap", "datafusion-common", "datafusion-expr", @@ -2881,11 +2880,12 @@ dependencies = [ [[package]] name = "datafusion-expr" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "574b9b6977fedbd2a611cbff12e5caf90f31640ad9dc5870f152836d94bad0dd" +checksum = "6932f4d71eed9c8d9341476a2b845aadfabde5495d08dbcd8fc23881f49fa7a0" dependencies = [ "arrow", + "arrow-schema", "async-trait", "chrono", "datafusion-common", @@ -2896,7 +2896,6 @@ dependencies = [ "datafusion-physical-expr-common", "indexmap 2.14.0", "itertools 0.14.0", - "paste", "recursive", "serde_json", "sqlparser", @@ -2904,22 +2903,21 @@ dependencies = [ [[package]] name = "datafusion-expr-common" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7d7c3adf3db8bf61e92eb90cb659c8e8b734593a8f7c8e12a843c7ddba24b87e" +checksum = "0225491839a31b1f7d2cb8092c2d50792e2fe1c1724e4e6d08e011f5feaf4ed2" dependencies = [ "arrow", "datafusion-common", "indexmap 2.14.0", "itertools 0.14.0", - "paste", ] [[package]] name = "datafusion-functions" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f28aa4e10384e782774b10e72aca4d93ef7b31aa653095d9d4536b0a3dbc51b6" +checksum = "14872c47bfc3d21e53ec82f57074e6987a15941c1e2f43cde4ac6ae2746634e3" dependencies = [ "arrow", "arrow-buffer", @@ -2934,26 +2932,25 @@ dependencies = [ "datafusion-expr", "datafusion-expr-common", "datafusion-macros", + "datafusion-physical-expr-common", "hex", "itertools 0.14.0", "log", - "md-5 0.10.6", + "md-5 0.11.0", "memchr", "num-traits", "rand 0.9.4", "regex", - "sha2 0.10.9", - "unicode-segmentation", + "sha2 0.11.0", "uuid", ] [[package]] name = "datafusion-functions-aggregate" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "00aa6217e56098ba84e0a338176fe52f0a84cca398021512c6c8c5eff806d0ad" +checksum = "75a2ca14e1b609be21e657e2d3130b2f446456b08393b377bb721a33952d2e09" dependencies = [ - "ahash 0.8.12", "arrow", "datafusion-common", "datafusion-doc", @@ -2963,19 +2960,18 @@ dependencies = [ "datafusion-macros", "datafusion-physical-expr", "datafusion-physical-expr-common", + "foldhash 0.2.0", "half", "log", "num-traits", - "paste", ] [[package]] name = "datafusion-functions-aggregate-common" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b511250349407db7c43832ab2de63f5557b19a20dfd236b39ca2c04468b50d47" +checksum = "1ece74ba09092d2ef9c9b54a38445450aea292a1f8b04faf531936b723a24b3c" dependencies = [ - "ahash 0.8.12", "arrow", "datafusion-common", "datafusion-expr-common", @@ -2984,9 +2980,9 @@ dependencies = [ [[package]] name = "datafusion-functions-nested" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef13a858e20d50f0a9bb5e96e7ac82b4e7597f247515bccca4fdd2992df0212a" +checksum = "3f3e3f9ee8ca59bf70518802107de6f1b88a9509efdc629fadc5de9d6b2d5ef5" dependencies = [ "arrow", "arrow-ord", @@ -3000,34 +2996,34 @@ dependencies = [ "datafusion-functions-aggregate-common", "datafusion-macros", "datafusion-physical-expr-common", - "hashbrown 0.16.1", + "hashbrown 0.17.1", "itertools 0.14.0", "itoa", "log", - "paste", + "memchr", ] [[package]] name = "datafusion-functions-table" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "72b40d3f5bbb3905f9ccb1ce9485a9595c77b69758a7c24d3ba79e334ff51e7e" +checksum = "89161dffc22cf2b50f9f4b1bee83b5221d3b4ed7c2e37fd7aa2b22a5297b3a26" dependencies = [ "arrow", "async-trait", "datafusion-catalog", "datafusion-common", "datafusion-expr", + "datafusion-physical-expr", "datafusion-physical-plan", "parking_lot 0.12.5", - "paste", ] [[package]] name = "datafusion-functions-window" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d4e88ec9d57c9b685d02f58bfee7be62d72610430ddcedb82a08e5d9925dbfb6" +checksum = "d7339345b226b3874037708bf5023ba1c2de705128f8457a095aae5ae9cb9c78" dependencies = [ "arrow", "datafusion-common", @@ -3038,14 +3034,13 @@ dependencies = [ "datafusion-physical-expr", "datafusion-physical-expr-common", "log", - "paste", ] [[package]] name = "datafusion-functions-window-common" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8307bb93519b1a91913723a1130cfafeee3f72200d870d88e91a6fc5470ede5c" +checksum = "fa84836dc2392df6f43d6a29d37fb56a8ebdc8b3f4e10ae8dc15861fd20278fb" dependencies = [ "datafusion-common", "datafusion-physical-expr-common", @@ -3053,9 +3048,9 @@ dependencies = [ [[package]] name = "datafusion-macros" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2e367e6a71051d0ebdd29b2f85d12059b38b1d1f172c6906e80016da662226bd" +checksum = "587164e03ad68732aa9e7bfe5686e3f25970d4c64fd4bd80790749840892dae5" dependencies = [ "datafusion-doc", "quote", @@ -3064,9 +3059,9 @@ dependencies = [ [[package]] name = "datafusion-optimizer" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e929015451a67f77d9d8b727b2bf3a40c4445fdef6cdc53281d7d97c76888ace" +checksum = "77f20e8cf9e8654d92f4c16b24c487353ee5bf153ffc12d5772cd399ab8cd281" dependencies = [ "arrow", "chrono", @@ -3084,11 +3079,10 @@ dependencies = [ [[package]] name = "datafusion-physical-expr" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4b1e68aba7a4b350401cfdf25a3d6f989ad898a7410164afe9ca52080244cb59" +checksum = "f015a4a82f6f7ff7e1d8d4bf3870a936752fa38b17705dfcc14adef95aa8922c" dependencies = [ - "ahash 0.8.12", "arrow", "datafusion-common", "datafusion-expr", @@ -3096,11 +3090,10 @@ dependencies = [ "datafusion-functions-aggregate-common", "datafusion-physical-expr-common", "half", - "hashbrown 0.16.1", + "hashbrown 0.17.1", "indexmap 2.14.0", "itertools 0.14.0", "parking_lot 0.12.5", - "paste", "petgraph 0.8.3", "recursive", "tokio", @@ -3108,9 +3101,9 @@ dependencies = [ [[package]] name = "datafusion-physical-expr-adapter" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ea22315f33cf2e0adc104e8ec42e285f6ed93998d565c65e82fec6a9ee9f9db4" +checksum = "51e6ffff8acdfe54e0ea15ccf38115c4a9184433b0439f42907637928d00a235" dependencies = [ "arrow", "datafusion-common", @@ -3123,26 +3116,26 @@ dependencies = [ [[package]] name = "datafusion-physical-expr-common" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b04b45ea8ad3ac2d78f2ea2a76053e06591c9629c7a603eda16c10649ecf4362" +checksum = "7967a3e171c6a4bf09474b3f7a14f1a3db13ed1714ba12156f33fcce2bba54e8" dependencies = [ - "ahash 0.8.12", "arrow", "chrono", "datafusion-common", "datafusion-expr-common", - "hashbrown 0.16.1", + "hashbrown 0.17.1", "indexmap 2.14.0", "itertools 0.14.0", "parking_lot 0.12.5", + "pin-project", ] [[package]] name = "datafusion-physical-optimizer" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7cb13397809a425918f608dfe8653f332015a3e330004ab191b4404187238b95" +checksum = "59ff803e2a96054cb6d83f35f9e60fd4f42eac515e1932bd1b2dbc91d5fcbf36" dependencies = [ "arrow", "datafusion-common", @@ -3159,12 +3152,13 @@ dependencies = [ [[package]] name = "datafusion-physical-plan" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5edc023675791af9d5fb4cc4c24abf5f7bd3bd4dcf9e5bd90ea1eff6976dcc79" +checksum = "776ee54d47d15bdb126452f9ca17b03761e3b004682914beaedd3f86eb507fbc" dependencies = [ - "ahash 0.8.12", "arrow", + "arrow-data", + "arrow-ipc", "arrow-ord", "arrow-schema", "async-trait", @@ -3179,7 +3173,7 @@ dependencies = [ "datafusion-physical-expr-common", "futures", "half", - "hashbrown 0.16.1", + "hashbrown 0.17.1", "indexmap 2.14.0", "itertools 0.14.0", "log", @@ -3191,9 +3185,9 @@ dependencies = [ [[package]] name = "datafusion-pruning" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ac8c76860e355616555081cab5968cec1af7a80701ff374510860bcd567e365a" +checksum = "d5fb9e5774660aa69c3ba93c610f175f75b65cb8c3776edb3626de8f3a4f4ee3" dependencies = [ "arrow", "datafusion-common", @@ -3202,15 +3196,14 @@ dependencies = [ "datafusion-physical-expr", "datafusion-physical-expr-common", "datafusion-physical-plan", - "itertools 0.14.0", "log", ] [[package]] name = "datafusion-session" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5412111aa48e2424ba926112e192f7a6b7e4ccb450145d25ce5ede9f19dc491e" +checksum = "15ce715fa2a61f4623cc234bcc14a3ef6a91f189128d5b14b468a6a17cdfc417" dependencies = [ "async-trait", "datafusion-common", @@ -3222,9 +3215,9 @@ dependencies = [ [[package]] name = "datafusion-sql" -version = "53.1.0" +version = "54.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fa0d133ddf8b9b3b872acac900157f783e7b879fe9a6bccf389abebbfac45ec1" +checksum = "6094ad36a3ed6d7ac87b20b479b2d0b118250f66cf997603829fdc65b44a7099" dependencies = [ "arrow", "bigdecimal", @@ -3253,7 +3246,7 @@ dependencies = [ "futures-util", "headers", "htmlescape", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "http-body-util", "libc", @@ -3643,7 +3636,7 @@ dependencies = [ "flatbuffers", "flate2", "futures", - "http 1.4.1", + "http 1.4.2", "md5", "rand 0.10.1", "rcgen", @@ -4367,7 +4360,7 @@ dependencies = [ "google-cloud-gax", "hex", "hmac 0.13.0", - "http 1.4.1", + "http 1.4.2", "jsonwebtoken", "reqwest", "rustc_version", @@ -4392,7 +4385,7 @@ dependencies = [ "futures", "google-cloud-rpc", "google-cloud-wkt", - "http 1.4.1", + "http 1.4.2", "pin-project", "rand 0.10.1", "serde", @@ -4414,7 +4407,7 @@ dependencies = [ "google-cloud-rpc", "google-cloud-wkt", "h2", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "http-body-util", "hyper", @@ -4526,7 +4519,7 @@ dependencies = [ "google-cloud-type", "google-cloud-wkt", "hex", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "md5", "percent-encoding", @@ -4606,7 +4599,7 @@ dependencies = [ "fnv", "futures-core", "futures-sink", - "http 1.4.1", + "http 1.4.2", "indexmap 2.14.0", "slab", "tokio", @@ -4693,7 +4686,7 @@ dependencies = [ "base64 0.22.1", "bytes", "headers-core", - "http 1.4.1", + "http 1.4.2", "httpdate", "mime", "sha1 0.10.6", @@ -4705,7 +4698,7 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "54b4a22553d4242c49fddb9ba998a99962b5cc6f22cb5a3482bec22522403ce4" dependencies = [ - "http 1.4.1", + "http 1.4.2", ] [[package]] @@ -4903,9 +4896,9 @@ dependencies = [ [[package]] name = "http" -version = "1.4.1" +version = "1.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8be7462df143984c4598a256ef469b251d7d7f9e271135073e78fc535414f3d0" +checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425" dependencies = [ "bytes", "itoa", @@ -4929,7 +4922,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184" dependencies = [ "bytes", - "http 1.4.1", + "http 1.4.2", ] [[package]] @@ -4940,7 +4933,7 @@ checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a" dependencies = [ "bytes", "futures-core", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "pin-project-lite", ] @@ -4986,7 +4979,7 @@ dependencies = [ "futures-channel", "futures-core", "h2", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "httparse", "httpdate", @@ -5003,7 +4996,7 @@ version = "0.27.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f" dependencies = [ - "http 1.4.1", + "http 1.4.2", "hyper", "hyper-util", "log", @@ -5038,7 +5031,7 @@ dependencies = [ "bytes", "futures-channel", "futures-util", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "hyper", "ipnet", @@ -6753,7 +6746,7 @@ dependencies = [ "base64 0.22.1", "chrono", "getrandom 0.2.17", - "http 1.4.1", + "http 1.4.2", "rand 0.8.6", "serde", "serde_json", @@ -6860,7 +6853,7 @@ dependencies = [ "futures-channel", "futures-core", "futures-util", - "http 1.4.1", + "http 1.4.2", "humantime", "itertools 0.14.0", "parking_lot 0.12.5", @@ -6916,7 +6909,7 @@ dependencies = [ "dyn-clone", "ed25519-dalek 2.2.0", "hmac 0.12.1", - "http 1.4.1", + "http 1.4.2", "itertools 0.10.5", "log", "oauth2", @@ -6977,7 +6970,7 @@ checksum = "5683015d09e2df236ef005b17f6f196f0d5f6313c4fa43a7b6a53b52776e4331" dependencies = [ "async-trait", "bytes", - "http 1.4.1", + "http 1.4.2", "opentelemetry", "reqwest", ] @@ -6989,7 +6982,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35" dependencies = [ "flate2", - "http 1.4.1", + "http 1.4.2", "opentelemetry", "opentelemetry-http", "opentelemetry-proto", @@ -8268,7 +8261,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.60.2", + "windows-sys 0.59.0", ] [[package]] @@ -8662,7 +8655,7 @@ dependencies = [ "futures-core", "futures-util", "h2", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "http-body-util", "hyper", @@ -8817,7 +8810,7 @@ dependencies = [ "async-tungstenite", "futures-io", "futures-util", - "http 1.4.1", + "http 1.4.2", "rustls-native-certs", "rustls-pki-types", "rustls-webpki", @@ -8847,7 +8840,7 @@ dependencies = [ "flume", "futures-io", "futures-util", - "http 1.4.1", + "http 1.4.2", "log", "mqttbytes-core-next", "rumqttc-core-next", @@ -9057,7 +9050,7 @@ dependencies = [ "hashbrown 0.17.1", "hex-simd", "hmac 0.13.0", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "http-body-util", "hyper", @@ -9182,7 +9175,7 @@ dependencies = [ "base64-simd", "bytes", "crc-fast", - "http 1.4.1", + "http 1.4.2", "md-5 0.11.0", "pretty_assertions", "sha1 0.11.0", @@ -9297,7 +9290,7 @@ dependencies = [ "google-cloud-storage", "hex-simd", "hmac 0.13.0", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "http-body-util", "hyper", @@ -9403,7 +9396,7 @@ dependencies = [ "anyhow", "async-trait", "futures", - "http 1.4.1", + "http 1.4.2", "metrics", "rustfs-common", "rustfs-config", @@ -9432,7 +9425,7 @@ dependencies = [ "async-trait", "base64-simd", "futures", - "http 1.4.1", + "http 1.4.2", "jsonwebtoken", "moka", "openidconnect", @@ -9537,7 +9530,7 @@ version = "1.0.0-beta.7" dependencies = [ "bytes", "futures", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "http-body-util", "hyper", @@ -9761,7 +9754,7 @@ dependencies = [ "futures-util", "hex", "hmac 0.13.0", - "http 1.4.1", + "http 1.4.2", "http-body-util", "hyper", "hyper-util", @@ -9834,7 +9827,7 @@ dependencies = [ "faster-hex", "futures", "hex-simd", - "http 1.4.1", + "http 1.4.2", "http-body-util", "md-5 0.11.0", "pin-project-lite", @@ -9903,7 +9896,7 @@ dependencies = [ "datafusion", "futures", "futures-core", - "http 1.4.1", + "http 1.4.2", "metrics", "object_store", "parking_lot 0.12.5", @@ -9944,7 +9937,7 @@ dependencies = [ "async-trait", "chrono", "futures", - "http 1.4.1", + "http 1.4.2", "metrics", "rand 0.10.1", "rmp-serde", @@ -9981,7 +9974,7 @@ version = "1.0.0-beta.7" dependencies = [ "base64-simd", "bytes", - "http 1.4.1", + "http 1.4.2", "hyper", "rustfs-utils", "s3s", @@ -10063,7 +10056,7 @@ version = "1.0.0-beta.7" dependencies = [ "async-trait", "axum", - "http 1.4.1", + "http 1.4.2", "ipnetwork", "metrics", "moka", @@ -10096,7 +10089,7 @@ dependencies = [ "hex-simd", "highway", "hmac 0.13.0", - "http 1.4.1", + "http 1.4.2", "hyper", "local-ip-address", "lz4", @@ -10125,8 +10118,12 @@ version = "1.0.0-beta.7" dependencies = [ "astral-tokio-tar", "async-compression", + "criterion", + "tempfile", + "thiserror 2.0.18", "tokio", "tokio-stream", + "zip", ] [[package]] @@ -10147,7 +10144,7 @@ dependencies = [ "anyhow", "async-trait", "bytes", - "http 1.4.1", + "http 1.4.2", "reqwest", "rustify_derive", "serde", @@ -10332,7 +10329,7 @@ dependencies = [ "futures", "hex-simd", "hmac 0.13.0", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "http-body-util", "httparse", @@ -11053,9 +11050,9 @@ dependencies = [ [[package]] name = "sqlparser" -version = "0.61.0" +version = "0.62.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dbf5ea8d4d7c808e1af1cbabebca9a2abe603bcefc22294c5b95018d53200cb7" +checksum = "13c6d1b651dc4edf07eead2a0c6c78016ce971bc2c10da5266861b13f25e7cec" dependencies = [ "log", "recursive", @@ -11247,9 +11244,9 @@ checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" [[package]] name = "suppaftp" -version = "8.0.3" +version = "8.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4275c142b5be3af2eeadd70dd368caf3b65546c8af1035839372dd7a1436127d" +checksum = "4cf00e4d8418c477a8cb3c13ae5396a68d31658e760c74280bdbd34926e3b94b" dependencies = [ "async-trait", "chrono", @@ -11786,7 +11783,7 @@ dependencies = [ "bytes", "futures-core", "futures-sink", - "http 1.4.1", + "http 1.4.2", "httparse", "rand 0.8.6", "ring", @@ -11839,7 +11836,7 @@ dependencies = [ "bytes", "flate2", "h2", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "http-body-util", "hyper", @@ -11928,7 +11925,7 @@ dependencies = [ "bytes", "futures-core", "futures-util", - "http 1.4.1", + "http 1.4.2", "http-body 1.0.1", "http-body-util", "pin-project-lite", @@ -12102,7 +12099,7 @@ checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8" dependencies = [ "bytes", "data-encoding", - "http 1.4.1", + "http 1.4.2", "httparse", "log", "rand 0.9.4", @@ -12283,7 +12280,7 @@ checksum = "30ffcc0e81025065dda612ec1e26a3d81bb16ef3062354873d17a35965d68522" dependencies = [ "async-trait", "derive_builder", - "http 1.4.1", + "http 1.4.2", "reqwest", "rustify", "rustify_derive", @@ -12714,7 +12711,7 @@ version = "0.52.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" dependencies = [ - "windows-targets 0.52.6", + "windows-targets", ] [[package]] @@ -12723,16 +12720,7 @@ version = "0.59.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" dependencies = [ - "windows-targets 0.52.6", -] - -[[package]] -name = "windows-sys" -version = "0.60.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb" -dependencies = [ - "windows-targets 0.53.5", + "windows-targets", ] [[package]] @@ -12750,31 +12738,14 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" dependencies = [ - "windows_aarch64_gnullvm 0.52.6", - "windows_aarch64_msvc 0.52.6", - "windows_i686_gnu 0.52.6", - "windows_i686_gnullvm 0.52.6", - "windows_i686_msvc 0.52.6", - "windows_x86_64_gnu 0.52.6", - "windows_x86_64_gnullvm 0.52.6", - "windows_x86_64_msvc 0.52.6", -] - -[[package]] -name = "windows-targets" -version = "0.53.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3" -dependencies = [ - "windows-link", - "windows_aarch64_gnullvm 0.53.1", - "windows_aarch64_msvc 0.53.1", - "windows_i686_gnu 0.53.1", - "windows_i686_gnullvm 0.53.1", - "windows_i686_msvc 0.53.1", - "windows_x86_64_gnu 0.53.1", - "windows_x86_64_gnullvm 0.53.1", - "windows_x86_64_msvc 0.53.1", + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", ] [[package]] @@ -12792,96 +12763,48 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" -[[package]] -name = "windows_aarch64_gnullvm" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53" - [[package]] name = "windows_aarch64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" -[[package]] -name = "windows_aarch64_msvc" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006" - [[package]] name = "windows_i686_gnu" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" -[[package]] -name = "windows_i686_gnu" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3" - [[package]] name = "windows_i686_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" -[[package]] -name = "windows_i686_gnullvm" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c" - [[package]] name = "windows_i686_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" -[[package]] -name = "windows_i686_msvc" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2" - [[package]] name = "windows_x86_64_gnu" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" -[[package]] -name = "windows_x86_64_gnu" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499" - [[package]] name = "windows_x86_64_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" -[[package]] -name = "windows_x86_64_gnullvm" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1" - [[package]] name = "windows_x86_64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" -[[package]] -name = "windows_x86_64_msvc" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650" - [[package]] name = "winnow" version = "1.0.3" diff --git a/Cargo.toml b/Cargo.toml index 95714a8db..038da9bce 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -136,7 +136,7 @@ lapin = { version = "4.10.0", default-features = false, features = ["tokio", "ru hyper = { version = "1.10.1", features = ["http2", "http1", "server"] } hyper-rustls = { version = "0.27.9", default-features = false, features = ["native-tokio", "http1", "tls12", "logging", "http2", "aws-lc-rs", "webpki-roots"] } hyper-util = { version = "0.1.20", features = ["tokio", "server-auto", "server-graceful", "tracing"] } -http = "1.4.1" +http = "1.4.2" http-body = "1.0.1" http-body-util = "0.1.3" reqwest = { version = "0.13.4", default-features = false, features = ["rustls", "charset", "http2", "system-proxy", "stream", "json", "blocking", "query", "form"] } @@ -219,7 +219,7 @@ crossbeam-queue = "0.3.12" crossbeam-channel = "0.5.15" crossbeam-deque = "0.8.6" crossbeam-utils = "0.8.21" -datafusion = "53.1.0" +datafusion = "54.0.0" derive_builder = "0.20.2" enumset = "1.1.13" faster-hex = "0.10.0" @@ -310,9 +310,9 @@ pyroscope = { version = "=2.0.5", features = ["backend-pprof-rs", "backend-jemal # FTP and SFTP libunftp = { version = "0.23.0", features = ["experimental"] } unftp-core = "0.1.0" -suppaftp = { version = "8.0.3", features = ["tokio", "tokio-rustls-aws-lc-rs"] } +suppaftp = { version = "8.0.4", features = ["tokio", "tokio-rustls-aws-lc-rs"] } rcgen = "0.14.8" -russh = { version = "0.61.2",features = ["serde"] } +russh = { version = "0.61.2", features = ["serde"] } russh-sftp = "2.3.0" # WebDAV diff --git a/crates/s3select-query/src/data_source/table_source.rs b/crates/s3select-query/src/data_source/table_source.rs index cc07659b2..72782d7e1 100644 --- a/crates/s3select-query/src/data_source/table_source.rs +++ b/crates/s3select-query/src/data_source/table_source.rs @@ -12,7 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::any::Any; use std::borrow::Cow; use std::fmt::Display; use std::sync::Arc; @@ -78,10 +77,6 @@ impl TableSourceAdapter { #[async_trait] impl TableSource for TableSourceAdapter { - fn as_any(&self) -> &dyn Any { - self - } - fn schema(&self) -> SchemaRef { self.table_handle.schema() } diff --git a/crates/s3select-query/src/dispatcher/parquet_table.rs b/crates/s3select-query/src/dispatcher/parquet_table.rs index ca65e69c2..909291a85 100644 --- a/crates/s3select-query/src/dispatcher/parquet_table.rs +++ b/crates/s3select-query/src/dispatcher/parquet_table.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::{any::Any, fmt, sync::Arc}; +use std::{fmt, sync::Arc}; use async_trait::async_trait; use datafusion::{ @@ -88,8 +88,7 @@ impl ParquetSelectTable { fn partitioned_file(&self) -> PartitionedFile { let file = PartitionedFile::new(self.object_path.clone(), self.object_size); if let Some(access_plan) = self.access_plan.as_ref() { - let extensions: Arc = access_plan.clone(); - file.with_extensions(extensions) + file.with_extension(access_plan.as_ref().clone()) } else { file } @@ -98,10 +97,6 @@ impl ParquetSelectTable { #[async_trait] impl TableProvider for ParquetSelectTable { - fn as_any(&self) -> &dyn Any { - self - } - fn schema(&self) -> SchemaRef { Arc::clone(&self.schema) } diff --git a/crates/s3select-query/src/metadata/mod.rs b/crates/s3select-query/src/metadata/mod.rs index 5b518321e..3609c5d4f 100644 --- a/crates/s3select-query/src/metadata/mod.rs +++ b/crates/s3select-query/src/metadata/mod.rs @@ -19,7 +19,7 @@ use datafusion::arrow::datatypes::DataType; use datafusion::common::Result as DFResult; use datafusion::datasource::TableProvider; use datafusion::logical_expr::var_provider::is_system_variables; -use datafusion::logical_expr::{AggregateUDF, ScalarUDF, TableSource, WindowUDF}; +use datafusion::logical_expr::{AggregateUDF, HigherOrderUDF, ScalarUDF, TableSource, WindowUDF}; use datafusion::variable::VarType; use datafusion::{ config::ConfigOptions, @@ -95,6 +95,10 @@ impl ContextProvider for MetadataProvider { self.func_manager.udaf(name).ok() } + fn get_higher_order_meta(&self, _name: &str) -> Option> { + None + } + fn get_variable_type(&self, variable_names: &[String]) -> Option { if variable_names.is_empty() { return None; @@ -130,6 +134,10 @@ impl ContextProvider for MetadataProvider { self.func_manager.udfs() } + fn higher_order_function_names(&self) -> Vec { + Vec::new() + } + fn udaf_names(&self) -> Vec { self.func_manager.udafs() } diff --git a/crates/zip/Cargo.toml b/crates/zip/Cargo.toml index 29241b06d..ca3152537 100644 --- a/crates/zip/Cargo.toml +++ b/crates/zip/Cargo.toml @@ -25,6 +25,12 @@ keywords = ["zip", "compression", "rustfs", "Minio"] categories = ["web-programming", "development-tools", "compression"] documentation = "https://docs.rs/rustfs-zip/latest/rustfs_zip/" +[lib] +doctest = false + +[[bench]] +name = "zip_benchmark" +harness = false [dependencies] async-compression = { workspace = true, features = [ @@ -38,10 +44,13 @@ async-compression = { workspace = true, features = [ tokio = { workspace = true, features = ["io-uring","fs","io-util","macros"] } tokio-stream = { workspace = true } astral-tokio-tar = { workspace = true } +thiserror = { workspace = true } +zip = { workspace = true } + +[dev-dependencies] +criterion = { workspace = true, features = ["html_reports"] } +tempfile = { workspace = true } [lints] workspace = true - -[lib] -doctest = false diff --git a/crates/zip/README.md b/crates/zip/README.md index 7fc0fc53a..78e105f5f 100644 --- a/crates/zip/README.md +++ b/crates/zip/README.md @@ -1,6 +1,6 @@ [![RustFS](https://rustfs.com/images/rustfs-github.png)](https://rustfs.com) -# RustFS Zip - Compression & Archiving +# RustFS Zip - Archive And Compression Primitives

High-performance compression and archiving for RustFS object storage @@ -17,16 +17,54 @@ ## 📖 Overview -**RustFS Zip** provides high-performance compression and archiving capabilities for the [RustFS](https://rustfs.com) distributed object storage system. For the complete RustFS experience, please visit the [main RustFS repository](https://github.com/rustfs/rustfs). +**RustFS Zip** provides archive and compression primitives for the [RustFS](https://rustfs.com) distributed object storage system. Today it is primarily used by RustFS archive extract flows to: -## ✨ Features +- identify archive/compression formats by extension +- stream tar and tar+compression inputs through async decoders +- provide small ZIP read/write helpers for local archive workflows -- Multiple compression algorithms (Zstd, LZ4, Gzip, Brotli) -- Streaming compression for memory efficiency -- Parallel processing for large files -- Archive format support (ZIP, TAR, custom formats) -- Adaptive compression with content-type detection -- Compression analytics and performance metrics +## Current Features + +- A clearer type model with: + - `CompressionCodec` for stream codecs + - `ArchiveKind` for container families + - `ArchiveFormat` for concrete archive/container combinations +- Async stream codecs for `gzip`, `bzip2`, `zlib`, `xz`, and `zstd` +- Tar archive iteration over async readers through `read_archive_entries()` / `extract_tar_entries()` +- Archive guardrails through `ArchiveLimits` for entry count, entry size, total unpacked size, and path length +- In-memory compression helpers for payload round-trip workflows +- Blocking ZIP create/extract helpers for local archive files +- ZIP helper metadata via `ZipEntry`, including: + - `compression_method` + - `archive_kind` + - `format` + - `unix_mode` +- ZIP helper options via `ZipWriteOptions`, including: + - `compression_level` + - `create_directory_entries` + +## Compatibility + +- `CompressionFormat` is retained as a compatibility layer for existing callers +- New code should prefer `ArchiveFormat`, `ArchiveKind`, and `CompressionCodec` when expressing archive semantics + +## ZIP Helper Scope + +The file-based ZIP helper APIs are best suited for: + +- local archive import/export flows +- admin-side packaging helpers +- test fixtures and tooling + +They are not intended to be a remote streaming ZIP access engine. + +## Current Boundaries + +- ZIP is supported via file-based helper APIs, not the tar-family async stream APIs +- Tar-family stream APIs are intended for `tar`, `tar.gz`, `tar.bz2`, `tar.xz`, `tar.zst`, and similar compressed tar flows +- Default archive guardrails are intentionally conservative and do not replace higher-level RustFS object-path validation +- This crate does not currently implement a general-purpose parallel archive engine +- Archive extraction safety policy remains the responsibility of the RustFS caller for object-store flows ## 📚 Documentation diff --git a/crates/zip/benches/zip_benchmark.rs b/crates/zip/benches/zip_benchmark.rs new file mode 100644 index 000000000..c7047918b --- /dev/null +++ b/crates/zip/benches/zip_benchmark.rs @@ -0,0 +1,416 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use rustfs_zip::{ + ArchiveLimits, CompressionFormat, CompressionLevel, ZipWriteOptions, create_zip_with_options, extract_tar_entries, + extract_zip_to_path_with_limits, extract_zip_with_limits, +}; +use std::hint::black_box; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use tempfile::tempdir; +use tokio::runtime::Builder; +use tokio_tar::{Builder as TarBuilder, Header}; +use zip::ZipArchive; + +fn build_runtime() -> tokio::runtime::Runtime { + Builder::new_current_thread() + .enable_all() + .build() + .expect("build tokio runtime for rustfs-zip benchmarks") +} + +async fn build_tar_payload(entry_count: usize, payload_size: usize) -> Vec { + let sink = tokio::io::duplex(64 * 1024); + let (writer, mut reader) = sink; + let write_task = tokio::spawn(async move { + let mut builder = TarBuilder::new(writer); + let payload = vec![b'a'; payload_size]; + for index in 0..entry_count { + let mut header = Header::new_gnu(); + header.set_size(payload.len() as u64); + header.set_mode(0o644); + header.set_cksum(); + builder + .append_data(&mut header, format!("entry-{index}.txt"), &payload[..]) + .await + .expect("append tar benchmark entry"); + } + builder.finish().await.expect("finish tar benchmark archive"); + }); + + let mut output = Vec::new(); + tokio::io::copy(&mut reader, &mut output) + .await + .expect("read tar benchmark archive"); + write_task.await.expect("join tar writer task"); + output +} + +async fn build_compressed_tar_payload(format: CompressionFormat, entry_count: usize, payload_size: usize) -> Vec { + let tar_payload = build_tar_payload(entry_count, payload_size).await; + rustfs_zip::Compressor::new(format) + .compress(&tar_payload) + .await + .expect("compress tar benchmark payload") +} + +fn bench_tar_family_extract(c: &mut Criterion) { + let runtime = build_runtime(); + let mut group = c.benchmark_group("zip_tar_family_extract"); + + for (name, format, entry_count, payload_size) in [ + ("tar_gzip_small_many", CompressionFormat::Gzip, 64usize, 256usize), + ("tar_zstd_medium", CompressionFormat::Zstd, 16usize, 16 * 1024usize), + ] { + let payload = runtime.block_on(build_compressed_tar_payload(format, entry_count, payload_size)); + group.throughput(Throughput::Bytes(payload.len() as u64)); + group.bench_with_input(BenchmarkId::new(name, payload.len()), &payload, |b, payload| { + b.iter(|| { + runtime.block_on(async { + let seen = Arc::new(AtomicUsize::new(0)); + let seen_ref = Arc::clone(&seen); + extract_tar_entries(std::io::Cursor::new(payload.clone()), format, move |_entry| { + let seen_ref = Arc::clone(&seen_ref); + async move { + seen_ref.fetch_add(1, Ordering::Relaxed); + Ok(()) + } + }) + .await + .expect("extract tar benchmark payload"); + black_box(seen.load(Ordering::Relaxed)); + }); + }); + }); + } + + group.finish(); +} + +fn bench_zip_helper_round_trip(c: &mut Criterion) { + let runtime = build_runtime(); + let mut group = c.benchmark_group("zip_helper_round_trip"); + + let zip_matrix = [ + ("stored_flat_32x128", CompressionLevel::Fastest, 32usize, 128usize, "flat"), + ("stored_nested_32x256", CompressionLevel::Fastest, 32usize, 256usize, "nested"), + ("stored_flat_256x128", CompressionLevel::Fastest, 256usize, 128usize, "flat"), + ("deflated_flat_32x1k", CompressionLevel::Best, 32usize, 1024usize, "flat"), + ("deflated_nested_256x1k", CompressionLevel::Best, 256usize, 1024usize, "nested"), + ("deflated_deep_1024x4k", CompressionLevel::Best, 1024usize, 4 * 1024usize, "deep"), + ]; + + for (name, compression_level, file_count, payload_size, layout) in zip_matrix { + let files = (0..file_count) + .map(|index| { + let path = match layout { + "flat" => format!("file-{index}.txt"), + "nested" => format!("batch-{}/file-{index}.txt", index % 8), + "deep" => format!("lvl1/lvl2-{}/lvl3-{}/file-{index}.txt", index % 16, index % 32), + _ => format!("file-{index}.txt"), + }; + (path, vec![b'b'; payload_size]) + }) + .collect::>(); + let total_bytes = (file_count * payload_size) as u64; + group.throughput(Throughput::Bytes(total_bytes)); + + group.bench_with_input(BenchmarkId::new(name, total_bytes), &files, |b, files| { + b.iter(|| { + let temp = tempdir().expect("create benchmark tempdir"); + let zip_path = temp.path().join("archive.zip"); + let extract_path = temp.path().join("extract"); + runtime.block_on(async { + create_zip_with_options( + &zip_path, + files.clone(), + ZipWriteOptions { + compression_level, + create_directory_entries: true, + }, + ) + .await + .expect("create zip benchmark archive"); + + let entries = extract_zip_with_limits(&zip_path, &extract_path, ArchiveLimits::default()) + .await + .expect("extract zip benchmark archive"); + black_box(entries.len()); + }); + }); + }); + } + + group.finish(); +} + +fn bench_zip_helper_hotspot_breakdown(c: &mut Criterion) { + let runtime = build_runtime(); + let mut group = c.benchmark_group("zip_helper_hotspot_breakdown"); + let files = (0..32) + .map(|index| (format!("batch/file-{index}.txt"), vec![b'c'; 256])) + .collect::>(); + let total_bytes = (32 * 256) as u64; + group.throughput(Throughput::Bytes(total_bytes)); + + group.bench_function("fs_setup_cleanup_only", |b| { + b.iter(|| { + let temp = tempdir().expect("create benchmark tempdir"); + let zip_path = temp.path().join("archive.zip"); + let extract_path = temp.path().join("extract"); + black_box((zip_path, extract_path)); + }); + }); + + group.bench_function("zip_create_only_stored_small", |b| { + b.iter(|| { + let temp = tempdir().expect("create benchmark tempdir"); + let zip_path = temp.path().join("archive.zip"); + runtime.block_on(async { + create_zip_with_options( + &zip_path, + files.clone(), + ZipWriteOptions { + compression_level: CompressionLevel::Fastest, + create_directory_entries: true, + }, + ) + .await + .expect("create zip benchmark archive"); + }); + }); + }); + + let payload_for_extract = { + let temp = tempdir().expect("create benchmark tempdir"); + let zip_path = temp.path().join("archive.zip"); + runtime.block_on(async { + create_zip_with_options( + &zip_path, + files.clone(), + ZipWriteOptions { + compression_level: CompressionLevel::Fastest, + create_directory_entries: true, + }, + ) + .await + .expect("prepare zip benchmark extract payload"); + }); + std::fs::read(&zip_path).expect("read benchmark zip payload") + }; + + group.bench_function("zip_extract_only_stored_small", |b| { + b.iter(|| { + let temp = tempdir().expect("create benchmark tempdir"); + let zip_path = temp.path().join("archive.zip"); + let extract_path = temp.path().join("extract"); + std::fs::write(&zip_path, &payload_for_extract).expect("write benchmark zip payload"); + runtime.block_on(async { + let entries = extract_zip_with_limits(&zip_path, &extract_path, ArchiveLimits::default()) + .await + .expect("extract zip benchmark archive"); + black_box(entries.len()); + }); + }); + }); + + group.bench_function("zip_extract_only_stored_small_summary_only", |b| { + b.iter(|| { + let temp = tempdir().expect("create benchmark tempdir"); + let zip_path = temp.path().join("archive.zip"); + let extract_path = temp.path().join("extract"); + std::fs::write(&zip_path, &payload_for_extract).expect("write benchmark zip payload"); + runtime.block_on(async { + let summary = extract_zip_to_path_with_limits(&zip_path, &extract_path, ArchiveLimits::default()) + .await + .expect("extract zip benchmark summary path"); + black_box(summary.entry_count); + }); + }); + }); + + group.bench_function("zip_reader_only_stored_small", |b| { + b.iter(|| { + let cursor = std::io::Cursor::new(payload_for_extract.clone()); + let mut archive = ZipArchive::new(cursor).expect("open zip archive for reader-only benchmark"); + let mut total_bytes = 0usize; + for index in 0..archive.len() { + let mut zip_file = archive.by_index(index).expect("access zip entry by index"); + let enclosed_name = zip_file + .enclosed_name() + .expect("resolve enclosed zip entry name") + .to_string_lossy() + .replace('\\', "/"); + let size = zip_file.size(); + assert!(!enclosed_name.is_empty(), "zip reader-only benchmark expects non-empty names"); + assert!( + size <= ArchiveLimits::default().max_entry_size, + "zip reader-only benchmark expects small entries" + ); + if !zip_file.is_dir() { + let mut sink = [0_u8; 256]; + let bytes_read = + std::io::Read::read(&mut zip_file, &mut sink).expect("read zip entry payload for reader-only benchmark"); + total_bytes += bytes_read; + } + } + black_box(total_bytes); + }); + }); + + group.bench_function("file_write_only_stored_small", |b| { + b.iter(|| { + let temp = tempdir().expect("create benchmark tempdir"); + let extract_path = temp.path().join("extract"); + std::fs::create_dir_all(&extract_path).expect("create extract dir for file-write-only benchmark"); + let mut total_bytes = 0usize; + for index in 0..32 { + let path = extract_path.join(format!("file-{index}.txt")); + std::fs::write(&path, [b'c'; 256]).expect("write small file for file-write-only benchmark"); + total_bytes += 256; + } + black_box(total_bytes); + }); + }); + + group.finish(); +} + +fn build_object_archive_files( + metadata_count: usize, + metadata_size: usize, + payload_count: usize, + payload_size: usize, +) -> Vec<(String, Vec)> { + let mut files = Vec::with_capacity(metadata_count * 2 + payload_count); + + for index in 0..metadata_count { + let key_prefix = format!( + "bucket-a/shard-{}/tenant-{}/dataset-{}/object-{index:04}", + index % 8, + index % 16, + index % 32 + ); + files.push(( + format!("{key_prefix}/meta.json"), + format!( + "{{\"key\":\"object-{index:04}\",\"etag\":\"{:032x}\",\"size\":{},\"content_type\":\"application/octet-stream\"}}", + index, + payload_size + ) + .into_bytes(), + )); + files.push((format!("{key_prefix}/tags.txt"), vec![b'm'; metadata_size])); + } + + for index in 0..payload_count { + let payload_prefix = format!( + "bucket-a/shard-{}/tenant-{}/dataset-{}/object-{index:04}", + index % 8, + index % 16, + index % 32 + ); + files.push((format!("{payload_prefix}/part-00000.bin"), vec![b'p'; payload_size])); + } + + files +} + +fn bench_zip_object_archive_extract(c: &mut Criterion) { + let runtime = build_runtime(); + let mut group = c.benchmark_group("zip_object_archive_extract"); + + for (name, compression_level, metadata_count, metadata_size, payload_count, payload_size) in [ + ( + "stored_metadata_heavy_384m_24p", + CompressionLevel::Fastest, + 384usize, + 192usize, + 24usize, + 32 * 1024usize, + ), + ( + "deflated_mixed_192m_32p", + CompressionLevel::Best, + 192usize, + 256usize, + 32usize, + 64 * 1024usize, + ), + ] { + let files = build_object_archive_files(metadata_count, metadata_size, payload_count, payload_size); + let total_bytes = files.iter().map(|(_, payload)| payload.len() as u64).sum::(); + let payload = { + let temp = tempdir().expect("create benchmark tempdir"); + let zip_path = temp.path().join("object-archive.zip"); + runtime.block_on(async { + create_zip_with_options( + &zip_path, + files.clone(), + ZipWriteOptions { + compression_level, + create_directory_entries: true, + }, + ) + .await + .expect("create object archive benchmark payload"); + }); + std::fs::read(&zip_path).expect("read object archive benchmark payload") + }; + + group.throughput(Throughput::Bytes(total_bytes)); + group.bench_function(BenchmarkId::new("extract_full", name), |b| { + b.iter(|| { + let temp = tempdir().expect("create benchmark tempdir"); + let zip_path = temp.path().join("archive.zip"); + let extract_path = temp.path().join("extract"); + std::fs::write(&zip_path, &payload).expect("write object archive benchmark payload"); + runtime.block_on(async { + let entries = extract_zip_with_limits(&zip_path, &extract_path, ArchiveLimits::default()) + .await + .expect("extract object archive benchmark payload"); + black_box(entries.len()); + }); + }); + }); + + group.bench_function(BenchmarkId::new("extract_summary_only", name), |b| { + b.iter(|| { + let temp = tempdir().expect("create benchmark tempdir"); + let zip_path = temp.path().join("archive.zip"); + let extract_path = temp.path().join("extract"); + std::fs::write(&zip_path, &payload).expect("write object archive benchmark payload"); + runtime.block_on(async { + let summary = extract_zip_to_path_with_limits(&zip_path, &extract_path, ArchiveLimits::default()) + .await + .expect("extract object archive benchmark summary"); + black_box(summary.file_count); + }); + }); + }); + } + + group.finish(); +} + +criterion_group!( + benches, + bench_tar_family_extract, + bench_zip_helper_round_trip, + bench_zip_helper_hotspot_breakdown, + bench_zip_object_archive_extract +); +criterion_main!(benches); diff --git a/crates/zip/src/lib.rs b/crates/zip/src/lib.rs index 5bf2d1789..127e61bb6 100644 --- a/crates/zip/src/lib.rs +++ b/crates/zip/src/lib.rs @@ -14,21 +14,86 @@ use async_compression::tokio::bufread::{BzDecoder, GzipDecoder, XzDecoder, ZlibDecoder, ZstdDecoder}; use async_compression::tokio::write::{BzEncoder, GzipEncoder, XzEncoder, ZlibEncoder, ZstdEncoder}; -use std::path::Path; +use std::collections::HashSet; +use std::future::Future; +use std::io::{Read, Write}; +use std::path::{Component, Path, PathBuf}; +use std::pin::Pin; +use std::sync::{Arc, Mutex}; +use std::task::{Context, Poll}; +use thiserror::Error; use tokio::fs::File; -use tokio::io::{self, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader, BufWriter}; +use tokio::io::{self, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader, BufWriter}; +use tokio::task::spawn_blocking; use tokio_stream::StreamExt; use tokio_tar::Archive; +use zip::{CompressionMethod, ZipArchive, ZipWriter, write::SimpleFileOptions}; -#[derive(Debug, PartialEq, Clone, Copy)] +pub type Result = std::result::Result; + +#[derive(Debug, Error)] +pub enum ZipError { + #[error("unsupported {operation} for format {format:?}")] + UnsupportedFormat { + format: CompressionFormat, + operation: &'static str, + }, + #[error("invalid compression level {0}: value exceeds i32::MAX")] + InvalidCompressionLevel(u32), + #[error("unsafe archive entry path: {0}")] + UnsafeEntryPath(String), + #[error("archive entry path length {length} exceeds limit {limit}: {path}")] + EntryPathTooLong { path: String, length: usize, limit: usize }, + #[error("archive entry count {count} exceeds limit {limit}")] + EntryCountLimitExceeded { count: usize, limit: usize }, + #[error("archive entry '{path}' size {size} exceeds limit {limit}")] + EntrySizeLimitExceeded { path: String, size: u64, limit: u64 }, + #[error("archive total unpacked size {size} exceeds limit {limit}")] + TotalUnpackedSizeLimitExceeded { size: u64, limit: u64 }, + #[error(transparent)] + Io(#[from] io::Error), + #[error(transparent)] + Zip(#[from] zip::result::ZipError), + #[error(transparent)] + Join(#[from] tokio::task::JoinError), +} + +#[derive(Debug, PartialEq, Eq, Clone, Copy)] +pub enum CompressionCodec { + Gzip, + Bzip2, + Xz, + Zlib, + Zstd, +} + +#[derive(Debug, PartialEq, Eq, Clone, Copy)] +pub enum ArchiveKind { + Tar, + Zip, +} + +#[derive(Debug, PartialEq, Eq, Clone, Copy)] +pub enum ArchiveFormat { + Tar, + TarGzip, + TarBzip2, + TarXz, + TarZlib, + TarZstd, + Zip, + Unknown, +} + +#[derive(Debug, PartialEq, Eq, Clone, Copy)] pub enum CompressionFormat { - Gzip, //.gz - Bzip2, //.bz2 - Zip, //.zip - Xz, //.xz - Zlib, //.z - Zstd, //.zst - Tar, //.tar (uncompressed) + Gzip, + Bzip2, + Zip, + Xz, + Zlib, + Zstd, + Tar, Unknown, } @@ -41,32 +106,144 @@ pub enum CompressionLevel { Level(u32), } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ZipEntry { + pub name: String, + pub size: u64, + pub compressed_size: u64, + pub is_dir: bool, + pub compression_method: String, + pub archive_kind: ArchiveKind, + pub format: ArchiveFormat, + pub unix_mode: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct ZipExtractSummary { + pub entry_count: usize, + pub directory_count: usize, + pub file_count: usize, + pub total_unpacked_size: u64, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ArchiveLimits { + pub max_entries: usize, + pub max_entry_size: u64, + pub max_total_unpacked_size: u64, + pub max_path_length: usize, + pub validate_entry_paths: bool, +} + +impl Default for ArchiveLimits { + fn default() -> Self { + Self { + max_entries: 100_000, + max_entry_size: 1_073_741_824, + max_total_unpacked_size: 10_737_418_240, + max_path_length: 1024, + validate_entry_paths: true, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ZipWriteOptions { + pub compression_level: CompressionLevel, + pub create_directory_entries: bool, +} + +impl Default for ZipWriteOptions { + fn default() -> Self { + Self { + compression_level: CompressionLevel::Default, + create_directory_entries: false, + } + } +} + +const SMALL_ZIP_EXTRACT_FAST_PATH_LIMIT: u64 = 8 * 1024; + +#[derive(Clone, Default)] +struct SharedBuffer { + inner: Arc>>, +} + +impl SharedBuffer { + fn into_vec(self) -> Vec { + self.inner.lock().expect("shared in-memory writer lock poisoned").clone() + } +} + +impl AsyncWrite for SharedBuffer { + fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + let mut inner = self + .inner + .lock() + .map_err(|_| io::Error::other("shared in-memory writer lock poisoned"))?; + inner.extend_from_slice(buf); + Poll::Ready(Ok(buf.len())) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } +} + impl CompressionFormat { - /// Identify compression format from file extension pub fn from_extension(ext: &str) -> Self { - match ext.to_lowercase().as_str() { - "gz" | "gzip" | "tgz" => CompressionFormat::Gzip, - "bz2" | "bzip2" | "tbz" | "tbz2" => CompressionFormat::Bzip2, - "zip" => CompressionFormat::Zip, - "xz" | "txz" => CompressionFormat::Xz, - "zlib" => CompressionFormat::Zlib, - "zst" | "zstd" | "tzst" => CompressionFormat::Zstd, - "tar" => CompressionFormat::Tar, - _ => CompressionFormat::Unknown, + Self::from_archive_format(ArchiveFormat::from_extension(ext)) + } + + pub fn from_archive_format(format: ArchiveFormat) -> Self { + match format { + ArchiveFormat::TarGzip => CompressionFormat::Gzip, + ArchiveFormat::TarBzip2 => CompressionFormat::Bzip2, + ArchiveFormat::TarXz => CompressionFormat::Xz, + ArchiveFormat::TarZlib => CompressionFormat::Zlib, + ArchiveFormat::TarZstd => CompressionFormat::Zstd, + ArchiveFormat::Tar => CompressionFormat::Tar, + ArchiveFormat::Zip => CompressionFormat::Zip, + ArchiveFormat::Unknown => CompressionFormat::Unknown, + } + } + + pub fn archive_format_from_path>(path: P) -> ArchiveFormat { + ArchiveFormat::from_path(path) + } + + pub fn archive_kind(&self) -> Option { + match self { + CompressionFormat::Tar => Some(ArchiveKind::Tar), + CompressionFormat::Zip => Some(ArchiveKind::Zip), + CompressionFormat::Gzip + | CompressionFormat::Bzip2 + | CompressionFormat::Xz + | CompressionFormat::Zlib + | CompressionFormat::Zstd + | CompressionFormat::Unknown => None, + } + } + + pub fn compression_codec(&self) -> Option { + match self { + CompressionFormat::Gzip => Some(CompressionCodec::Gzip), + CompressionFormat::Bzip2 => Some(CompressionCodec::Bzip2), + CompressionFormat::Xz => Some(CompressionCodec::Xz), + CompressionFormat::Zlib => Some(CompressionCodec::Zlib), + CompressionFormat::Zstd => Some(CompressionCodec::Zstd), + CompressionFormat::Tar | CompressionFormat::Zip | CompressionFormat::Unknown => None, } } - /// Identify compression format from file path pub fn from_path>(path: P) -> Self { - let path = path.as_ref(); - if let Some(ext) = path.extension().and_then(|s| s.to_str()) { - Self::from_extension(ext) - } else { - CompressionFormat::Unknown - } + Self::from_archive_format(ArchiveFormat::from_path(path)) } - /// Get file extension corresponding to the format pub fn extension(&self) -> &'static str { match self { CompressionFormat::Gzip => "gz", @@ -80,13 +257,11 @@ impl CompressionFormat { } } - /// Check if format is supported pub fn is_supported(&self) -> bool { !matches!(self, CompressionFormat::Unknown) } - /// Create decompressor - pub fn get_decoder(&self, input: R) -> io::Result> + pub fn get_decoder(&self, input: R) -> Result> where R: AsyncRead + Send + Unpin + 'static, { @@ -100,65 +275,58 @@ impl CompressionFormat { CompressionFormat::Zstd => Box::new(ZstdDecoder::new(reader)), CompressionFormat::Tar => Box::new(reader), CompressionFormat::Zip => { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "Zip format requires special handling, use extract_zip function instead", - )); + return Err(ZipError::UnsupportedFormat { + format: *self, + operation: "stream decoding", + }); } CompressionFormat::Unknown => { - return Err(io::Error::new(io::ErrorKind::InvalidInput, "Unsupported file format")); + return Err(ZipError::UnsupportedFormat { + format: *self, + operation: "decoding", + }); } }; Ok(decoder) } - /// Convert CompressionLevel to async_compression::Level - fn convert_level(level: CompressionLevel) -> async_compression::Level { + + fn convert_level(level: CompressionLevel) -> Result { match level { - CompressionLevel::Fastest => async_compression::Level::Fastest, - CompressionLevel::Best => async_compression::Level::Best, - CompressionLevel::Default => async_compression::Level::Default, - CompressionLevel::Level(n) => async_compression::Level::Precise(n as i32), + CompressionLevel::Fastest => Ok(async_compression::Level::Fastest), + CompressionLevel::Best => Ok(async_compression::Level::Best), + CompressionLevel::Default => Ok(async_compression::Level::Default), + CompressionLevel::Level(n) => { + let level = i32::try_from(n).map_err(|_| ZipError::InvalidCompressionLevel(n))?; + Ok(async_compression::Level::Precise(level)) + } } } - /// Create compressor - pub fn get_encoder(&self, output: W, level: CompressionLevel) -> io::Result> + pub fn get_encoder(&self, output: W, level: CompressionLevel) -> Result> where W: AsyncWrite + Send + Unpin + 'static, { let writer = BufWriter::new(output); let encoder: Box = match self { - CompressionFormat::Gzip => { - let level = Self::convert_level(level); - Box::new(GzipEncoder::with_quality(writer, level)) - } - CompressionFormat::Bzip2 => { - let level = Self::convert_level(level); - Box::new(BzEncoder::with_quality(writer, level)) - } - CompressionFormat::Zlib => { - let level = Self::convert_level(level); - Box::new(ZlibEncoder::with_quality(writer, level)) - } - CompressionFormat::Xz => { - let level = Self::convert_level(level); - Box::new(XzEncoder::with_quality(writer, level)) - } - CompressionFormat::Zstd => { - let level = Self::convert_level(level); - Box::new(ZstdEncoder::with_quality(writer, level)) - } + CompressionFormat::Gzip => Box::new(GzipEncoder::with_quality(writer, Self::convert_level(level)?)), + CompressionFormat::Bzip2 => Box::new(BzEncoder::with_quality(writer, Self::convert_level(level)?)), + CompressionFormat::Zlib => Box::new(ZlibEncoder::with_quality(writer, Self::convert_level(level)?)), + CompressionFormat::Xz => Box::new(XzEncoder::with_quality(writer, Self::convert_level(level)?)), + CompressionFormat::Zstd => Box::new(ZstdEncoder::with_quality(writer, Self::convert_level(level)?)), CompressionFormat::Tar => Box::new(writer), CompressionFormat::Zip => { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "Zip format requires special handling, use create_zip function instead", - )); + return Err(ZipError::UnsupportedFormat { + format: *self, + operation: "stream encoding", + }); } CompressionFormat::Unknown => { - return Err(io::Error::new(io::ErrorKind::InvalidInput, "Unsupported file format")); + return Err(ZipError::UnsupportedFormat { + format: *self, + operation: "encoding", + }); } }; @@ -166,55 +334,432 @@ impl CompressionFormat { } } -/// Decompress tar format compressed files -pub async fn decompress(input: R, format: CompressionFormat, mut callback: F) -> io::Result<()> +impl ArchiveFormat { + pub fn from_extension(ext: &str) -> Self { + match ext.to_ascii_lowercase().as_str() { + "gz" | "gzip" | "tgz" => ArchiveFormat::TarGzip, + "bz2" | "bzip2" | "tbz" | "tbz2" => ArchiveFormat::TarBzip2, + "xz" | "txz" => ArchiveFormat::TarXz, + "zlib" | "zz" => ArchiveFormat::TarZlib, + "zst" | "zstd" | "tzst" => ArchiveFormat::TarZstd, + "tar" => ArchiveFormat::Tar, + "zip" => ArchiveFormat::Zip, + _ => ArchiveFormat::Unknown, + } + } + + pub fn from_path>(path: P) -> Self { + let path = path.as_ref(); + let lower_name = path.file_name().and_then(|name| name.to_str()).map(str::to_ascii_lowercase); + + if let Some(name) = lower_name { + if name.ends_with(".tar.gz") || name.ends_with(".tgz") { + return ArchiveFormat::TarGzip; + } + if name.ends_with(".tar.bz2") || name.ends_with(".tbz") || name.ends_with(".tbz2") { + return ArchiveFormat::TarBzip2; + } + if name.ends_with(".tar.xz") || name.ends_with(".txz") { + return ArchiveFormat::TarXz; + } + if name.ends_with(".tar.zst") || name.ends_with(".tzst") { + return ArchiveFormat::TarZstd; + } + if name.ends_with(".tar.zlib") { + return ArchiveFormat::TarZlib; + } + } + + path.extension() + .and_then(|s| s.to_str()) + .map(Self::from_extension) + .unwrap_or(ArchiveFormat::Unknown) + } + + pub fn archive_kind(&self) -> Option { + match self { + ArchiveFormat::Tar + | ArchiveFormat::TarGzip + | ArchiveFormat::TarBzip2 + | ArchiveFormat::TarXz + | ArchiveFormat::TarZlib + | ArchiveFormat::TarZstd => Some(ArchiveKind::Tar), + ArchiveFormat::Zip => Some(ArchiveKind::Zip), + ArchiveFormat::Unknown => None, + } + } + + pub fn compression_codec(&self) -> Option { + match self { + ArchiveFormat::TarGzip => Some(CompressionCodec::Gzip), + ArchiveFormat::TarBzip2 => Some(CompressionCodec::Bzip2), + ArchiveFormat::TarXz => Some(CompressionCodec::Xz), + ArchiveFormat::TarZlib => Some(CompressionCodec::Zlib), + ArchiveFormat::TarZstd => Some(CompressionCodec::Zstd), + ArchiveFormat::Tar | ArchiveFormat::Zip | ArchiveFormat::Unknown => None, + } + } + + pub fn extension(&self) -> &'static str { + match self { + ArchiveFormat::Tar => "tar", + ArchiveFormat::TarGzip => "tar.gz", + ArchiveFormat::TarBzip2 => "tar.bz2", + ArchiveFormat::TarXz => "tar.xz", + ArchiveFormat::TarZlib => "tar.zlib", + ArchiveFormat::TarZstd => "tar.zst", + ArchiveFormat::Zip => "zip", + ArchiveFormat::Unknown => "", + } + } +} + +/// Read entries from a tar-family archive stream. +/// +/// Supported formats are: +/// - `CompressionFormat::Tar` +/// - `CompressionFormat::Gzip` +/// - `CompressionFormat::Bzip2` +/// - `CompressionFormat::Xz` +/// - `CompressionFormat::Zlib` +/// - `CompressionFormat::Zstd` +/// +/// `CompressionFormat::Zip` is intentionally not supported here because ZIP +/// requires central-directory semantics and is handled through file-based +/// helper APIs. +pub async fn read_archive_entries(input: R, format: CompressionFormat, callback: F) -> Result<()> where R: AsyncRead + Send + Unpin + 'static, - F: AsyncFnMut(tokio_tar::Entry>>) -> io::Result<()> + Send + 'static, + F: FnMut(tokio_tar::Entry>>) -> Fut + Send + 'static, + Fut: Future> + Send, +{ + read_archive_entries_with_limits(input, format, ArchiveLimits::default(), callback).await +} + +pub async fn read_archive_entries_with_limits( + input: R, + format: CompressionFormat, + limits: ArchiveLimits, + mut callback: F, +) -> Result<()> +where + R: AsyncRead + Send + Unpin + 'static, + F: FnMut(tokio_tar::Entry>>) -> Fut + Send + 'static, + Fut: Future> + Send, { let decoder = format.get_decoder(input)?; let mut ar = Archive::new(decoder); let mut entries = ar.entries()?; + let mut entry_count = 0_usize; + let mut total_unpacked_size = 0_u64; while let Some(entry) = entries.next().await { let entry = entry?; + entry_count += 1; + validate_archive_entry_count(entry_count, limits)?; + + let entry_path = entry.path()?.to_string_lossy().into_owned(); + validate_archive_entry_name(&entry_path, limits)?; + + let entry_size = entry.header().size()?; + validate_archive_entry_size(&entry_path, entry_size, limits)?; + total_unpacked_size = total_unpacked_size.saturating_add(entry_size); + validate_archive_total_size(total_unpacked_size, limits)?; + callback(entry).await?; } Ok(()) } -/// ZIP file entry information -#[derive(Debug, Clone)] -pub struct ZipEntry { - pub name: String, - pub size: u64, - pub compressed_size: u64, - pub is_dir: bool, - pub compression_method: String, +/// Backward-compatible wrapper for archive entry iteration. +pub async fn decompress(input: R, format: CompressionFormat, callback: F) -> Result<()> +where + R: AsyncRead + Send + Unpin + 'static, + F: FnMut(tokio_tar::Entry>>) -> Fut + Send + 'static, + Fut: Future> + Send, +{ + read_archive_entries(input, format, callback).await } -/// Simplified ZIP file processing (temporarily using standard library zip crate) -pub async fn extract_zip_simple>(zip_path: P, extract_to: P) -> io::Result> { - // Use standard library zip processing, return empty list as placeholder for now - // Actual implementation needs to be improved in future versions - let _zip_path = zip_path.as_ref(); - let _extract_to = extract_to.as_ref(); - - Ok(Vec::new()) +/// Explicit tar-family alias for callers that want a clearer name than +/// `decompress()`. +pub async fn extract_tar_entries(input: R, format: CompressionFormat, callback: F) -> Result<()> +where + R: AsyncRead + Send + Unpin + 'static, + F: FnMut(tokio_tar::Entry>>) -> Fut + Send + 'static, + Fut: Future> + Send, +{ + read_archive_entries(input, format, callback).await +} + +fn normalize_zip_entry_name(name: &str) -> Result { + let path = Path::new(name); + let mut normalized = PathBuf::new(); + + for component in path.components() { + match component { + Component::Normal(part) => normalized.push(part), + Component::CurDir => {} + Component::ParentDir | Component::RootDir | Component::Prefix(_) => { + return Err(ZipError::UnsafeEntryPath(name.to_string())); + } + } + } + + let normalized = normalized.to_string_lossy().replace('\\', "/"); + if normalized.is_empty() { + return Err(ZipError::UnsafeEntryPath(name.to_string())); + } + + Ok(normalized) +} + +fn validate_archive_entry_name(name: &str, limits: ArchiveLimits) -> Result<()> { + if !limits.validate_entry_paths { + return Ok(()); + } + + let normalized = normalize_zip_entry_name(name)?; + let length = normalized.len(); + if length > limits.max_path_length { + return Err(ZipError::EntryPathTooLong { + path: normalized, + length, + limit: limits.max_path_length, + }); + } + + Ok(()) +} + +fn validate_archive_entry_size(path: &str, size: u64, limits: ArchiveLimits) -> Result<()> { + if size > limits.max_entry_size { + return Err(ZipError::EntrySizeLimitExceeded { + path: path.to_string(), + size, + limit: limits.max_entry_size, + }); + } + + Ok(()) +} + +fn validate_archive_entry_count(count: usize, limits: ArchiveLimits) -> Result<()> { + if count > limits.max_entries { + return Err(ZipError::EntryCountLimitExceeded { + count, + limit: limits.max_entries, + }); + } + + Ok(()) +} + +fn validate_archive_total_size(total_size: u64, limits: ArchiveLimits) -> Result<()> { + if total_size > limits.max_total_unpacked_size { + return Err(ZipError::TotalUnpackedSizeLimitExceeded { + size: total_size, + limit: limits.max_total_unpacked_size, + }); + } + + Ok(()) +} + +fn zip_method_for_level(level: CompressionLevel) -> CompressionMethod { + match level { + CompressionLevel::Fastest => CompressionMethod::Stored, + CompressionLevel::Best | CompressionLevel::Default | CompressionLevel::Level(_) => CompressionMethod::Deflated, + } +} + +fn parent_directories_for(path: &str) -> Vec { + let path = Path::new(path); + let mut current = PathBuf::new(); + let mut directories = Vec::new(); + + if let Some(parent) = path.parent() { + for component in parent.components() { + if let Component::Normal(part) = component { + current.push(part); + directories.push(format!("{}/", current.to_string_lossy().replace('\\', "/"))); + } + } + } + + directories +} + +fn ensure_directory(path: &Path, created_directories: &mut HashSet) -> Result<()> { + let path = path.to_path_buf(); + if created_directories.insert(path.clone()) { + std::fs::create_dir_all(&path)?; + } + + Ok(()) +} + +fn write_small_zip_entry(reader: &mut R, output_path: &Path, size: u64) -> Result<()> { + let size = usize::try_from(size).map_err(|_| io::Error::other("small zip entry size overflow"))?; + let mut buffer = [0_u8; SMALL_ZIP_EXTRACT_FAST_PATH_LIMIT as usize]; + reader.read_exact(&mut buffer[..size])?; + std::fs::write(output_path, &buffer[..size])?; + Ok(()) +} + +pub async fn extract_zip_simple, Q: AsRef>(zip_path: P, extract_to: Q) -> Result> { + extract_zip_with_limits(zip_path, extract_to, ArchiveLimits::default()).await +} + +pub async fn extract_zip_to_path_with_limits, Q: AsRef>( + zip_path: P, + extract_to: Q, + limits: ArchiveLimits, +) -> Result { + let zip_path = zip_path.as_ref().to_path_buf(); + let extract_to = extract_to.as_ref().to_path_buf(); + + spawn_blocking(move || extract_zip_impl(zip_path, extract_to, limits, false).map(|(_, summary)| summary)).await? +} + +pub async fn extract_zip_with_limits, Q: AsRef>( + zip_path: P, + extract_to: Q, + limits: ArchiveLimits, +) -> Result> { + let zip_path = zip_path.as_ref().to_path_buf(); + let extract_to = extract_to.as_ref().to_path_buf(); + + spawn_blocking(move || extract_zip_impl(zip_path, extract_to, limits, true).map(|(entries, _)| entries.unwrap_or_default())) + .await? +} + +fn extract_zip_impl( + zip_path: PathBuf, + extract_to: PathBuf, + limits: ArchiveLimits, + collect_entries: bool, +) -> Result<(Option>, ZipExtractSummary)> { + let file = std::fs::File::open(&zip_path)?; + let mut archive = ZipArchive::new(file)?; + std::fs::create_dir_all(&extract_to)?; + let mut created_directories = HashSet::from([extract_to.clone()]); + + let mut entries = collect_entries.then(|| Vec::with_capacity(archive.len())); + let mut summary = ZipExtractSummary::default(); + for index in 0..archive.len() { + validate_archive_entry_count(index + 1, limits)?; + let mut zip_file = archive.by_index(index)?; + let enclosed_name = zip_file + .enclosed_name() + .ok_or_else(|| ZipError::UnsafeEntryPath(zip_file.name().to_string()))?; + let entry_name = enclosed_name.to_string_lossy().replace('\\', "/"); + let is_dir = zip_file.is_dir(); + let size = zip_file.size(); + validate_archive_entry_name(&entry_name, limits)?; + validate_archive_entry_size(&entry_name, size, limits)?; + summary.total_unpacked_size = summary.total_unpacked_size.saturating_add(size); + validate_archive_total_size(summary.total_unpacked_size, limits)?; + let output_path = extract_to.join(&enclosed_name); + + if is_dir { + ensure_directory(&output_path, &mut created_directories)?; + summary.directory_count += 1; + } else { + if let Some(parent) = output_path.parent() { + ensure_directory(parent, &mut created_directories)?; + } + if size <= SMALL_ZIP_EXTRACT_FAST_PATH_LIMIT { + write_small_zip_entry(&mut zip_file, &output_path, size)?; + } else { + let mut output = std::fs::File::create(&output_path)?; + std::io::copy(&mut zip_file, &mut output)?; + } + summary.file_count += 1; + } + summary.entry_count += 1; + + if let Some(ref mut entries) = entries { + entries.push(ZipEntry { + name: entry_name, + size, + compressed_size: zip_file.compressed_size(), + is_dir, + compression_method: format!("{:?}", zip_file.compression()), + archive_kind: ArchiveKind::Zip, + format: ArchiveFormat::Zip, + unix_mode: zip_file.unix_mode(), + }); + } + } + + Ok((entries, summary)) } -/// Simplified ZIP file creation pub async fn create_zip_simple>( - _zip_path: P, - _files: Vec<(String, Vec)>, // (filename, file content) - _compression_level: CompressionLevel, -) -> io::Result<()> { - // Return unimplemented error for now - Err(io::Error::new(io::ErrorKind::Unsupported, "ZIP creation not yet implemented")) + zip_path: P, + files: Vec<(String, Vec)>, + compression_level: CompressionLevel, +) -> Result<()> { + create_zip_with_options( + zip_path, + files, + ZipWriteOptions { + compression_level, + ..ZipWriteOptions::default() + }, + ) + .await +} + +pub async fn create_zip_with_options>( + zip_path: P, + files: Vec<(String, Vec)>, + options: ZipWriteOptions, +) -> Result<()> { + let zip_path = zip_path.as_ref().to_path_buf(); + + spawn_blocking(move || -> Result<()> { + if let Some(parent) = zip_path.parent() { + std::fs::create_dir_all(parent)?; + } + + let file = std::fs::File::create(&zip_path)?; + let mut writer = ZipWriter::new(file); + let file_options = SimpleFileOptions::default().compression_method(zip_method_for_level(options.compression_level)); + let explicit_directories = files + .iter() + .filter(|(name, _)| name.ends_with('/')) + .map(|(name, _)| normalize_zip_entry_name(name)) + .collect::>>()?; + let mut written_directories = HashSet::new(); + + for (name, contents) in files { + let entry_name = normalize_zip_entry_name(&name)?; + if name.ends_with('/') { + if written_directories.insert(entry_name.clone()) { + writer.add_directory(entry_name, file_options)?; + } + } else { + if options.create_directory_entries { + for directory in parent_directories_for(&entry_name) { + if !explicit_directories.contains(&directory) && written_directories.insert(directory.clone()) { + writer.add_directory(directory, file_options)?; + } + } + } + writer.start_file(entry_name, file_options)?; + writer.write_all(&contents)?; + } + } + + writer.finish()?; + Ok(()) + }) + .await? } -/// Compression utility struct pub struct Compressor { format: CompressionFormat, level: CompressionLevel, @@ -233,34 +778,28 @@ impl Compressor { self } - /// Compress data - pub async fn compress(&self, input: &[u8]) -> io::Result> { - let output = Vec::new(); - let cursor = std::io::Cursor::new(output); - let mut encoder = self.format.get_encoder(cursor, self.level)?; + pub async fn compress(&self, input: &[u8]) -> Result> { + let sink = SharedBuffer::default(); + let mut encoder = self.format.get_encoder(sink.clone(), self.level)?; + let mut reader = input; - io::copy(&mut std::io::Cursor::new(input), &mut encoder).await?; + io::copy(&mut reader, &mut encoder).await?; encoder.shutdown().await?; + drop(encoder); - // Get compressed data - // Note: API needs to be redesigned here as we cannot retrieve data from encoder - // Return empty vector as placeholder for now - Ok(Vec::new()) + Ok(sink.into_vec()) } - /// Decompress data - pub async fn decompress(&self, input: Vec) -> io::Result> { + pub async fn decompress(&self, input: Vec) -> Result> { let mut output = Vec::new(); let cursor = std::io::Cursor::new(input); let mut decoder = self.format.get_decoder(cursor)?; - io::copy(&mut decoder, &mut output).await?; - + decoder.read_to_end(&mut output).await?; Ok(output) } } -/// Decompression utility struct pub struct Decompressor { format: CompressionFormat, } @@ -271,12 +810,12 @@ impl Decompressor { } pub fn auto_detect>(path: P) -> Self { - let format = CompressionFormat::from_path(path); - Self { format } + Self { + format: CompressionFormat::from_path(path), + } } - /// Decompress file - pub async fn decompress_file>(&self, input_path: P, output_path: P) -> io::Result<()> { + pub async fn decompress_file>(&self, input_path: P, output_path: P) -> Result<()> { let input_file = File::open(&input_path).await?; let output_file = File::create(&output_path).await?; @@ -293,694 +832,835 @@ impl Decompressor { #[cfg(test)] mod tests { use super::*; - use std::io::Cursor; + use std::mem::size_of; + use tempfile::tempdir; + use tokio::fs; use tokio::io::AsyncReadExt; + use tokio_tar::{Builder, Header}; + use zip::write::FileOptions; + + async fn build_tar_bytes(files: &[(&str, &[u8])]) -> io::Result> { + let sink = SharedBuffer::default(); + let handle = sink.clone(); + let mut builder = Builder::new(sink); + + for (path, content) in files { + let mut header = Header::new_gnu(); + header.set_size(content.len() as u64); + header.set_mode(0o644); + header.set_cksum(); + builder.append_data(&mut header, *path, &content[..]).await?; + } + + builder.finish().await?; + Ok(handle.into_vec()) + } + + async fn build_compressed_tar_bytes(format: CompressionFormat, files: &[(&str, &[u8])]) -> Result> { + let tar_bytes = build_tar_bytes(files).await?; + Compressor::new(format).compress(&tar_bytes).await + } + + async fn build_zip_file_with_entries(path: &Path, files: &[(&str, &[u8])]) -> Result<()> { + let path = path.to_path_buf(); + let files = files + .iter() + .map(|(name, content)| ((*name).to_string(), content.to_vec())) + .collect::>(); + spawn_blocking(move || -> Result<()> { + let file = std::fs::File::create(path)?; + let mut writer = ZipWriter::new(file); + let options: FileOptions<'_, ()> = FileOptions::default().compression_method(CompressionMethod::Stored); + for (name, content) in files { + writer.start_file(name, options)?; + writer.write_all(&content)?; + } + writer.finish()?; + Ok(()) + }) + .await??; + Ok(()) + } + + async fn collect_archive_entries(payload: Vec, format: CompressionFormat) -> Result)>> { + let seen = Arc::new(Mutex::new(Vec::<(String, Vec)>::new())); + let seen_ref = Arc::clone(&seen); + let cursor = std::io::Cursor::new(payload); + + read_archive_entries(cursor, format, move |mut entry| { + let seen_ref = Arc::clone(&seen_ref); + async move { + let path = entry.path()?.to_string_lossy().into_owned(); + let mut content = Vec::new(); + entry.read_to_end(&mut content).await?; + seen_ref.lock().expect("seen collection lock poisoned").push((path, content)); + Ok(()) + } + }) + .await?; + + Ok(seen.lock().expect("seen collection lock poisoned").clone()) + } #[test] fn test_compression_format_from_extension() { - // Test supported compression format recognition assert_eq!(CompressionFormat::from_extension("gz"), CompressionFormat::Gzip); - assert_eq!(CompressionFormat::from_extension("gzip"), CompressionFormat::Gzip); - assert_eq!(CompressionFormat::from_extension("tgz"), CompressionFormat::Gzip); - assert_eq!(CompressionFormat::from_extension("bz2"), CompressionFormat::Bzip2); - assert_eq!(CompressionFormat::from_extension("bzip2"), CompressionFormat::Bzip2); - assert_eq!(CompressionFormat::from_extension("tbz"), CompressionFormat::Bzip2); - assert_eq!(CompressionFormat::from_extension("tbz2"), CompressionFormat::Bzip2); - assert_eq!(CompressionFormat::from_extension("zip"), CompressionFormat::Zip); - assert_eq!(CompressionFormat::from_extension("xz"), CompressionFormat::Xz); - assert_eq!(CompressionFormat::from_extension("txz"), CompressionFormat::Xz); - assert_eq!(CompressionFormat::from_extension("zlib"), CompressionFormat::Zlib); - assert_eq!(CompressionFormat::from_extension("zst"), CompressionFormat::Zstd); - assert_eq!(CompressionFormat::from_extension("zstd"), CompressionFormat::Zstd); + assert_eq!(CompressionFormat::from_extension("ZIP"), CompressionFormat::Zip); assert_eq!(CompressionFormat::from_extension("tzst"), CompressionFormat::Zstd); - assert_eq!(CompressionFormat::from_extension("tar"), CompressionFormat::Tar); - - // Test case insensitivity - assert_eq!(CompressionFormat::from_extension("GZ"), CompressionFormat::Gzip); - assert_eq!(CompressionFormat::from_extension("TGZ"), CompressionFormat::Gzip); - assert_eq!(CompressionFormat::from_extension("ZIP"), CompressionFormat::Zip); - - // Test unknown formats - assert_eq!(CompressionFormat::from_extension("unknown"), CompressionFormat::Unknown); assert_eq!(CompressionFormat::from_extension("txt"), CompressionFormat::Unknown); - assert_eq!(CompressionFormat::from_extension(""), CompressionFormat::Unknown); } #[test] - fn test_compression_format_case_sensitivity() { - // Test case insensitivity (now supports case insensitivity) - assert_eq!(CompressionFormat::from_extension("GZ"), CompressionFormat::Gzip); - assert_eq!(CompressionFormat::from_extension("Gz"), CompressionFormat::Gzip); - assert_eq!(CompressionFormat::from_extension("BZ2"), CompressionFormat::Bzip2); - assert_eq!(CompressionFormat::from_extension("ZIP"), CompressionFormat::Zip); + fn test_archive_format_from_extension() { + assert_eq!(ArchiveFormat::from_extension("gz"), ArchiveFormat::TarGzip); + assert_eq!(ArchiveFormat::from_extension("tbz2"), ArchiveFormat::TarBzip2); + assert_eq!(ArchiveFormat::from_extension("txz"), ArchiveFormat::TarXz); + assert_eq!(ArchiveFormat::from_extension("zip"), ArchiveFormat::Zip); + assert_eq!(ArchiveFormat::from_extension("txt"), ArchiveFormat::Unknown); } #[test] - fn test_compression_format_edge_cases() { - // Test edge cases - assert_eq!(CompressionFormat::from_extension("gz "), CompressionFormat::Unknown); - assert_eq!(CompressionFormat::from_extension(" gz"), CompressionFormat::Unknown); - assert_eq!(CompressionFormat::from_extension("gz.bak"), CompressionFormat::Unknown); - assert_eq!(CompressionFormat::from_extension("tar.gz"), CompressionFormat::Unknown); - } - - #[test] - fn test_compression_format_debug() { - // Test Debug trait implementation - let format = CompressionFormat::Gzip; - let debug_str = format!("{format:?}"); - assert_eq!(debug_str, "Gzip"); - - let unknown_format = CompressionFormat::Unknown; - let unknown_debug_str = format!("{unknown_format:?}"); - assert_eq!(unknown_debug_str, "Unknown"); - } - - #[test] - fn test_compression_format_equality() { - // Test PartialEq trait implementation - assert_eq!(CompressionFormat::Gzip, CompressionFormat::Gzip); - assert_eq!(CompressionFormat::Unknown, CompressionFormat::Unknown); - assert_ne!(CompressionFormat::Gzip, CompressionFormat::Bzip2); - assert_ne!(CompressionFormat::Zip, CompressionFormat::Unknown); - } - - #[tokio::test] - async fn test_get_decoder_supported_formats() { - // Test that supported formats can create decoders - let test_data = b"test data"; - let cursor = Cursor::new(test_data); - - let gzip_format = CompressionFormat::Gzip; - let decoder_result = gzip_format.get_decoder(cursor); - assert!(decoder_result.is_ok(), "Gzip decoder should be created successfully"); - } - - #[tokio::test] - async fn test_get_decoder_unsupported_formats() { - // Test that unsupported formats return errors - let test_data = b"test data"; - let cursor = Cursor::new(test_data); - - let unknown_format = CompressionFormat::Unknown; - let decoder_result = unknown_format.get_decoder(cursor); - assert!(decoder_result.is_err(), "Unknown format should return error"); - - if let Err(e) = decoder_result { - assert_eq!(e.kind(), io::ErrorKind::InvalidInput); - assert_eq!(e.to_string(), "Unsupported file format"); - } - } - - #[tokio::test] - async fn test_get_decoder_zip_format() { - // Test Zip format (currently not supported) - let test_data = b"test data"; - let cursor = Cursor::new(test_data); - - let zip_format = CompressionFormat::Zip; - let decoder_result = zip_format.get_decoder(cursor); - assert!(decoder_result.is_err(), "Zip format should return error (not implemented)"); - } - - #[tokio::test] - async fn test_get_decoder_all_supported_formats() { - // Test that all supported formats can create decoders successfully - let sample_content = b"Hello, compression world!"; - - let supported_formats = vec![ - CompressionFormat::Gzip, - CompressionFormat::Bzip2, - CompressionFormat::Zlib, - CompressionFormat::Xz, - CompressionFormat::Zstd, - CompressionFormat::Tar, - ]; - - for format in supported_formats { - let cursor = Cursor::new(sample_content); - let decoder_result = format.get_decoder(cursor); - assert!(decoder_result.is_ok(), "Format {format:?} should create decoder successfully"); - } - } - - #[tokio::test] - async fn test_decoder_type_consistency() { - // Test decoder return type consistency - let sample_content = b"Hello, compression world!"; - let cursor = Cursor::new(sample_content); - - let gzip_format = CompressionFormat::Gzip; - let mut decoder = gzip_format.get_decoder(cursor).unwrap(); - - // Verify that the returned decoder implements the correct trait object - let mut output_buffer = Vec::new(); - // This only verifies the type, we don't expect successful reading (since data is not actual gzip format) - let _read_result = decoder.read_to_end(&mut output_buffer).await; - } - - #[test] - fn test_compression_format_exhaustive_matching() { - // Test that all enum variants have corresponding handling - let all_formats = vec![ - CompressionFormat::Gzip, - CompressionFormat::Bzip2, - CompressionFormat::Zip, - CompressionFormat::Xz, - CompressionFormat::Zlib, - CompressionFormat::Zstd, - CompressionFormat::Unknown, - ]; - - for format in all_formats { - // Verify each format has corresponding Debug implementation - let _debug_str = format!("{format:?}"); - - // Verify each format has corresponding PartialEq implementation - assert_eq!(format, format); - } - } - - #[test] - fn test_extension_mapping_completeness() { - // Test completeness of extension mapping - let extension_mappings = vec![ - ("gz", CompressionFormat::Gzip), - ("gzip", CompressionFormat::Gzip), - ("bz2", CompressionFormat::Bzip2), - ("bzip2", CompressionFormat::Bzip2), - ("zip", CompressionFormat::Zip), - ("xz", CompressionFormat::Xz), - ("zlib", CompressionFormat::Zlib), - ("zst", CompressionFormat::Zstd), - ("zstd", CompressionFormat::Zstd), - ("tar", CompressionFormat::Tar), - ]; - - for (ext, expected_format) in extension_mappings { - assert_eq!( - CompressionFormat::from_extension(ext), - expected_format, - "Extension '{ext}' should map to {expected_format:?}" - ); - } - } - - #[test] - fn test_format_string_representations() { - // Test string representation of formats - let format_strings = vec![ - (CompressionFormat::Gzip, "Gzip"), - (CompressionFormat::Bzip2, "Bzip2"), - (CompressionFormat::Zip, "Zip"), - (CompressionFormat::Xz, "Xz"), - (CompressionFormat::Zlib, "Zlib"), - (CompressionFormat::Zstd, "Zstd"), - (CompressionFormat::Unknown, "Unknown"), - ]; - - for (format, expected_str) in format_strings { - assert_eq!( - format!("{format:?}"), - expected_str, - "Format {:?} should have string representation '{}'", - format, - expected_str - ); - } - } - - #[tokio::test] - async fn test_decoder_error_handling() { - // Test decoder error handling with edge cases - let empty_content = b""; - let cursor = Cursor::new(empty_content); - - let gzip_format = CompressionFormat::Gzip; - let decoder_result = gzip_format.get_decoder(cursor); - - // Decoder creation should succeed even with empty content - assert!(decoder_result.is_ok(), "Decoder creation should succeed even with empty content"); - } - - #[test] - fn test_compression_format_memory_efficiency() { - // Verify enum size is reasonable - let size = size_of::(); - assert!(size <= 8, "CompressionFormat should be memory efficient, got {size} bytes"); - - // Verify Option size - let option_size = size_of::>(); - assert!( - option_size <= 16, - "Option should be efficient, got {option_size} bytes" - ); - } - - #[test] - fn test_extension_validation() { - // Test edge cases of extension validation - let test_cases = vec![ - // Normal cases - ("gz", true), - ("bz2", true), - ("xz", true), - // Edge cases - ("", false), - ("g", false), - ("gzz", false), - ("gz2", false), - // Special characters - ("gz.", false), - (".gz", false), - ("gz-", false), - ("gz_", false), - ]; - - for (ext, should_be_known) in test_cases { - let format = CompressionFormat::from_extension(ext); - let is_known = format != CompressionFormat::Unknown; - assert_eq!( - is_known, should_be_known, - "Extension '{ext}' recognition mismatch: expected {should_be_known}, got {is_known}" - ); - } - } - - #[tokio::test] - async fn test_decoder_trait_bounds() { - // Test decoder trait bounds compliance - let sample_content = b"Hello, compression world!"; - let cursor = Cursor::new(sample_content); - - let gzip_format = CompressionFormat::Gzip; - let decoder = gzip_format.get_decoder(cursor).unwrap(); - - // Verify that the returned decoder satisfies required trait bounds - fn check_trait_bounds(_: &T) {} - check_trait_bounds(&*decoder); - } - - #[test] - fn test_format_consistency_with_extensions() { - // Test format consistency with extensions - let consistency_tests = vec![ - (CompressionFormat::Gzip, "gz"), - (CompressionFormat::Bzip2, "bz2"), - (CompressionFormat::Zip, "zip"), - (CompressionFormat::Xz, "xz"), - (CompressionFormat::Zlib, "zlib"), - (CompressionFormat::Zstd, "zst"), - ]; - - for (format, ext) in consistency_tests { - let parsed_format = CompressionFormat::from_extension(ext); - assert_eq!(parsed_format, format, "Extension '{ext}' should consistently map to {format:?}"); - } - } - - #[tokio::test] - async fn test_decompress_with_invalid_format() { - // Test decompression with invalid format - use std::sync::Arc; - use std::sync::atomic::{AtomicUsize, Ordering}; - - let sample_content = b"Hello, compression world!"; - let cursor = Cursor::new(sample_content); - - let processed_entries_count = Arc::new(AtomicUsize::new(0)); - let processed_entries_count_clone = processed_entries_count.clone(); - - let decompress_result = decompress(cursor, CompressionFormat::Unknown, move |_archive_entry| { - processed_entries_count_clone.fetch_add(1, Ordering::SeqCst); - async move { Ok(()) } - }) - .await; - - assert!(decompress_result.is_err(), "Decompress with Unknown format should fail"); - assert_eq!( - processed_entries_count.load(Ordering::SeqCst), - 0, - "No entries should be processed with invalid format" - ); - } - - #[tokio::test] - async fn test_decompress_with_zip_format() { - // Test decompression with Zip format (currently not supported) - use std::sync::Arc; - use std::sync::atomic::{AtomicUsize, Ordering}; - - let sample_content = b"Hello, compression world!"; - let cursor = Cursor::new(sample_content); - - let processed_entries_count = Arc::new(AtomicUsize::new(0)); - let processed_entries_count_clone = processed_entries_count.clone(); - - let decompress_result = decompress(cursor, CompressionFormat::Zip, move |_archive_entry| { - processed_entries_count_clone.fetch_add(1, Ordering::SeqCst); - async move { Ok(()) } - }) - .await; - - assert!(decompress_result.is_err(), "Decompress with Zip format should fail (not implemented)"); - assert_eq!( - processed_entries_count.load(Ordering::SeqCst), - 0, - "No entries should be processed with unsupported format" - ); - } - - #[tokio::test] - async fn test_decompress_error_propagation() { - // Test error propagation during decompression process - use std::sync::Arc; - use std::sync::atomic::{AtomicUsize, Ordering}; - - let sample_content = b"Hello, compression world!"; - let cursor = Cursor::new(sample_content); - - let callback_invocation_count = Arc::new(AtomicUsize::new(0)); - let callback_invocation_count_clone = callback_invocation_count.clone(); - - let decompress_result = decompress(cursor, CompressionFormat::Gzip, move |_archive_entry| { - let invocation_number = callback_invocation_count_clone.fetch_add(1, Ordering::SeqCst); - async move { - if invocation_number == 0 { - // First invocation returns an error - Err(io::Error::other("Simulated callback error")) - } else { - Ok(()) - } - } - }) - .await; - - // Since input data is not valid gzip format, it may fail during parsing phase - // This mainly tests the error handling mechanism - assert!(decompress_result.is_err(), "Should propagate callback errors"); - } - - #[tokio::test] - async fn test_decompress_callback_execution() { - // Test callback function execution during decompression - use std::sync::Arc; - use std::sync::atomic::{AtomicBool, Ordering}; - - let sample_content = b"Hello, compression world!"; - let cursor = Cursor::new(sample_content); - - let callback_was_invoked = Arc::new(AtomicBool::new(false)); - let callback_was_invoked_clone = callback_was_invoked.clone(); - - let _decompress_result = decompress(cursor, CompressionFormat::Gzip, move |_archive_entry| { - callback_was_invoked_clone.store(true, Ordering::SeqCst); - async move { Ok(()) } - }) - .await; - - // Note: Since test data is not valid gzip format, callback may not be invoked - // This test mainly verifies function signature and basic flow - } - - #[test] - fn test_compression_format_clone_and_copy() { - // Test if CompressionFormat can be copied - let format = CompressionFormat::Gzip; - let format_copy = format; - - // Verify copied values are equal - assert_eq!(format, format_copy); - - // Verify original value is still usable - assert_eq!(format, CompressionFormat::Gzip); - } - - #[test] - fn test_compression_format_match_exhaustiveness() { - // Test match statement completeness - fn handle_format(format: CompressionFormat) -> &'static str { - match format { - CompressionFormat::Gzip => "gzip", - CompressionFormat::Bzip2 => "bzip2", - CompressionFormat::Zip => "zip", - CompressionFormat::Xz => "xz", - CompressionFormat::Zlib => "zlib", - CompressionFormat::Zstd => "zstd", - CompressionFormat::Tar => "tar", - CompressionFormat::Unknown => "unknown", - } - } - - // Test all variants have corresponding handlers - assert_eq!(handle_format(CompressionFormat::Gzip), "gzip"); - assert_eq!(handle_format(CompressionFormat::Bzip2), "bzip2"); - assert_eq!(handle_format(CompressionFormat::Zip), "zip"); - assert_eq!(handle_format(CompressionFormat::Xz), "xz"); - assert_eq!(handle_format(CompressionFormat::Zlib), "zlib"); - assert_eq!(handle_format(CompressionFormat::Zstd), "zstd"); - assert_eq!(handle_format(CompressionFormat::Unknown), "unknown"); - } - - #[test] - fn test_extension_parsing_performance() { - // Test extension parsing performance (simple performance test) - let extensions = vec!["gz", "bz2", "zip", "xz", "zlib", "zst", "unknown"]; - - // Multiple calls to test performance consistency - for _ in 0..1000 { - for ext in &extensions { - let _format = CompressionFormat::from_extension(ext); - } - } - - // Extension parsing performance test completed - } - - #[test] - fn test_format_default_behavior() { - // Test format default behavior - let unknown_extensions = vec!["", "txt", "doc", "pdf", "unknown_ext"]; - - for ext in unknown_extensions { - let format = CompressionFormat::from_extension(ext); - assert_eq!(format, CompressionFormat::Unknown, "Extension '{ext}' should default to Unknown"); - } - } - - #[test] - fn test_compression_level() { - // Test compression level - let default_level = CompressionLevel::default(); - assert_eq!(default_level, CompressionLevel::Default); - - let fastest = CompressionLevel::Fastest; - let best = CompressionLevel::Best; - let custom = CompressionLevel::Level(5); - - assert_ne!(fastest, best); - assert_ne!(default_level, custom); - } - - #[test] - fn test_format_extension() { - // Test format extension retrieval - assert_eq!(CompressionFormat::Gzip.extension(), "gz"); - assert_eq!(CompressionFormat::Bzip2.extension(), "bz2"); - assert_eq!(CompressionFormat::Zip.extension(), "zip"); - assert_eq!(CompressionFormat::Xz.extension(), "xz"); - assert_eq!(CompressionFormat::Zlib.extension(), "zlib"); - assert_eq!(CompressionFormat::Zstd.extension(), "zst"); - assert_eq!(CompressionFormat::Tar.extension(), "tar"); - assert_eq!(CompressionFormat::Unknown.extension(), ""); - } - - #[test] - fn test_format_is_supported() { - // Test format support check - assert!(CompressionFormat::Gzip.is_supported()); - assert!(CompressionFormat::Bzip2.is_supported()); - assert!(CompressionFormat::Zip.is_supported()); - assert!(CompressionFormat::Xz.is_supported()); - assert!(CompressionFormat::Zlib.is_supported()); - assert!(CompressionFormat::Zstd.is_supported()); - assert!(CompressionFormat::Tar.is_supported()); - assert!(!CompressionFormat::Unknown.is_supported()); - } - - #[test] - fn test_format_from_path() { - // Test format recognition from path - use std::path::Path; - - assert_eq!(CompressionFormat::from_path("file.gz"), CompressionFormat::Gzip); + fn test_compression_format_from_path_handles_compound_suffixes() { + assert_eq!(CompressionFormat::from_path("archive.tar.gz"), CompressionFormat::Gzip); + assert_eq!(CompressionFormat::from_path("archive.tgz"), CompressionFormat::Gzip); + assert_eq!(CompressionFormat::from_path("archive.tar.bz2"), CompressionFormat::Bzip2); assert_eq!(CompressionFormat::from_path("archive.zip"), CompressionFormat::Zip); - assert_eq!(CompressionFormat::from_path("/path/to/file.tar.gz"), CompressionFormat::Gzip); - assert_eq!(CompressionFormat::from_path("no_extension"), CompressionFormat::Unknown); - - let path = Path::new("test.bz2"); - assert_eq!(CompressionFormat::from_path(path), CompressionFormat::Bzip2); - } - - #[tokio::test] - async fn test_get_encoder_supported_formats() { - // Test supported formats can create encoders - use std::io::Cursor; - - let output = Vec::new(); - let cursor = Cursor::new(output); - - let gzip_format = CompressionFormat::Gzip; - let encoder_result = gzip_format.get_encoder(cursor, CompressionLevel::Default); - assert!(encoder_result.is_ok(), "Gzip encoder should be created successfully"); - } - - #[tokio::test] - async fn test_get_encoder_unsupported_formats() { - // Test unsupported formats return errors - use std::io::Cursor; - - let output1 = Vec::new(); - let cursor1 = Cursor::new(output1); - - let unknown_format = CompressionFormat::Unknown; - let encoder_result = unknown_format.get_encoder(cursor1, CompressionLevel::Default); - assert!(encoder_result.is_err(), "Unknown format should return error"); - - let output2 = Vec::new(); - let cursor2 = Cursor::new(output2); - let zip_format = CompressionFormat::Zip; - let zip_encoder_result = zip_format.get_encoder(cursor2, CompressionLevel::Default); - assert!(zip_encoder_result.is_err(), "Zip format should return error (requires special handling)"); - } - - #[tokio::test] - async fn test_compressor_basic_functionality() { - // Test basic compressor functionality (Note: current implementation returns empty vector as placeholder) - let compressor = Compressor::new(CompressionFormat::Gzip); - let sample_text = b"Hello, World! This is a sample string for compression testing."; - - let compression_result = compressor.compress(sample_text).await; - assert!(compression_result.is_ok(), "Compression should succeed"); - - // Note: Current implementation returns empty vector as placeholder - let compressed_output = compression_result.unwrap(); - // assert!(!compressed_output.is_empty(), "Compressed data should not be empty"); - // assert_ne!(compressed_output.as_slice(), sample_text, "Compressed data should be different from original"); - - // Temporarily verify that function can be called normally - assert!(compressed_output.is_empty(), "Current implementation returns empty vector as placeholder"); - } - - #[tokio::test] - async fn test_compressor_with_level() { - // Test compressor with custom compression level - let compressor = Compressor::new(CompressionFormat::Gzip).with_level(CompressionLevel::Best); - - let sample_text = b"Sample text for compression level testing"; - let compression_result = compressor.compress(sample_text).await; - assert!(compression_result.is_ok(), "Compression with custom level should succeed"); + assert_eq!(CompressionFormat::from_path("archive"), CompressionFormat::Unknown); } #[test] - fn test_decompressor_creation() { - // Test decompressor creation - let decompressor = Decompressor::new(CompressionFormat::Gzip); - assert_eq!(decompressor.format, CompressionFormat::Gzip); - - let auto_decompressor = Decompressor::auto_detect("test.gz"); - assert_eq!(auto_decompressor.format, CompressionFormat::Gzip); + fn test_archive_format_from_path_handles_compound_suffixes() { + assert_eq!(ArchiveFormat::from_path("archive.tar.gz"), ArchiveFormat::TarGzip); + assert_eq!(ArchiveFormat::from_path("archive.tar.bz2"), ArchiveFormat::TarBzip2); + assert_eq!(ArchiveFormat::from_path("archive.tar.xz"), ArchiveFormat::TarXz); + assert_eq!(ArchiveFormat::from_path("archive.tar.zst"), ArchiveFormat::TarZstd); + assert_eq!(ArchiveFormat::from_path("archive.zip"), ArchiveFormat::Zip); + assert_eq!(ArchiveFormat::from_path("archive"), ArchiveFormat::Unknown); } #[test] - fn test_zip_entry_creation() { - // Test ZIP entry info creation - let entry = ZipEntry { - name: "test.txt".to_string(), - size: 1024, - compressed_size: 512, - is_dir: false, - compression_method: "Deflate".to_string(), + fn test_archive_format_and_legacy_compression_format_are_compatible() { + assert_eq!(CompressionFormat::from_archive_format(ArchiveFormat::TarGzip), CompressionFormat::Gzip); + assert_eq!(CompressionFormat::from_archive_format(ArchiveFormat::Tar), CompressionFormat::Tar); + assert_eq!(CompressionFormat::from_archive_format(ArchiveFormat::Zip), CompressionFormat::Zip); + } + + #[test] + fn test_archive_format_exposes_archive_kind_and_codec() { + assert_eq!(ArchiveFormat::TarGzip.archive_kind(), Some(ArchiveKind::Tar)); + assert_eq!(ArchiveFormat::TarGzip.compression_codec(), Some(CompressionCodec::Gzip)); + assert_eq!(ArchiveFormat::Tar.archive_kind(), Some(ArchiveKind::Tar)); + assert_eq!(ArchiveFormat::Tar.compression_codec(), None); + assert_eq!(ArchiveFormat::Zip.archive_kind(), Some(ArchiveKind::Zip)); + assert_eq!(ArchiveFormat::Zip.compression_codec(), None); + } + + #[test] + fn test_legacy_compression_format_exposes_kind_and_codec() { + assert_eq!(CompressionFormat::Gzip.archive_kind(), None); + assert_eq!(CompressionFormat::Gzip.compression_codec(), Some(CompressionCodec::Gzip)); + assert_eq!(CompressionFormat::Tar.archive_kind(), Some(ArchiveKind::Tar)); + assert_eq!(CompressionFormat::Tar.compression_codec(), None); + assert_eq!(CompressionFormat::Zip.archive_kind(), Some(ArchiveKind::Zip)); + } + + #[test] + fn test_compression_format_size_is_small() { + assert!(size_of::() <= 8); + assert!(size_of::>() <= 16); + } + + #[test] + fn test_convert_level_rejects_overflow() { + let err = match CompressionFormat::Gzip.get_encoder(SharedBuffer::default(), CompressionLevel::Level(u32::MAX)) { + Ok(_) => panic!("overflow level should return an error"), + Err(err) => err, }; - - assert_eq!(entry.name, "test.txt"); - assert_eq!(entry.size, 1024); - assert_eq!(entry.compressed_size, 512); - assert!(!entry.is_dir); - assert_eq!(entry.compression_method, "Deflate"); + assert!(matches!(err, ZipError::InvalidCompressionLevel(u32::MAX))); } #[test] - fn test_compression_level_variants() { - // Test all compression level variants - let levels = vec![ - CompressionLevel::Fastest, - CompressionLevel::Best, - CompressionLevel::Default, - CompressionLevel::Level(1), - CompressionLevel::Level(9), - ]; - - for level in levels { - // Verify each level has corresponding Debug implementation - let _debug_str = format!("{level:?}"); - } + fn test_validate_archive_entry_name_rejects_absolute_path() { + let err = validate_archive_entry_name("/absolute.txt", ArchiveLimits::default()) + .expect_err("absolute path should fail validation"); + assert!(matches!(err, ZipError::UnsafeEntryPath(path) if path == "/absolute.txt")); } - #[test] - fn test_format_comprehensive_coverage() { - // Test comprehensive format coverage - let all_formats = vec![ + #[tokio::test] + async fn test_compressor_round_trip_gzip() { + let input = b"hello rustfs zip ".repeat(64); + let compressor = Compressor::new(CompressionFormat::Gzip); + + let compressed = compressor.compress(&input).await.expect("gzip compress should succeed"); + assert!(!compressed.is_empty()); + assert_ne!(compressed, input); + + let decompressed = compressor + .decompress(compressed) + .await + .expect("gzip decompress should succeed"); + assert_eq!(decompressed, input); + } + + #[tokio::test] + async fn test_compressor_round_trip_zstd() { + let input = b"zstd payload ".repeat(128); + let compressor = Compressor::new(CompressionFormat::Zstd).with_level(CompressionLevel::Best); + + let compressed = compressor.compress(&input).await.expect("zstd compress should succeed"); + let decompressed = compressor + .decompress(compressed) + .await + .expect("zstd decompress should succeed"); + assert_eq!(decompressed, input); + } + + #[tokio::test] + async fn test_zip_stream_encoder_is_rejected() { + let err = CompressionFormat::Zip + .get_encoder(SharedBuffer::default(), CompressionLevel::Default) + .err() + .expect("zip stream encoder should be rejected"); + assert!(matches!( + err, + ZipError::UnsupportedFormat { + format: CompressionFormat::Zip, + operation: "stream encoding", + } + )); + } + + #[tokio::test] + async fn test_read_archive_entries_iterates_tar_gzip_entries() { + let gzip_bytes = + build_compressed_tar_bytes(CompressionFormat::Gzip, &[("nested/hello.txt", b"hello"), ("world.txt", b"world")]) + .await + .expect("tar.gz build should succeed"); + + let seen = collect_archive_entries(gzip_bytes, CompressionFormat::Gzip) + .await + .expect("tar.gz archive iteration should succeed"); + assert_eq!(seen.len(), 2); + assert_eq!(seen[0].0, "nested/hello.txt"); + assert_eq!(seen[0].1, b"hello"); + assert_eq!(seen[1].0, "world.txt"); + assert_eq!(seen[1].1, b"world"); + } + + #[tokio::test] + async fn test_read_archive_entries_iterates_tar_bzip2_entries() { + let payload = + build_compressed_tar_bytes(CompressionFormat::Bzip2, &[("nested/hello.txt", b"hello"), ("world.txt", b"world")]) + .await + .expect("tar.bz2 build should succeed"); + + let seen = collect_archive_entries(payload, CompressionFormat::Bzip2) + .await + .expect("tar.bz2 archive iteration should succeed"); + assert_eq!(seen.len(), 2); + assert_eq!(seen[0].0, "nested/hello.txt"); + assert_eq!(seen[1].0, "world.txt"); + } + + #[tokio::test] + async fn test_read_archive_entries_iterates_tar_xz_entries() { + let payload = + build_compressed_tar_bytes(CompressionFormat::Xz, &[("nested/hello.txt", b"hello"), ("world.txt", b"world")]) + .await + .expect("tar.xz build should succeed"); + + let seen = collect_archive_entries(payload, CompressionFormat::Xz) + .await + .expect("tar.xz archive iteration should succeed"); + assert_eq!(seen.len(), 2); + assert_eq!(seen[0].0, "nested/hello.txt"); + assert_eq!(seen[1].0, "world.txt"); + } + + #[tokio::test] + async fn test_read_archive_entries_iterates_tar_zstd_entries() { + let payload = + build_compressed_tar_bytes(CompressionFormat::Zstd, &[("nested/hello.txt", b"hello"), ("world.txt", b"world")]) + .await + .expect("tar.zst build should succeed"); + + let seen = collect_archive_entries(payload, CompressionFormat::Zstd) + .await + .expect("tar.zst archive iteration should succeed"); + assert_eq!(seen.len(), 2); + assert_eq!(seen[0].0, "nested/hello.txt"); + assert_eq!(seen[1].0, "world.txt"); + } + + #[tokio::test] + async fn test_extract_tar_entries_alias_matches_stream_behavior() { + let payload = build_compressed_tar_bytes(CompressionFormat::Gzip, &[("hello.txt", b"hello")]) + .await + .expect("tar.gz build should succeed"); + let seen = Arc::new(Mutex::new(Vec::::new())); + let seen_ref = Arc::clone(&seen); + + extract_tar_entries(std::io::Cursor::new(payload), CompressionFormat::Gzip, move |entry| { + let seen_ref = Arc::clone(&seen_ref); + async move { + seen_ref + .lock() + .expect("seen collection lock poisoned") + .push(entry.path()?.to_string_lossy().into_owned()); + Ok(()) + } + }) + .await + .expect("extract_tar_entries alias should succeed"); + + assert_eq!(seen.lock().expect("seen collection lock poisoned").as_slice(), ["hello.txt"]); + } + + #[tokio::test] + async fn test_read_archive_entries_rejects_zip_streams() { + let err = read_archive_entries(std::io::Cursor::new(Vec::::new()), CompressionFormat::Zip, |_entry| async { Ok(()) }) + .await + .expect_err("zip stream should be rejected"); + + assert!(matches!( + err, + ZipError::UnsupportedFormat { + format: CompressionFormat::Zip, + operation: "stream decoding", + } + )); + } + + #[tokio::test] + async fn test_read_archive_entries_rejects_corrupt_tar_gzip_stream() { + let err = read_archive_entries( + std::io::Cursor::new(b"not-a-valid-gzip-stream".to_vec()), CompressionFormat::Gzip, - CompressionFormat::Bzip2, - CompressionFormat::Zip, - CompressionFormat::Xz, - CompressionFormat::Zlib, - CompressionFormat::Zstd, - CompressionFormat::Tar, - CompressionFormat::Unknown, - ]; + |_entry| async { Ok(()) }, + ) + .await + .expect_err("corrupt tar.gz stream should fail"); - for format in all_formats { - // Verify each format has an extension - let _ext = format.extension(); + assert!(matches!(err, ZipError::Io(_))); + } - // Verify support status check - let _supported = format.is_supported(); + #[tokio::test] + async fn test_read_archive_entries_rejects_truncated_tar_gzip_stream() { + let payload = build_compressed_tar_bytes(CompressionFormat::Gzip, &[("hello.txt", b"hello world")]) + .await + .expect("tar.gz build should succeed"); + let truncated = payload[..payload.len() / 2].to_vec(); - // Verify Debug implementation - let _debug = format!("{format:?}"); + let err = read_archive_entries(std::io::Cursor::new(truncated), CompressionFormat::Gzip, |_entry| async { Ok(()) }) + .await + .expect_err("truncated tar.gz stream should fail"); + + assert!(matches!(err, ZipError::Io(_))); + } + + #[tokio::test] + async fn test_read_archive_entries_rejects_too_many_entries() { + let payload = build_compressed_tar_bytes(CompressionFormat::Gzip, &[("one.txt", b"1"), ("two.txt", b"2")]) + .await + .expect("tar.gz build should succeed"); + + let err = read_archive_entries_with_limits( + std::io::Cursor::new(payload), + CompressionFormat::Gzip, + ArchiveLimits { + max_entries: 1, + ..ArchiveLimits::default() + }, + |_entry| async { Ok(()) }, + ) + .await + .expect_err("entry count limit should fail"); + + assert!(matches!(err, ZipError::EntryCountLimitExceeded { count: 2, limit: 1 })); + } + + #[tokio::test] + async fn test_read_archive_entries_rejects_oversized_entry() { + let payload = build_compressed_tar_bytes(CompressionFormat::Gzip, &[("big.txt", b"hello world")]) + .await + .expect("tar.gz build should succeed"); + + let err = read_archive_entries_with_limits( + std::io::Cursor::new(payload), + CompressionFormat::Gzip, + ArchiveLimits { + max_entry_size: 4, + ..ArchiveLimits::default() + }, + |_entry| async { Ok(()) }, + ) + .await + .expect_err("entry size limit should fail"); + + assert!(matches!( + err, + ZipError::EntrySizeLimitExceeded { + path, + size: 11, + limit: 4, + } if path == "big.txt" + )); + } + + #[tokio::test] + async fn test_read_archive_entries_rejects_total_unpacked_size_limit() { + let payload = build_compressed_tar_bytes(CompressionFormat::Gzip, &[("one.txt", b"12345"), ("two.txt", b"67890")]) + .await + .expect("tar.gz build should succeed"); + + let err = read_archive_entries_with_limits( + std::io::Cursor::new(payload), + CompressionFormat::Gzip, + ArchiveLimits { + max_total_unpacked_size: 9, + ..ArchiveLimits::default() + }, + |_entry| async { Ok(()) }, + ) + .await + .expect_err("total unpacked size limit should fail"); + + assert!(matches!(err, ZipError::TotalUnpackedSizeLimitExceeded { size: 10, limit: 9 })); + } + + #[tokio::test] + async fn test_read_archive_entries_rejects_entry_path_length_limit() { + let payload = build_compressed_tar_bytes(CompressionFormat::Gzip, &[("nested/hello.txt", b"hello")]) + .await + .expect("tar.gz build should succeed"); + + let err = read_archive_entries_with_limits( + std::io::Cursor::new(payload), + CompressionFormat::Gzip, + ArchiveLimits { + max_path_length: 5, + ..ArchiveLimits::default() + }, + |_entry| async { Ok(()) }, + ) + .await + .expect_err("path length limit should fail"); + + assert!(matches!( + err, + ZipError::EntryPathTooLong { path, limit: 5, .. } if path == "nested/hello.txt" + )); + } + + #[tokio::test] + async fn test_create_and_extract_zip_round_trip() { + let temp = tempdir().expect("tempdir should be created"); + let zip_path = temp.path().join("archive.zip"); + let extract_path = temp.path().join("extract"); + + create_zip_simple( + &zip_path, + vec![ + ("nested/hello.txt".to_string(), b"hello".to_vec()), + ("world.txt".to_string(), b"world".to_vec()), + ], + CompressionLevel::Default, + ) + .await + .expect("zip creation should succeed"); + + let entries = extract_zip_simple(&zip_path, &extract_path) + .await + .expect("zip extraction should succeed"); + assert_eq!(entries.len(), 2); + assert_eq!( + fs::read(extract_path.join("nested/hello.txt")) + .await + .expect("nested zip entry should be extracted"), + b"hello" + ); + assert_eq!( + fs::read(extract_path.join("world.txt")) + .await + .expect("root zip entry should be extracted"), + b"world" + ); + assert!(entries.iter().all(|entry| entry.archive_kind == ArchiveKind::Zip)); + assert!(entries.iter().all(|entry| entry.format == ArchiveFormat::Zip)); + } + + #[tokio::test] + async fn test_create_zip_with_directory_entries_and_extract_directory_scenarios() { + let temp = tempdir().expect("tempdir should be created"); + let explicit_zip_path = temp.path().join("explicit-directories.zip"); + let explicit_extract_path = temp.path().join("explicit-extract"); + let auto_zip_path = temp.path().join("auto-directories.zip"); + let auto_extract_path = temp.path().join("auto-extract"); + + create_zip_with_options( + &explicit_zip_path, + vec![ + ("nested/".to_string(), Vec::new()), + ("nested/deeper/".to_string(), Vec::new()), + ], + ZipWriteOptions { + compression_level: CompressionLevel::Default, + create_directory_entries: false, + }, + ) + .await + .expect("zip creation with explicit directory entries should succeed"); + + let explicit_entries = extract_zip_with_limits(&explicit_zip_path, &explicit_extract_path, ArchiveLimits::default()) + .await + .expect("zip extraction with explicit directory entries should succeed"); + + assert!( + explicit_entries + .iter() + .any(|entry| entry.name.trim_end_matches('/') == "nested" && entry.is_dir) + ); + assert!( + explicit_entries + .iter() + .any(|entry| entry.name.trim_end_matches('/') == "nested/deeper" && entry.is_dir) + ); + assert!( + fs::metadata(explicit_extract_path.join("nested")) + .await + .expect("nested directory should exist") + .is_dir() + ); + assert!( + fs::metadata(explicit_extract_path.join("nested/deeper")) + .await + .expect("nested deeper directory should exist") + .is_dir() + ); + + create_zip_with_options( + &auto_zip_path, + vec![("nested/deeper/file.txt".to_string(), b"hello".to_vec())], + ZipWriteOptions { + compression_level: CompressionLevel::Default, + create_directory_entries: true, + }, + ) + .await + .expect("zip creation with automatic directory entries should succeed"); + + let entries = extract_zip_with_limits(&auto_zip_path, &auto_extract_path, ArchiveLimits::default()) + .await + .expect("zip extraction with automatic directory entries should succeed"); + + assert!( + entries + .iter() + .any(|entry| entry.name.trim_end_matches('/') == "nested" && entry.is_dir) + ); + assert!( + entries + .iter() + .any(|entry| entry.name.trim_end_matches('/') == "nested/deeper" && entry.is_dir) + ); + assert!( + fs::metadata(auto_extract_path.join("nested")) + .await + .expect("nested directory should exist") + .is_dir() + ); + assert!( + fs::metadata(auto_extract_path.join("nested/deeper")) + .await + .expect("nested deeper directory should exist") + .is_dir() + ); + assert_eq!( + fs::read(auto_extract_path.join("nested/deeper/file.txt")) + .await + .expect("nested file should be extracted"), + b"hello" + ); + } + + #[tokio::test] + async fn test_create_zip_with_lots_of_small_files_round_trip() { + let temp = tempdir().expect("tempdir should be created"); + let zip_path = temp.path().join("many-small-files.zip"); + let extract_path = temp.path().join("extract"); + let files = (0..32) + .map(|index| (format!("batch/file-{index}.txt"), format!("payload-{index}").into_bytes())) + .collect::>(); + + create_zip_with_options( + &zip_path, + files.clone(), + ZipWriteOptions { + compression_level: CompressionLevel::Default, + create_directory_entries: true, + }, + ) + .await + .expect("zip creation for many small files should succeed"); + + let entries = extract_zip_with_limits(&zip_path, &extract_path, ArchiveLimits::default()) + .await + .expect("zip extraction for many small files should succeed"); + + assert!(entries.len() >= files.len()); + for (path, expected) in files { + assert_eq!( + fs::read(extract_path.join(path)) + .await + .expect("small file should be extracted"), + expected + ); } } + + #[tokio::test] + async fn test_extract_zip_to_path_with_limits_returns_summary() { + let temp = tempdir().expect("tempdir should be created"); + let zip_path = temp.path().join("summary.zip"); + let extract_path = temp.path().join("extract"); + + create_zip_with_options( + &zip_path, + vec![ + ("nested/".to_string(), Vec::new()), + ("nested/hello.txt".to_string(), b"hello".to_vec()), + ("world.txt".to_string(), b"world".to_vec()), + ], + ZipWriteOptions { + compression_level: CompressionLevel::Default, + create_directory_entries: false, + }, + ) + .await + .expect("zip creation for summary should succeed"); + + let summary = extract_zip_to_path_with_limits(&zip_path, &extract_path, ArchiveLimits::default()) + .await + .expect("zip extract summary should succeed"); + + assert_eq!(summary.entry_count, 3); + assert_eq!(summary.directory_count, 1); + assert_eq!(summary.file_count, 2); + assert_eq!(summary.total_unpacked_size, 10); + assert_eq!( + fs::read(extract_path.join("nested/hello.txt")) + .await + .expect("nested file should be extracted"), + b"hello" + ); + assert_eq!( + fs::read(extract_path.join("world.txt")) + .await + .expect("world file should be extracted"), + b"world" + ); + } + + #[tokio::test] + async fn test_zip_helper_exposes_stored_vs_deflated_metadata() { + let temp = tempdir().expect("tempdir should be created"); + let stored_zip_path = temp.path().join("stored.zip"); + let deflated_zip_path = temp.path().join("deflated.zip"); + let stored_extract_path = temp.path().join("stored-extract"); + let deflated_extract_path = temp.path().join("deflated-extract"); + let payload = b"compressible-content-".repeat(64); + + create_zip_with_options( + &stored_zip_path, + vec![("payload.txt".to_string(), payload.clone())], + ZipWriteOptions { + compression_level: CompressionLevel::Fastest, + create_directory_entries: false, + }, + ) + .await + .expect("stored zip creation should succeed"); + + create_zip_with_options( + &deflated_zip_path, + vec![("payload.txt".to_string(), payload.clone())], + ZipWriteOptions { + compression_level: CompressionLevel::Best, + create_directory_entries: false, + }, + ) + .await + .expect("deflated zip creation should succeed"); + + let stored_entries = extract_zip_with_limits(&stored_zip_path, &stored_extract_path, ArchiveLimits::default()) + .await + .expect("stored zip extraction should succeed"); + let deflated_entries = extract_zip_with_limits(&deflated_zip_path, &deflated_extract_path, ArchiveLimits::default()) + .await + .expect("deflated zip extraction should succeed"); + + assert_eq!(stored_entries[0].compression_method, "Stored"); + assert_eq!(deflated_entries[0].compression_method, "Deflated"); + assert_eq!( + fs::read(stored_extract_path.join("payload.txt")) + .await + .expect("stored payload should be extracted"), + payload + ); + assert_eq!( + fs::read(deflated_extract_path.join("payload.txt")) + .await + .expect("deflated payload should be extracted"), + payload + ); + assert!(deflated_entries[0].compressed_size <= stored_entries[0].compressed_size); + } + + #[tokio::test] + async fn test_create_zip_rejects_unsafe_entry_name() { + let temp = tempdir().expect("tempdir should be created"); + let zip_path = temp.path().join("archive.zip"); + + let err = create_zip_simple( + &zip_path, + vec![("../escape.txt".to_string(), b"escape".to_vec())], + CompressionLevel::Default, + ) + .await + .unwrap_err(); + + assert!(matches!(err, ZipError::UnsafeEntryPath(path) if path == "../escape.txt")); + } + + #[tokio::test] + async fn test_extract_zip_rejects_too_many_entries() { + let temp = tempdir().expect("tempdir should be created"); + let zip_path = temp.path().join("too-many.zip"); + let extract_path = temp.path().join("extract"); + + build_zip_file_with_entries(&zip_path, &[("one.txt", b"1"), ("two.txt", b"2")]) + .await + .expect("zip fixture should be created"); + + let err = extract_zip_with_limits( + &zip_path, + &extract_path, + ArchiveLimits { + max_entries: 1, + ..ArchiveLimits::default() + }, + ) + .await + .expect_err("zip entry count limit should fail"); + + assert!(matches!(err, ZipError::EntryCountLimitExceeded { count: 2, limit: 1 })); + } + + #[tokio::test] + async fn test_extract_zip_rejects_oversized_entry() { + let temp = tempdir().expect("tempdir should be created"); + let zip_path = temp.path().join("oversized.zip"); + let extract_path = temp.path().join("extract"); + + build_zip_file_with_entries(&zip_path, &[("big.txt", b"hello world")]) + .await + .expect("zip fixture should be created"); + + let err = extract_zip_with_limits( + &zip_path, + &extract_path, + ArchiveLimits { + max_entry_size: 4, + ..ArchiveLimits::default() + }, + ) + .await + .expect_err("zip entry size limit should fail"); + + assert!(matches!( + err, + ZipError::EntrySizeLimitExceeded { + path, + size: 11, + limit: 4, + } if path == "big.txt" + )); + } + + #[tokio::test] + async fn test_extract_zip_rejects_total_unpacked_size_limit() { + let temp = tempdir().expect("tempdir should be created"); + let zip_path = temp.path().join("too-large-total.zip"); + let extract_path = temp.path().join("extract"); + + build_zip_file_with_entries(&zip_path, &[("one.txt", b"12345"), ("two.txt", b"67890")]) + .await + .expect("zip fixture should be created"); + + let err = extract_zip_with_limits( + &zip_path, + &extract_path, + ArchiveLimits { + max_total_unpacked_size: 9, + ..ArchiveLimits::default() + }, + ) + .await + .expect_err("zip total size limit should fail"); + + assert!(matches!(err, ZipError::TotalUnpackedSizeLimitExceeded { size: 10, limit: 9 })); + } + + #[tokio::test] + async fn test_extract_zip_rejects_entry_path_length_limit() { + let temp = tempdir().expect("tempdir should be created"); + let zip_path = temp.path().join("long-path.zip"); + let extract_path = temp.path().join("extract"); + + build_zip_file_with_entries(&zip_path, &[("nested/hello.txt", b"hello")]) + .await + .expect("zip fixture should be created"); + + let err = extract_zip_with_limits( + &zip_path, + &extract_path, + ArchiveLimits { + max_path_length: 5, + ..ArchiveLimits::default() + }, + ) + .await + .expect_err("zip path length limit should fail"); + + assert!(matches!( + err, + ZipError::EntryPathTooLong { path, limit: 5, .. } if path == "nested/hello.txt" + )); + } + + #[tokio::test] + async fn test_decompress_file_round_trip() { + let temp = tempdir().expect("tempdir should be created"); + let input_path = temp.path().join("payload.txt.gz"); + let output_path = temp.path().join("payload.txt"); + let compressed = Compressor::new(CompressionFormat::Gzip) + .compress(b"payload") + .await + .expect("gzip compress should succeed"); + fs::write(&input_path, compressed) + .await + .expect("compressed input file should be written"); + + Decompressor::auto_detect(&input_path) + .decompress_file(&input_path, &output_path) + .await + .expect("gzip file decompress should succeed"); + + assert_eq!( + fs::read(&output_path) + .await + .expect("decompressed output file should be readable"), + b"payload" + ); + } } - -// #[tokio::test] -// async fn test_decompress() -> io::Result<()> { -// use std::path::Path; -// use tokio::fs::File; - -// let input_path = "/Users/weisd/Downloads/wsd.tar.gz"; // Replace with your compressed file path - -// let f = File::open(input_path).await?; - -// let Some(ext) = Path::new(input_path).extension().and_then(|s| s.to_str()) else { -// return Err(io::Error::new(io::ErrorKind::InvalidInput, "Unsupported file format")); -// }; - -// match decompress( -// f, -// CompressionFormat::from_extension(ext), -// |entry: tokio_tar::Entry>>| async move { -// let path = entry.path().unwrap(); -// println!("Extracted: {}", path.display()); -// Ok(()) -// }, -// ) -// .await -// { -// Ok(_) => println!("Decompression successful!"), -// Err(e) => println!("Decompression failed: {}", e), -// } - -// Ok(()) -// }