feat: Add ZIP compression type (#18257)

* feat: Add ZIP compression type

* chore: add meta test for Zip

Author: KKould

Cargo.lock

@@ -540,6 +540,7 @@ wiremock = "0.6"
 wkt = "0.11.1"
 xorf = { version = "0.11.0", default-features = false, features = ["binary-fuse"] }
 xorfilter-rs = "0.5"
+zip = "3.0.0"
 zstd = "0.12.3"
 
 # AST needed

Cargo.toml

@@ -16,6 +16,7 @@ futures = { workspace = true }
 log = { workspace = true }
 pin-project = { workspace = true }
 serde = { workspace = true }
+zip = { workspace = true }
 
 [dev-dependencies]
 env_logger = { workspace = true }

src/common/compress/Cargo.toml

@@ -40,6 +40,8 @@ pub enum CompressAlgorithm {
     ///
     /// Similar to [`CompressAlgorithm::Deflate`] and [`CompressAlgorithm::Gzip`]
     Zlib,
+    /// [Zip](https://pkware.cachefly.net/webdocs/APPNOTE/APPNOTE-6.3.10.TXT) compress format.
+    Zip,
     /// [Zstd](https://github.com/facebook/zstd) compression algorithm
     Zstd,
 }
@@ -56,6 +58,7 @@ impl CompressAlgorithm {
             CompressAlgorithm::Xz => "xz",
             CompressAlgorithm::Zlib => "zl",
             CompressAlgorithm::Zstd => "zstd",
+            CompressAlgorithm::Zip => "zip",
         }
     }
 
@@ -72,6 +75,7 @@ impl CompressAlgorithm {
             "xz" => Some(CompressAlgorithm::Xz),
             "zl" => Some(CompressAlgorithm::Zlib),
             "zstd" | "zst" => Some(CompressAlgorithm::Zstd),
+            "zip" => Some(CompressAlgorithm::Zip),
             _ => None,
         }
     }

src/common/compress/src/compress_algorithms.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::io::Cursor;
+use std::io::Read;
 use std::io::Result;
 use std::pin::Pin;
 use std::task::Context;
@@ -36,6 +38,7 @@ use futures::AsyncBufRead;
 use futures::AsyncRead;
 use log::trace;
 use pin_project::pin_project;
+use zip::ZipArchive;
 
 use crate::CompressAlgorithm;
 
@@ -73,6 +76,9 @@ impl From<CompressAlgorithm> for DecompressCodec {
             CompressAlgorithm::Xz => DecompressCodec::Xz(XzDecoder::new()),
             CompressAlgorithm::Zlib => DecompressCodec::Zlib(ZlibDecoder::new()),
             CompressAlgorithm::Zstd => DecompressCodec::Zstd(ZstdDecoder::new()),
+            CompressAlgorithm::Zip => {
+                unreachable!("Zip type requires additional judgment and use `decompress_all_zip`")
+            }
         }
     }
 }
@@ -350,6 +356,25 @@ impl DecompressDecoder {
         main.extend_from_slice(&tail);
         Ok(main)
     }
+
+    pub fn decompress_all_zip(compressed: &[u8]) -> databend_common_exception::Result<Vec<u8>> {
+        let mut zip = ZipArchive::new(Cursor::new(compressed)).map_err(|e| {
+            ErrorCode::InvalidCompressionData(format!("compression data invalid: {e}"))
+        })?;
+        if zip.len() > 1 {
+            return Err(ErrorCode::InvalidCompressionData(
+                "Zip only supports single file",
+            ));
+        }
+        let mut file = zip.by_index(0).map_err(|e| {
+            ErrorCode::InvalidCompressionData(format!("compression data invalid: {e}"))
+        })?;
+        let mut bytes = Vec::new();
+        file.read_to_end(&mut bytes)?;
+
+        Ok(bytes)
+    }
+
     // need to finish the decoding by adding a empty input
     pub fn decompress_batch(
         &mut self,
@@ -577,6 +602,22 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_decompress_reader_zip() -> databend_common_exception::Result<()> {
+        let _ = env_logger::try_init();
+
+        let mut rng = ThreadRng::default();
+        let mut content = vec![0; 16 * 1024 * 1024];
+        rng.fill_bytes(&mut content);
+
+        let compressed_content = CompressCodec::compress_all_zip(&content)?;
+        let result = DecompressDecoder::decompress_all_zip(&compressed_content)?;
+
+        assert_eq!(result, content);
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn test_decompress_reader_ontime_gzip() -> Result<()> {
         let _ = env_logger::try_init();

src/common/compress/src/decode.rs

@@ -12,7 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::io::Cursor;
 use std::io::Result;
+use std::io::Write;
 
 use async_compression::codec::BrotliEncoder;
 use async_compression::codec::BzEncoder;
@@ -27,6 +29,8 @@ use async_compression::util::PartialBuffer;
 use async_compression::Level;
 use brotli::enc::backward_references::BrotliEncoderParams;
 use databend_common_exception::ErrorCode;
+use zip::write::FileOptions;
+use zip::ZipWriter;
 
 use crate::CompressAlgorithm;
 
@@ -78,6 +82,9 @@ impl From<CompressAlgorithm> for CompressCodec {
             CompressAlgorithm::Zstd => {
                 CompressCodec::Zstd(ZstdEncoder::new(Level::Default.into_zstd()))
             }
+            CompressAlgorithm::Zip => {
+                unreachable!("Zip type requires additional judgment and use `compress_all_zip`")
+            }
         }
     }
 }
@@ -175,6 +182,26 @@ impl CompressCodec {
         }
         Ok(compress_bufs.concat())
     }
+
+    pub fn compress_all_zip(to_compress: &[u8]) -> databend_common_exception::Result<Vec<u8>> {
+        let mut cursor = Cursor::new(Vec::new());
+
+        let mut zip = ZipWriter::new(&mut cursor);
+        let options: FileOptions<()> = FileOptions::default()
+            .compression_method(zip::CompressionMethod::Deflated)
+            .unix_permissions(0o644);
+
+        // zip for archive files
+        zip.start_file("tmp", options)
+            .map_err(|e| ErrorCode::InvalidCompressionData(format!("zip start_file error: {e}")))?;
+
+        zip.write_all(to_compress)
+            .map_err(|e| ErrorCode::InvalidCompressionData(format!("zip write error: {e}")))?;
+
+        zip.finish()
+            .map_err(|e| ErrorCode::InvalidCompressionData(format!("zip finish error: {e}")))?;
+        Ok(cursor.into_inner())
+    }
 }
 
 #[cfg(test)]

src/common/compress/src/encode.rs

@@ -108,6 +108,7 @@ pub enum StageFileCompression {
     Lzo,
     Snappy,
     Xz,
+    Zip,
     None,
 }
 
@@ -138,7 +139,8 @@ impl FromStr for StageFileCompression {
             "snappy" => Ok(StageFileCompression::Snappy),
             "xz" => Ok(StageFileCompression::Xz),
             "none" => Ok(StageFileCompression::None),
-            _ => Err("Unknown file compression type, must one of { auto | gzip | bz2 | brotli | zstd | deflate | raw_deflate | lzo | snappy | xz | none }"
+            "zip" => Ok(StageFileCompression::Zip),
+            _ => Err("Unknown file compression type, must one of { auto | gzip | bz2 | brotli | zstd | deflate | raw_deflate | lzo | snappy | xz | zip | none }"
                 .to_string()),
         }
     }
@@ -157,6 +159,7 @@ impl Display for StageFileCompression {
             StageFileCompression::Lzo => write!(f, "lzo"),
             StageFileCompression::Snappy => write!(f, "snappy"),
             StageFileCompression::Xz => write!(f, "xz"),
+            StageFileCompression::Zip => write!(f, "zip"),
             StageFileCompression::None => write!(f, "none"),
         }
     }

src/meta/app/src/principal/user_stage.rs

@@ -83,6 +83,7 @@ impl FromToProtoEnum for mt::principal::StageFileCompression {
             pb::StageFileCompression::Snappy => Ok(mt::principal::StageFileCompression::Snappy),
             pb::StageFileCompression::None => Ok(mt::principal::StageFileCompression::None),
             pb::StageFileCompression::Xz => Ok(mt::principal::StageFileCompression::Xz),
+            pb::StageFileCompression::Zip => Ok(mt::principal::StageFileCompression::Zip),
         }
     }
 
@@ -101,6 +102,7 @@ impl FromToProtoEnum for mt::principal::StageFileCompression {
             mt::principal::StageFileCompression::Snappy => Ok(pb::StageFileCompression::Snappy),
             mt::principal::StageFileCompression::None => Ok(pb::StageFileCompression::None),
             mt::principal::StageFileCompression::Xz => Ok(pb::StageFileCompression::Xz),
+            mt::principal::StageFileCompression::Zip => Ok(pb::StageFileCompression::Zip),
         }
     }
 }

src/meta/proto-conv/src/file_format_from_to_protobuf_impl.rs

@@ -162,6 +162,7 @@ const META_CHANGE_LOG: &[(u64, &str)] = &[
     (130, "2025-06-19: Add: New UDF imports and packages in udf definition"),
     (131, "2025-06-24: Add: add use_logic_type in ParquetFileFormatParam and AvroFileFormatParam"),
     (132, "2025-06-25: Remove: SequenceMeta.start"),
+    (133, "2025-06-25: Add: Add new StageFileCompression Zip"),
     // Dear developer:
     //      If you're gonna add a new metadata version, you'll have to add a test for it.
     //      You could just copy an existing test file(e.g., `../tests/it/v024_table_meta.rs`)