feat: add Mosaic data file reader

.github/workflows/ci.yml

@@ -58,7 +58,7 @@ jobs:
         run: cargo fmt --all -- --check
 
       - name: Clippy
-        run: cargo clippy --all-targets --workspace --features fulltext,vortex -- -D warnings
+        run: cargo clippy --all-targets --workspace --features fulltext,vortex,mosaic -- -D warnings
 
   build:
     runs-on: ${{ matrix.os }}
@@ -71,7 +71,7 @@ jobs:
     steps:
       - uses: actions/checkout@v6
       - name: Build
-        run: cargo build --features fulltext,vortex
+        run: cargo build --features fulltext,vortex,mosaic
 
   unit:
     runs-on: ${{ matrix.os }}
@@ -85,7 +85,7 @@ jobs:
       - uses: actions/checkout@v6
 
       - name: Test
-        run: cargo test -p paimon --all-targets --features fulltext,vortex
+        run: cargo test -p paimon --all-targets --features fulltext,vortex,mosaic
         env:
           RUST_LOG: DEBUG
           RUST_BACKTRACE: full

crates/integrations/datafusion/Cargo.toml

@@ -29,6 +29,7 @@ keywords = ["paimon", "datafusion", "integrations"]
 
 [features]
 fulltext = ["paimon/fulltext"]
+mosaic = ["paimon/mosaic"]
 vortex = ["paimon/vortex"]
 
 [dependencies]

crates/paimon/Cargo.toml

@@ -42,6 +42,7 @@ storage-all = [
     "storage-hdfs",
 ]
 fulltext = ["tantivy", "tempfile"]
+mosaic = ["dep:paimon-mosaic-core"]
 vortex = ["dep:vortex", "dep:kanal"]
 
 storage-memory = ["opendal/services-memory"]
@@ -101,6 +102,7 @@ uuid = { version = "1", features = ["v4"] }
 urlencoding = "2.1"
 tantivy = { version = "0.22", optional = true }
 tempfile = { version = "3", optional = true }
+paimon-mosaic-core = { version = "0.1.0", optional = true }
 vortex = { version = "0.68", features = ["tokio"], optional = true }
 kanal = { version = "0.1.1", optional = true }
 libloading = "0.9"

crates/paimon/src/arrow/format/mod.rs

@@ -17,6 +17,8 @@
 
 mod avro;
 pub(crate) mod blob;
+#[cfg(feature = "mosaic")]
+mod mosaic;
 mod orc;
 mod parquet;
 #[cfg(feature = "vortex")]
@@ -105,18 +107,36 @@ pub(crate) fn create_format_reader(
     } else if lower.ends_with(".avro") {
         Ok(Box::new(avro::AvroFormatReader))
     } else {
+        #[cfg(feature = "mosaic")]
+        if lower.ends_with(".mosaic") {
+            return Ok(Box::new(mosaic::MosaicFormatReader));
+        }
         #[cfg(feature = "vortex")]
         if lower.ends_with(".vortex") {
             return Ok(Box::new(vortex::VortexFormatReader));
         }
         Err(Error::Unsupported {
             message: format!(
-                "unsupported file format: expected .parquet, .blob, .orc, or .avro, got: {path}"
+                "unsupported file format: expected {}, got: {path}",
+                supported_read_formats().join(", ")
             ),
         })
     }
 }
 
+fn supported_read_formats() -> Vec<&'static str> {
+    vec![
+        ".parquet",
+        ".blob",
+        ".orc",
+        ".avro",
+        #[cfg(feature = "mosaic")]
+        ".mosaic",
+        #[cfg(feature = "vortex")]
+        ".vortex",
+    ]
+}
+
 /// Create a format writer that streams directly to storage.
 pub(crate) async fn create_format_writer(
     output: &OutputFile,