diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 962e2a89..c32b5ff3 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/crates/integrations/datafusion/Cargo.toml b/crates/integrations/datafusion/Cargo.toml index e78d10fa..d7576d0b 100644 --- a/crates/integrations/datafusion/Cargo.toml +++ b/crates/integrations/datafusion/Cargo.toml @@ -29,7 +29,6 @@ keywords = ["paimon", "datafusion", "integrations"] [features] fulltext = ["paimon/fulltext"] -mosaic = ["paimon/mosaic"] vortex = ["paimon/vortex"] [dependencies] diff --git a/crates/integrations/datafusion/tests/mosaic_tables.rs b/crates/integrations/datafusion/tests/mosaic_tables.rs index 4a6f6799..426515d1 100644 --- a/crates/integrations/datafusion/tests/mosaic_tables.rs +++ b/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; diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml index 72fbfe92..d33debf6 100644 --- a/crates/paimon/Cargo.toml +++ b/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" diff --git a/crates/paimon/src/arrow/format/mod.rs b/crates/paimon/src/arrow/format/mod.rs index 87614bc6..2a2fcbde 100644 --- a/crates/paimon/src/arrow/format/mod.rs +++ b/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", diff --git a/crates/paimon/src/arrow/format/mosaic.rs b/crates/paimon/src/arrow/format/mosaic.rs index 2ab9ae89..0242475d 100644 --- a/crates/paimon/src/arrow/format/mosaic.rs +++ b/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, + validities: Vec, + key_nullable: bool, + value_nullable: bool, + keys: Vec>, + values: Vec>, + ) -> 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::() + .unwrap(); + assert_eq!(nums.value_offsets(), &[0, 3, 3, 5]); + assert!(nums.is_null(1)); + let nums_values = nums.values().as_any().downcast_ref::().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::() + .unwrap(); + assert_eq!(labels.value_offsets(), &[0, 2, 2, 3]); + assert!(labels.is_null(1)); + let label_keys = labels + .keys() + .as_any() + .downcast_ref::() + .unwrap(); + let label_values = labels + .values() + .as_any() + .downcast_ref::() + .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() { diff --git a/crates/paimon/src/table/data_file_reader.rs b/crates/paimon/src/table/data_file_reader.rs index 5a7e0bfe..84109929 100644 --- a/crates/paimon/src/table/data_file_reader.rs +++ b/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; diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md index b0317e53..2d43fadc 100644 --- a/docs/src/getting-started.md +++ b/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 diff --git a/docs/src/sql.md b/docs/src/sql.md index 86cf6128..c2b11cb1 100644 --- a/docs/src/sql.md +++ b/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.