This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new de7734e Enable Mosaic reads by default (#491)
de7734e is described below
commit de7734e2559eb21a7d2e5b38ee8fa53f923a5c74
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Jul 9 15:11:14 2026 +0800
Enable Mosaic reads by default (#491)
---
.github/workflows/ci.yml | 6 +-
crates/integrations/datafusion/Cargo.toml | 1 -
.../integrations/datafusion/tests/mosaic_tables.rs | 2 -
crates/paimon/Cargo.toml | 3 +-
crates/paimon/src/arrow/format/mod.rs | 3 -
crates/paimon/src/arrow/format/mosaic.rs | 186 ++++++++++++++++-----
crates/paimon/src/table/data_file_reader.rs | 2 +-
docs/src/getting-started.md | 11 +-
docs/src/sql.md | 14 +-
9 files changed, 149 insertions(+), 79 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 962e2a8..c32b5ff 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 e78d10f..d7576d0 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 4a6f679..426515d 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 72fbfe9..d33debf 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 87614bc..2a2fcbd 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 2ab9ae8..0242475 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<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() {
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 5a7e0bf..8410992 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 b0317e5..2d43fad 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 86cf612..c2b11cb 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.
