Enable Mosaic reads by default

.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,mosaic -- -D warnings
+        run: cargo clippy --all-targets --workspace --features fulltext,vortex -- -D warnings
 
   build:
     runs-on: ${{ matrix.os }}
@@ -71,7 +71,7 @@ jobs:
     steps:
       - uses: actions/checkout@v7
       - name: Build
-        run: cargo build --features fulltext,vortex,mosaic
+        run: cargo build --features fulltext,vortex
 
   unit:
     runs-on: ${{ matrix.os }}
@@ -85,7 +85,7 @@ jobs:
       - uses: actions/checkout@v7
 
       - name: Test
-        run: cargo test -p paimon --all-targets --features fulltext,vortex,mosaic
+        run: cargo test -p paimon --all-targets --features fulltext,vortex
         env:
           RUST_LOG: DEBUG
           RUST_BACKTRACE: full

crates/integrations/datafusion/Cargo.toml

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

crates/integrations/datafusion/tests/mosaic_tables.rs

@@ -15,8 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#![cfg(feature = "mosaic")]
-
 //! Mosaic file format read compatibility tests.
 
 use std::path::Path;

crates/paimon/Cargo.toml

@@ -42,7 +42,6 @@ storage-all = [
     "storage-hdfs",
 ]
 fulltext = ["tantivy", "tempfile"]
-mosaic = ["dep:paimon-mosaic-core"]
 vortex = ["dep:vortex"]
 
 storage-memory = ["opendal/services-memory"]
@@ -104,7 +103,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 }
+paimon-mosaic-core = "0.2.0"
 paimon-vindex-core = "0.2.0"
 vortex = { version = "0.75.0", features = ["tokio"], optional = true }
 libloading = "0.9"

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

@@ -17,7 +17,6 @@
 
 mod avro;
 pub(crate) mod blob;
-#[cfg(feature = "mosaic")]
 mod mosaic;
 mod orc;
 mod parquet;
@@ -128,7 +127,6 @@ pub(crate) fn create_format_reader(
     } else if lower.ends_with(".row") {
         Box::new(row::RowFormatReader)
     } else {
-        #[cfg(feature = "mosaic")]
         if lower.ends_with(".mosaic") {
             return Ok(shredding::maybe_wrap_reader(
                 Box::new(mosaic::MosaicFormatReader),
@@ -159,7 +157,6 @@ fn supported_read_formats() -> Vec<&'static str> {
         ".orc",
         ".avro",
         ".row",
-        #[cfg(feature = "mosaic")]
         ".mosaic",
         #[cfg(feature = "vortex")]
         ".vortex",

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

@@ -24,7 +24,7 @@ use crate::table::{ArrowRecordBatchStream, RowRange};
 use crate::Error;
 use arrow_array::RecordBatch;
 use arrow_array::RecordBatchOptions;
-use arrow_schema::{DataType as ArrowDataType, SchemaRef, TimeUnit};
+use arrow_schema::{DataType as ArrowDataType, SchemaRef};
 use async_stream::try_stream;
 use async_trait::async_trait;
 use bytes::Bytes;
@@ -449,42 +449,7 @@ fn validate_mosaic_schema(schema: &SchemaRef) -> crate::Result<()> {
 }
 
 fn validate_mosaic_arrow_type(data_type: &ArrowDataType) -> Result<(), String> {
-    match data_type {
-        ArrowDataType::Boolean
-        | ArrowDataType::Int8
-        | ArrowDataType::Int16
-        | ArrowDataType::Int32
-        | ArrowDataType::Int64
-        | ArrowDataType::Float32
-        | ArrowDataType::Float64
-        | ArrowDataType::Date32
-        | ArrowDataType::Utf8
-        | ArrowDataType::Binary => Ok(()),
-        ArrowDataType::Time32(TimeUnit::Millisecond) => Ok(()),
-        ArrowDataType::Decimal128(precision, _) => {
-            if *precision == 0 || *precision > 38 {
-                Err(format!(
-                    "Decimal precision must be in 1..=38, got {precision}"
-                ))
-            } else {
-                Ok(())
-            }
-        }
-        ArrowDataType::Timestamp(
-            TimeUnit::Millisecond | TimeUnit::Microsecond | TimeUnit::Nanosecond,
-            _,
-        ) => Ok(()),
-        ArrowDataType::Struct(fields) if is_timestamp_nanos_struct(fields) => Ok(()),
-        other => Err(format!("unsupported Arrow type {other:?}")),
-    }
-}
-
-fn is_timestamp_nanos_struct(fields: &arrow_schema::Fields) -> bool {
-    fields.len() == 2
-        && fields[0].name() == "millis"
-        && *fields[0].data_type() == ArrowDataType::Int64
-        && fields[1].name() == "nanos_of_milli"
-        && *fields[1].data_type() == ArrowDataType::Int32
+    paimon_mosaic_core::types::validate_data_type(data_type)
 }
 
 fn selected_slices_for_row_group(
@@ -625,15 +590,17 @@ mod tests {
     use crate::arrow::format::{FilePredicates, FormatFileReader};
     use crate::spec::{
         ArrayType, BigIntType, BooleanType, DataType, DateType, Datum, DecimalType, DoubleType,
-        FloatType, IntType, LocalZonedTimestampType, Predicate, PredicateBuilder, RowType,
+        FloatType, IntType, LocalZonedTimestampType, MapType, Predicate, PredicateBuilder, RowType,
         SmallIntType, TimeType, TimestampType, TinyIntType, VarBinaryType, VarCharType,
     };
     use arrow_array::{
         Array, BinaryArray, BooleanArray, Date32Array, Decimal128Array, Float32Array, Float64Array,
-        Int16Array, Int32Array, Int64Array, Int8Array, StringArray, Time32MillisecondArray,
-        TimestampMicrosecondArray, TimestampMillisecondArray, TimestampNanosecondArray,
+        Int16Array, Int32Array, Int64Array, Int8Array, ListArray, MapArray, StringArray,
+        StructArray, Time32MillisecondArray, TimestampMicrosecondArray, TimestampMillisecondArray,
+        TimestampNanosecondArray,
     };
-    use arrow_schema::{DataType as ArrowDataType, Field, Schema};
+    use arrow_buffer::{BooleanBuffer, NullBuffer, OffsetBuffer, ScalarBuffer};
+    use arrow_schema::{DataType as ArrowDataType, Field, Schema, TimeUnit};
     use bytes::Bytes;
     use futures::TryStreamExt;
     use paimon_mosaic_core::spec::COMPRESSION_NONE;
@@ -739,6 +706,41 @@ mod tests {
         DataField::new(id, name.to_string(), data_type)
     }
 
+    fn test_map_array(
+        offsets: Vec<i32>,
+        validities: Vec<bool>,
+        key_nullable: bool,
+        value_nullable: bool,
+        keys: Vec<Option<&str>>,
+        values: Vec<Option<i32>>,
+    ) -> MapArray {
+        let entries = StructArray::try_new(
+            vec![
+                Arc::new(Field::new("key", ArrowDataType::Utf8, key_nullable)),
+                Arc::new(Field::new("value", ArrowDataType::Int32, value_nullable)),
+            ]
+            .into(),
+            vec![
+                Arc::new(StringArray::from(keys)),
+                Arc::new(Int32Array::from(values)),
+            ],
+            None,
+        )
+        .unwrap();
+        MapArray::try_new(
+            Arc::new(Field::new(
+                "entries",
+                ArrowDataType::Struct(entries.fields().clone()),
+                false,
+            )),
+            OffsetBuffer::new(ScalarBuffer::from(offsets)),
+            entries,
+            Some(NullBuffer::new(BooleanBuffer::from(validities))),
+            false,
+        )
+        .unwrap()
+    }
+
     fn arrow_schema() -> SchemaRef {
         Arc::new(Schema::new(vec![
             Field::new("id", ArrowDataType::Int32, false),
@@ -1302,8 +1304,12 @@ mod tests {
             fields[0].clone(),
             field(
                 3,
-                "new_items",
-                DataType::Array(ArrayType::new(DataType::Int(IntType::new()))),
+                "new_nested",
+                DataType::Row(RowType::new(vec![field(
+                    4,
+                    "value",
+                    DataType::Int(IntType::new()),
+                )])),
             ),
         ];
         let data = write_mosaic(&sample_batch());
@@ -1320,7 +1326,11 @@ mod tests {
         let projected = vec![field(
             0,
             "id",
-            DataType::Array(ArrayType::new(DataType::Int(IntType::new()))),
+            DataType::Row(RowType::new(vec![field(
+                1,
+                "value",
+                DataType::Int(IntType::new()),
+            )])),
         )];
         let data = write_mosaic(&sample_batch());
         let err = read_batches(data, &projected, None).await.unwrap_err();
@@ -1498,8 +1508,7 @@ mod tests {
     }
 
     /// Round-trips every scalar/temporal type Mosaic supports through write + read,
-    /// asserting values survive the format. ARRAY/MAP are intentionally excluded:
-    /// `paimon-mosaic-core` 0.1.0 does not support them and the reader rejects them.
+    /// asserting values survive the format. Collection types are covered separately.
     #[tokio::test]
     async fn test_read_full_types() {
         let fields = full_type_fields();
@@ -1658,6 +1667,91 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_read_array_and_map_types() {
+        let fields = vec![
+            field(
+                0,
+                "nums",
+                DataType::Array(ArrayType::new(DataType::Int(IntType::with_nullable(true)))),
+            ),
+            field(
+                1,
+                "labels",
+                DataType::Map(MapType::new(
+                    DataType::VarChar(VarCharType::with_nullable(false, 20).unwrap()),
+                    DataType::Int(IntType::with_nullable(true)),
+                )),
+            ),
+        ];
+        let schema = build_target_arrow_schema(&fields).unwrap();
+        let nums = ListArray::try_new(
+            Arc::new(Field::new("element", ArrowDataType::Int32, true)),
+            OffsetBuffer::new(ScalarBuffer::from(vec![0, 3, 3, 5])),
+            Arc::new(Int32Array::from(vec![
+                Some(1),
+                None,
+                Some(3),
+                Some(4),
+                Some(5),
+            ])),
+            Some(NullBuffer::new(BooleanBuffer::from(vec![
+                true, false, true,
+            ]))),
+        )
+        .unwrap();
+        let labels = test_map_array(
+            vec![0, 2, 2, 3],
+            vec![true, false, true],
+            false,
+            true,
+            vec![Some("a"), Some("b"), Some("c")],
+            vec![Some(10), None, Some(30)],
+        );
+        let batch = RecordBatch::try_new(schema, vec![Arc::new(nums), Arc::new(labels)]).unwrap();
+        let data = write_mosaic(&batch);
+        let batches = read_batches(data, &fields, None).await.unwrap();
+
+        assert_eq!(batches.len(), 1);
+        let result = &batches[0];
+        assert_eq!(result.num_rows(), 3);
+        assert_eq!(result.num_columns(), 2);
+
+        let nums = result
+            .column(0)
+            .as_any()
+            .downcast_ref::<ListArray>()
+            .unwrap();
+        assert_eq!(nums.value_offsets(), &[0, 3, 3, 5]);
+        assert!(nums.is_null(1));
+        let nums_values = nums.values().as_any().downcast_ref::<Int32Array>().unwrap();
+        assert_eq!(nums_values.value(0), 1);
+        assert!(nums_values.is_null(1));
+        assert_eq!(nums_values.value(4), 5);
+
+        let labels = result
+            .column(1)
+            .as_any()
+            .downcast_ref::<MapArray>()
+            .unwrap();
+        assert_eq!(labels.value_offsets(), &[0, 2, 2, 3]);
+        assert!(labels.is_null(1));
+        let label_keys = labels
+            .keys()
+            .as_any()
+            .downcast_ref::<StringArray>()
+            .unwrap();
+        let label_values = labels
+            .values()
+            .as_any()
+            .downcast_ref::<Int32Array>()
+            .unwrap();
+        assert_eq!(label_keys.value(0), "a");
+        assert_eq!(label_keys.value(2), "c");
+        assert!(label_values.is_null(1));
+        assert_eq!(label_values.value(2), 30);
+    }
+
     /// Null values in nullable columns must round-trip as nulls.
     #[tokio::test]
     async fn test_read_null_values() {

crates/paimon/src/table/data_file_reader.rs

@@ -976,7 +976,7 @@ mod row_tests {
     }
 }
 
-#[cfg(all(test, feature = "mosaic"))]
+#[cfg(test)]
 mod tests {
     use super::*;
     use crate::arrow::build_target_arrow_schema;

docs/src/getting-started.md

@@ -58,16 +58,9 @@ Available storage features:
 paimon = { git = "https://github.com/apache/paimon-rust", features = ["storage-cos"] }
 ```
 
-## Optional File Formats
+## Mosaic File Format
 
-Mosaic data files can be read by enabling the `mosaic` feature. This feature is not in the latest release yet; it is available on the `main` branch:
-
-```toml
-[dependencies]
-paimon = { git = "https://github.com/apache/paimon-rust", features = ["mosaic"] }
-```
-
-The current Mosaic support is read-only. Paimon Rust can read existing `.mosaic` data files in a Paimon table, but it does not write Mosaic data files yet.
+Mosaic data file reads are always available. The current Mosaic support is read-only: Paimon Rust can read existing `.mosaic` data files, including array and map columns, in a Paimon table, but it does not write Mosaic data files yet.
 
 ## Catalog Management
 

docs/src/sql.md

@@ -31,17 +31,7 @@ datafusion = "54.0.0"
 tokio = { version = "1", features = ["full"] }
 ```
 
-To query tables with Mosaic data files, enable the `mosaic` feature on both crates:
-
-```toml
-[dependencies]
-paimon = { version = "0.3.0", features = ["mosaic"] }
-paimon-datafusion = { version = "0.3.0", features = ["mosaic"] }
-datafusion = "54.0.0"
-tokio = { version = "1", features = ["full"] }
-```
-
-Mosaic support is currently read-only. SQL queries can read existing `.mosaic` files, but Paimon Rust does not write Mosaic data files yet.
+Mosaic support is always available and currently read-only. SQL queries can read existing `.mosaic` files, but Paimon Rust does not write Mosaic data files yet.
 
 ## SQL Support Scope
 
@@ -534,7 +524,7 @@ For primary-key tables, records with duplicate keys are deduplicated according t
 
 ### Mosaic Read Scope
 
-The Mosaic reader uses row-group statistics for conservative pruning when they are present. This pruning is not row-level filter enforcement; DataFusion still applies SQL filters above the reader to produce exact query results.
+The Mosaic reader supports scalar, temporal, array, and map columns. It uses row-group statistics for conservative pruning when they are present. This pruning is not row-level filter enforcement; DataFusion still applies SQL filters above the reader to produce exact query results.
 
 Unsupported or limited Mosaic areas include writing `.mosaic` files, emitting manifest `value_stats` for Mosaic writes, Mosaic bloom filters, and Mosaic-specific performance tuning.