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 7e283579 [datafusion] Support Utf8View schemas and residual filtering
(#543)
7e283579 is described below
commit 7e283579c74f612448003645f681c679847828f8
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Jul 19 21:16:08 2026 +0800
[datafusion] Support Utf8View schemas and residual filtering (#543)
---
benchmarks/tpcds/tests/smoke.rs | 38 +++++
bindings/python/src/context.rs | 69 +++++++-
bindings/python/tests/test_datafusion.py | 21 +++
bindings/python/tests/test_write.py | 6 +-
crates/integrations/datafusion/src/catalog.rs | 110 ++++++++----
crates/integrations/datafusion/src/merge_into.rs | 4 +-
.../datafusion/src/physical_plan/scan.rs | 36 +++-
.../datafusion/src/physical_plan/sink.rs | 51 +++++-
crates/integrations/datafusion/src/sql_context.rs | 5 +-
crates/integrations/datafusion/src/table/mod.rs | 121 ++++++++++++-
crates/integrations/datafusion/src/update.rs | 4 +-
.../datafusion/tests/append_merge_into.rs | 12 +-
crates/integrations/datafusion/tests/blob_tests.rs | 24 ++-
crates/integrations/datafusion/tests/common/mod.rs | 54 +++---
.../datafusion/tests/dynamic_bucket_tables.rs | 56 +++---
.../datafusion/tests/merge_into_tests.rs | 33 ++--
.../integrations/datafusion/tests/mosaic_tables.rs | 24 +--
crates/integrations/datafusion/tests/pk_tables.rs | 189 +++++++--------------
.../integrations/datafusion/tests/read_tables.rs | 52 +++---
crates/paimon/src/arrow/format/parquet.rs | 86 ++++++++++
crates/paimon/src/arrow/residual.rs | 43 +++--
21 files changed, 689 insertions(+), 349 deletions(-)
diff --git a/benchmarks/tpcds/tests/smoke.rs b/benchmarks/tpcds/tests/smoke.rs
index ff175b0f..c86d73ef 100644
--- a/benchmarks/tpcds/tests/smoke.rs
+++ b/benchmarks/tpcds/tests/smoke.rs
@@ -94,6 +94,44 @@ async fn parquet_fixture_loads_into_paimon() {
assert_eq!(counts.value(0), 3);
}
+#[tokio::test]
+async fn loaded_paimon_varchar_is_exposed_as_utf8_view() {
+ let data = TempDir::new().unwrap();
+ let warehouse = TempDir::new().unwrap();
+ write_fixture(&data, "store_sales");
+ let session = open_catalog_session(
+ &BenchmarkRuntimeConfig::default(),
+ warehouse.path(),
+ "tpcds",
+ )
+ .await
+ .unwrap();
+ load_parquet_table(
+ &session,
+ data.path(),
+ "store_sales",
+ ExistingTablePolicy::Error,
+ )
+ .await
+ .unwrap();
+
+ let batches = session
+ .sql
+ .sql("SELECT arrow_typeof(name) FROM paimon.tpcds.store_sales LIMIT 1")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let types = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+
+ assert_eq!(types.value(0), "Utf8View");
+}
+
#[tokio::test]
async fn loaded_paimon_table_runs_warmups_and_measured_iterations() {
let data = TempDir::new().unwrap();
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index b2f00f68..2536c5d5 100644
--- a/bindings/python/src/context.rs
+++ b/bindings/python/src/context.rs
@@ -18,8 +18,10 @@
use std::collections::HashMap;
use std::sync::Arc;
+use arrow::compute::cast;
use arrow::datatypes::{DataType as ArrowDataType, Field as ArrowField};
use arrow::pyarrow::{FromPyArrow, ToPyArrow};
+use arrow::record_batch::{RecordBatch, RecordBatchOptions};
use datafusion::catalog::CatalogProvider;
use datafusion::logical_expr::{Signature, TypeSignature, Volatility};
use datafusion_ffi::catalog_provider::FFI_CatalogProvider;
@@ -37,6 +39,47 @@ use crate::table::PyTable;
use crate::udf::{build_python_scalar_udf, udf, PyPythonScalarUDFObject};
use paimon_datafusion::runtime::runtime;
+fn pyarrow_compatible_batch(batch: &RecordBatch) ->
arrow::error::Result<RecordBatch> {
+ let mut changed = false;
+ let fields = batch
+ .schema()
+ .fields()
+ .iter()
+ .map(|field| {
+ if field.data_type() == &ArrowDataType::Utf8View {
+ changed = true;
+
Arc::new(field.as_ref().clone().with_data_type(ArrowDataType::Utf8))
+ } else {
+ Arc::clone(field)
+ }
+ })
+ .collect::<Vec<_>>();
+
+ if !changed {
+ return Ok(batch.clone());
+ }
+
+ let schema = Arc::new(arrow::datatypes::Schema::new_with_metadata(
+ fields,
+ batch.schema().metadata().clone(),
+ ));
+ let columns = batch
+ .columns()
+ .iter()
+ .zip(schema.fields())
+ .map(|(column, field)| {
+ if column.data_type() == field.data_type() {
+ Ok(Arc::clone(column))
+ } else {
+ cast(column.as_ref(), field.data_type())
+ }
+ })
+ .collect::<arrow::error::Result<Vec<_>>>()?;
+ let options =
RecordBatchOptions::new().with_row_count(Some(batch.num_rows()));
+
+ RecordBatch::try_new_with_options(schema, columns, &options)
+}
+
fn build_paimon_catalog(catalog_options: HashMap<String, String>) ->
PyResult<Arc<dyn Catalog>> {
let rt = runtime();
rt.block_on(async {
@@ -75,13 +118,16 @@ impl PaimonCatalog {
#[new]
fn new(catalog_options: HashMap<String, String>) -> PyResult<Self> {
let catalog = build_paimon_catalog(catalog_options)?;
- let provider = Arc::new(PaimonCatalogProvider::new(
- None,
- Arc::clone(&catalog),
- Default::default(),
- Default::default(),
- None,
- ));
+ let provider = Arc::new(
+ PaimonCatalogProvider::new(
+ None,
+ Arc::clone(&catalog),
+ Default::default(),
+ Default::default(),
+ None,
+ )
+ .with_schema_force_view_types(false),
+ );
Ok(Self { catalog, provider })
}
@@ -365,7 +411,14 @@ impl PySQLContext {
})?;
batches
.iter()
- .map(|batch| Ok(batch.to_pyarrow(py)?.unbind()))
+ .map(|batch| {
+ let batch = pyarrow_compatible_batch(batch).map_err(|err| {
+ PyValueError::new_err(format!(
+ "Failed to convert query result for PyArrow
compatibility: {err}"
+ ))
+ })?;
+ Ok(batch.to_pyarrow(py)?.unbind())
+ })
.collect()
}
}
diff --git a/bindings/python/tests/test_datafusion.py
b/bindings/python/tests/test_datafusion.py
index ca4b64ac..916e1cc3 100644
--- a/bindings/python/tests/test_datafusion.py
+++ b/bindings/python/tests/test_datafusion.py
@@ -478,6 +478,27 @@ def test_query_simple_table_via_catalog_provider():
]
+def test_catalog_provider_returns_pyarrow_compatible_strings():
+ with tempfile.TemporaryDirectory() as warehouse:
+ writer = SQLContext()
+ writer.register_catalog("paimon", {"warehouse": warehouse})
+ writer.sql("CREATE TABLE paimon.default.users (id INT, name STRING)")
+ writer.sql(
+ "INSERT INTO paimon.default.users VALUES (2, 'bob'), (1, 'alice')"
+ )
+
+ ctx = SessionContext()
+ ctx.register_catalog_provider(
+ "paimon", PaimonCatalog({"warehouse": warehouse})
+ )
+
+ batches = ctx.sql("SELECT id, name FROM
paimon.default.users").collect()
+ assert batches[0].schema.field("name").type == pa.string()
+ assert pa.Table.from_batches(batches).sort_by("id").to_pylist() == [
+ {"id": 1, "name": "alice"},
+ {"id": 2, "name": "bob"},
+ ]
+
def test_sql_context_ddl_dml():
with tempfile.TemporaryDirectory() as warehouse:
diff --git a/bindings/python/tests/test_write.py
b/bindings/python/tests/test_write.py
index 66d7d10d..8ef53e7a 100644
--- a/bindings/python/tests/test_write.py
+++ b/bindings/python/tests/test_write.py
@@ -56,9 +56,9 @@ def test_write_commit_read_roundtrip():
messages = write.prepare_commit()
assert len(messages) >= 1 # cover API shape in the
first test
wb.new_commit().commit(messages) # same wb → shared commit_user
- result = pa.Table.from_batches(
- ctx.sql("SELECT id, name FROM paimon.wdb.t")
- ).sort_by("id").to_pydict()
+ batches = ctx.sql("SELECT id, name FROM paimon.wdb.t")
+ assert batches[0].schema.field("name").type == pa.string()
+ result = pa.Table.from_batches(batches).sort_by("id").to_pydict()
assert result == {"id": [1, 2, 3], "name": ["a", "b", "c"]}
diff --git a/crates/integrations/datafusion/src/catalog.rs
b/crates/integrations/datafusion/src/catalog.rs
index 8ec6f861..26436397 100644
--- a/crates/integrations/datafusion/src/catalog.rs
+++ b/crates/integrations/datafusion/src/catalog.rs
@@ -64,6 +64,7 @@ pub struct PaimonCatalogProvider {
temp_tables: Arc<RwLock<HashMap<String, Arc<MemorySchemaProvider>>>>,
blob_reader_registry: BlobReaderRegistry,
session_state: Option<SessionStateProvider>,
+ schema_force_view_types: bool,
}
impl Debug for PaimonCatalogProvider {
@@ -88,8 +89,19 @@ impl PaimonCatalogProvider {
temp_tables: Arc::new(RwLock::new(HashMap::new())),
blob_reader_registry,
session_state,
+ schema_force_view_types: true,
}
}
+
+ /// Configure whether table schemas use Arrow view types when available.
+ ///
+ /// Disable this for consumers that cannot operate on Arrow view arrays.
This changes the
+ /// schema exposed to DataFusion, so query operators above the table scan
will use the classic
+ /// Arrow types as well.
+ pub fn with_schema_force_view_types(mut self, schema_force_view_types:
bool) -> Self {
+ self.schema_force_view_types = schema_force_view_types;
+ self
+ }
}
impl CatalogProvider for PaimonCatalogProvider {
@@ -112,6 +124,7 @@ impl CatalogProvider for PaimonCatalogProvider {
let blob_reader_registry = self.blob_reader_registry.clone();
let catalog_name = self.catalog_name.clone();
let session_state = self.session_state.clone();
+ let schema_force_view_types = self.schema_force_view_types;
let name = name.to_string();
let temp_provider = {
@@ -122,26 +135,32 @@ impl CatalogProvider for PaimonCatalogProvider {
block_on_with_runtime(
async move {
match catalog.get_database(&name).await {
- Ok(_) => Some(Arc::new(PaimonSchemaProvider::new(
- catalog_name,
- Arc::clone(&catalog),
- name,
- dynamic_options,
- temp_provider,
- blob_reader_registry,
- session_state,
- )) as Arc<dyn SchemaProvider>),
+ Ok(_) => Some(Arc::new(
+ PaimonSchemaProvider::new(
+ catalog_name,
+ Arc::clone(&catalog),
+ name,
+ dynamic_options,
+ temp_provider,
+ blob_reader_registry,
+ session_state,
+ )
+ .with_schema_force_view_types(schema_force_view_types),
+ ) as Arc<dyn SchemaProvider>),
Err(paimon::Error::DatabaseNotExist { .. }) => {
if temp_provider.is_some() {
- Some(Arc::new(PaimonSchemaProvider::new(
- catalog_name,
- Arc::clone(&catalog),
- name,
- dynamic_options,
- temp_provider,
- blob_reader_registry,
- session_state,
- )) as Arc<dyn SchemaProvider>)
+ Some(Arc::new(
+ PaimonSchemaProvider::new(
+ catalog_name,
+ Arc::clone(&catalog),
+ name,
+ dynamic_options,
+ temp_provider,
+ blob_reader_registry,
+ session_state,
+ )
+
.with_schema_force_view_types(schema_force_view_types),
+ ) as Arc<dyn SchemaProvider>)
} else {
None
}
@@ -166,6 +185,7 @@ impl CatalogProvider for PaimonCatalogProvider {
let blob_reader_registry = self.blob_reader_registry.clone();
let catalog_name = self.catalog_name.clone();
let session_state = self.session_state.clone();
+ let schema_force_view_types = self.schema_force_view_types;
let name = name.to_string();
block_on_with_runtime(
async move {
@@ -173,15 +193,18 @@ impl CatalogProvider for PaimonCatalogProvider {
.create_database(&name, false, HashMap::new())
.await
.map_err(to_datafusion_error)?;
- Ok(Some(Arc::new(PaimonSchemaProvider::new(
- catalog_name,
- Arc::clone(&catalog),
- name,
- dynamic_options,
- None,
- blob_reader_registry,
- session_state,
- )) as Arc<dyn SchemaProvider>))
+ Ok(Some(Arc::new(
+ PaimonSchemaProvider::new(
+ catalog_name,
+ Arc::clone(&catalog),
+ name,
+ dynamic_options,
+ None,
+ blob_reader_registry,
+ session_state,
+ )
+ .with_schema_force_view_types(schema_force_view_types),
+ ) as Arc<dyn SchemaProvider>))
},
"paimon catalog access thread panicked",
)
@@ -197,6 +220,7 @@ impl CatalogProvider for PaimonCatalogProvider {
let blob_reader_registry = self.blob_reader_registry.clone();
let catalog_name = self.catalog_name.clone();
let session_state = self.session_state.clone();
+ let schema_force_view_types = self.schema_force_view_types;
let name = name.to_string();
block_on_with_runtime(
async move {
@@ -204,15 +228,18 @@ impl CatalogProvider for PaimonCatalogProvider {
.drop_database(&name, false, cascade)
.await
.map_err(to_datafusion_error)?;
- Ok(Some(Arc::new(PaimonSchemaProvider::new(
- catalog_name,
- Arc::clone(&catalog),
- name,
- dynamic_options,
- None,
- blob_reader_registry,
- session_state,
- )) as Arc<dyn SchemaProvider>))
+ Ok(Some(Arc::new(
+ PaimonSchemaProvider::new(
+ catalog_name,
+ Arc::clone(&catalog),
+ name,
+ dynamic_options,
+ None,
+ blob_reader_registry,
+ session_state,
+ )
+ .with_schema_force_view_types(schema_force_view_types),
+ ) as Arc<dyn SchemaProvider>))
},
"paimon catalog access thread panicked",
)
@@ -313,6 +340,7 @@ pub struct PaimonSchemaProvider {
temp_provider: Option<Arc<MemorySchemaProvider>>,
blob_reader_registry: BlobReaderRegistry,
session_state: Option<SessionStateProvider>,
+ schema_force_view_types: bool,
}
impl Debug for PaimonSchemaProvider {
@@ -343,8 +371,14 @@ impl PaimonSchemaProvider {
temp_provider,
blob_reader_registry,
session_state,
+ schema_force_view_types: true,
}
}
+
+ fn with_schema_force_view_types(mut self, schema_force_view_types: bool)
-> Self {
+ self.schema_force_view_types = schema_force_view_types;
+ self
+ }
}
#[async_trait]
@@ -409,6 +443,7 @@ impl SchemaProvider for PaimonSchemaProvider {
let blob_reader_registry = self.blob_reader_registry.clone();
let catalog_name = self.catalog_name.clone();
let session_state = self.session_state.clone();
+ let schema_force_view_types = self.schema_force_view_types;
let identifier = Identifier::new(self.database.clone(),
object.table().to_string());
let branch = object.branch().map(str::to_string);
await_with_runtime(async move {
@@ -440,7 +475,8 @@ impl SchemaProvider for PaimonSchemaProvider {
blob_reader_registry,
table_definition,
)?
- };
+ }
+ .with_schema_force_view_types(schema_force_view_types)?;
Ok(Some(Arc::new(provider) as Arc<dyn TableProvider>))
}
Err(paimon::Error::TableNotExist { .. }) => {
diff --git a/crates/integrations/datafusion/src/merge_into.rs
b/crates/integrations/datafusion/src/merge_into.rs
index 2620d9d4..97c36079 100644
--- a/crates/integrations/datafusion/src/merge_into.rs
+++ b/crates/integrations/datafusion/src/merge_into.rs
@@ -1782,7 +1782,7 @@ mod tests {
let names = batch
.column(1)
.as_any()
- .downcast_ref::<datafusion::arrow::array::StringArray>()
+ .downcast_ref::<datafusion::arrow::array::StringViewArray>()
.unwrap();
let values = batch
.column(2)
@@ -1925,7 +1925,7 @@ mod tests {
let names = batch
.column(1)
.as_any()
- .downcast_ref::<datafusion::arrow::array::StringArray>()
+ .downcast_ref::<datafusion::arrow::array::StringViewArray>()
.unwrap();
let values = batch
.column(2)
diff --git a/crates/integrations/datafusion/src/physical_plan/scan.rs
b/crates/integrations/datafusion/src/physical_plan/scan.rs
index 55d28f3b..d31891ed 100644
--- a/crates/integrations/datafusion/src/physical_plan/scan.rs
+++ b/crates/integrations/datafusion/src/physical_plan/scan.rs
@@ -17,7 +17,9 @@
use std::sync::Arc;
+use datafusion::arrow::compute::cast;
use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datafusion::arrow::record_batch::{RecordBatch, RecordBatchOptions};
use datafusion::common::stats::Precision;
use datafusion::common::Statistics;
use datafusion::error::Result as DFResult;
@@ -33,6 +35,33 @@ use paimon::DataSplit;
use crate::error::to_datafusion_error;
+fn to_datafusion_batch(batch: RecordBatch, schema: &ArrowSchemaRef) ->
DFResult<RecordBatch> {
+ if batch.num_columns() != schema.fields().len() {
+ return Err(datafusion::error::DataFusionError::Execution(format!(
+ "Paimon reader returned {} columns for DataFusion schema with {}
fields",
+ batch.num_columns(),
+ schema.fields().len()
+ )));
+ }
+
+ let row_count = batch.num_rows();
+ let columns = batch
+ .columns()
+ .iter()
+ .zip(schema.fields())
+ .map(|(column, field)| {
+ if column.data_type() == field.data_type() {
+ Ok(Arc::clone(column))
+ } else {
+ cast(column.as_ref(), field.data_type()).map_err(Into::into)
+ }
+ })
+ .collect::<DFResult<Vec<_>>>()?;
+ let options = RecordBatchOptions::new().with_row_count(Some(row_count));
+
+ RecordBatch::try_new_with_options(Arc::clone(schema), columns,
&options).map_err(Into::into)
+}
+
/// Execution plan that scans a Paimon table with optional column projection.
///
/// Planning is performed eagerly in
[`super::super::table::PaimonTableProvider::scan`],
@@ -172,7 +201,12 @@ impl ExecutionPlan for PaimonTableScan {
let read = read_builder.new_read().map_err(to_datafusion_error)?;
let stream = read.to_arrow(&splits).map_err(to_datafusion_error)?;
- let stream = stream.map(|r| r.map_err(to_datafusion_error));
+ let batch_schema = Arc::clone(&schema);
+ let stream = stream.map(move |result| {
+ result
+ .map_err(to_datafusion_error)
+ .and_then(|batch| to_datafusion_batch(batch,
&batch_schema))
+ });
Ok::<_,
datafusion::error::DataFusionError>(RecordBatchStreamAdapter::new(
schema,
diff --git a/crates/integrations/datafusion/src/physical_plan/sink.rs
b/crates/integrations/datafusion/src/physical_plan/sink.rs
index 3d40e2ad..64575c02 100644
--- a/crates/integrations/datafusion/src/physical_plan/sink.rs
+++ b/crates/integrations/datafusion/src/physical_plan/sink.rs
@@ -21,7 +21,11 @@ use std::fmt;
use std::sync::Arc;
use async_trait::async_trait;
-use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datafusion::arrow::compute::cast;
+use datafusion::arrow::datatypes::{
+ DataType as ArrowDataType, Schema, SchemaRef as ArrowSchemaRef,
+};
+use datafusion::arrow::record_batch::{RecordBatch, RecordBatchOptions};
use datafusion::datasource::sink::DataSink;
use datafusion::error::Result as DFResult;
use datafusion::execution::SendableRecordBatchStream;
@@ -32,6 +36,49 @@ use paimon::table::Table;
use crate::error::to_datafusion_error;
+fn to_paimon_batch(batch: RecordBatch) -> DFResult<RecordBatch> {
+ if !batch
+ .schema()
+ .fields()
+ .iter()
+ .any(|field| field.data_type() == &ArrowDataType::Utf8View)
+ {
+ return Ok(batch);
+ }
+
+ let fields = batch
+ .schema()
+ .fields()
+ .iter()
+ .map(|field| {
+ if field.data_type() == &ArrowDataType::Utf8View {
+
Arc::new(field.as_ref().clone().with_data_type(ArrowDataType::Utf8))
+ } else {
+ Arc::clone(field)
+ }
+ })
+ .collect::<Vec<_>>();
+ let schema = Arc::new(Schema::new_with_metadata(
+ fields,
+ batch.schema().metadata().clone(),
+ ));
+ let columns = batch
+ .columns()
+ .iter()
+ .zip(schema.fields())
+ .map(|(column, field)| {
+ if column.data_type() == field.data_type() {
+ Ok(Arc::clone(column))
+ } else {
+ cast(column.as_ref(), field.data_type()).map_err(Into::into)
+ }
+ })
+ .collect::<DFResult<Vec<_>>>()?;
+ let options =
RecordBatchOptions::new().with_row_count(Some(batch.num_rows()));
+
+ RecordBatch::try_new_with_options(schema, columns,
&options).map_err(Into::into)
+}
+
/// DataSink that writes RecordBatches to a Paimon table.
///
/// Uses the Paimon write pipeline: `WriteBuilder` → `TableWrite` →
`TableCommit`.
@@ -84,7 +131,7 @@ impl DataSink for PaimonDataSink {
let mut row_count = 0u64;
while let Some(batch) = data.next().await {
- let batch = batch?;
+ let batch = to_paimon_batch(batch?)?;
row_count += batch.num_rows() as u64;
tw.write_arrow_batch(&batch)
.await
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index 4ad6897d..dedd7ebc 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -3272,6 +3272,7 @@ mod tests {
use std::sync::Mutex;
use async_trait::async_trait;
+ use datafusion::arrow::array::StringViewArray;
use paimon::catalog::Database;
use paimon::spec::{
DataField as PaimonDataField, DataType as PaimonDataType, IntType,
Schema as PaimonSchema,
@@ -6691,7 +6692,7 @@ mod tests {
let pts = batch
.column(0)
.as_any()
- .downcast_ref::<StringArray>()
+ .downcast_ref::<StringViewArray>()
.unwrap();
let ids = batch
.column(1)
@@ -6739,7 +6740,7 @@ mod tests {
let pts = batch
.column(0)
.as_any()
- .downcast_ref::<StringArray>()
+ .downcast_ref::<StringViewArray>()
.unwrap();
let ids = batch
.column(1)
diff --git a/crates/integrations/datafusion/src/table/mod.rs
b/crates/integrations/datafusion/src/table/mod.rs
index c84a48e0..1bfc47fc 100644
--- a/crates/integrations/datafusion/src/table/mod.rs
+++ b/crates/integrations/datafusion/src/table/mod.rs
@@ -21,7 +21,9 @@ use std::fmt::Write as _;
use std::sync::Arc;
use async_trait::async_trait;
-use datafusion::arrow::datatypes::{Field, Schema, SchemaRef as ArrowSchemaRef};
+use datafusion::arrow::datatypes::{
+ DataType as ArrowDataType, Field, Schema, SchemaRef as ArrowSchemaRef,
+};
use datafusion::catalog::Session;
use datafusion::datasource::sink::DataSinkExec;
use datafusion::datasource::{TableProvider, TableType};
@@ -44,6 +46,8 @@ use crate::filter_pushdown::{analyze_filters,
classify_filter_pushdown};
use crate::physical_plan::PaimonTableScan;
use crate::runtime::await_with_runtime;
+const PARQUET_FIELD_ID_META_KEY: &str = "PARQUET:field_id";
+
pub(crate) fn datafusion_read_fields(table: &Table) -> Vec<DataField> {
let mut fields = table.schema().fields().to_vec();
if CoreOptions::new(table.schema().options()).data_evolution_enabled() {
@@ -56,6 +60,38 @@ pub(crate) fn datafusion_read_fields(table: &Table) ->
Vec<DataField> {
fields
}
+fn datafusion_arrow_schema(
+ fields: &[DataField],
+ schema_force_view_types: bool,
+) -> DFResult<ArrowSchemaRef> {
+ let paimon_schema =
+
paimon::arrow::build_target_arrow_schema(fields).map_err(to_datafusion_error)?;
+ let fields = paimon_schema
+ .fields()
+ .iter()
+ .map(|field| {
+ let mut metadata = field.metadata().clone();
+ metadata.remove(PARQUET_FIELD_ID_META_KEY);
+ let data_type = match field.data_type() {
+ ArrowDataType::Utf8 if schema_force_view_types =>
ArrowDataType::Utf8View,
+ data_type => data_type.clone(),
+ };
+ Arc::new(
+ field
+ .as_ref()
+ .clone()
+ .with_data_type(data_type)
+ .with_metadata(metadata),
+ )
+ })
+ .collect::<Vec<_>>();
+
+ Ok(Arc::new(Schema::new_with_metadata(
+ fields,
+ paimon_schema.metadata().clone(),
+ )))
+}
+
/// Read-only table provider for a Paimon table.
///
/// Supports full table scan, column projection, and predicate pushdown for
@@ -86,8 +122,7 @@ impl PaimonTableProvider {
table_definition: Option<String>,
) -> DFResult<Self> {
let fields = datafusion_read_fields(&table);
- let schema =
-
paimon::arrow::build_target_arrow_schema(&fields).map_err(to_datafusion_error)?;
+ let schema = datafusion_arrow_schema(&fields, true)?;
Ok(Self {
table,
schema,
@@ -114,6 +149,18 @@ impl PaimonTableProvider {
Self::try_new_with_table_definition(table, table_definition)
}
+ pub(crate) fn with_schema_force_view_types(
+ mut self,
+ schema_force_view_types: bool,
+ ) -> DFResult<Self> {
+ if schema_force_view_types {
+ return Ok(self);
+ }
+ let fields = datafusion_read_fields(&self.table);
+ self.schema = datafusion_arrow_schema(&fields,
schema_force_view_types)?;
+ Ok(self)
+ }
+
pub fn table(&self) -> &Table {
&self.table
}
@@ -479,6 +526,7 @@ mod tests {
use datafusion::logical_expr::{col, lit, Expr};
use datafusion::prelude::{SessionConfig, SessionContext};
use paimon::catalog::Identifier;
+ use paimon::spec::{ArrayType, MapType, RowType, VarCharType};
use paimon::{Catalog, CatalogOptions, DataSplit, FileSystemCatalog,
Options};
use crate::physical_plan::PaimonTableScan;
@@ -749,6 +797,71 @@ mod tests {
PaimonTableProvider::try_new(table).expect("provider should be
created")
}
+ #[tokio::test]
+ async fn test_datafusion_schema_hides_paimon_field_ids() {
+ let provider = data_evolution_projection_pruning_provider().await;
+
+ for field in provider.schema().fields() {
+ assert!(
+ !field.metadata().contains_key("PARQUET:field_id"),
+ "storage field id leaked through DataFusion schema for {}",
+ field.name()
+ );
+ }
+ }
+
+ #[test]
+ fn test_datafusion_schema_uses_views_only_for_top_level_strings() {
+ let string_type = || DataType::VarChar(VarCharType::string_type());
+ let schema = datafusion_arrow_schema(
+ &[
+ DataField::new(0, "plain".to_string(), string_type()),
+ DataField::new(
+ 1,
+ "array".to_string(),
+ DataType::Array(ArrayType::new(string_type())),
+ ),
+ DataField::new(
+ 2,
+ "map".to_string(),
+ DataType::Map(MapType::new(string_type(), string_type())),
+ ),
+ DataField::new(
+ 3,
+ "row".to_string(),
+ DataType::Row(RowType::new(vec![DataField::new(
+ 4,
+ "nested".to_string(),
+ string_type(),
+ )])),
+ ),
+ ],
+ true,
+ )
+ .expect("DataFusion schema should be created");
+
+ assert_eq!(schema.field(0).data_type(), &ArrowDataType::Utf8View);
+
+ let ArrowDataType::List(element) = schema.field(1).data_type() else {
+ panic!("array field should map to an Arrow List");
+ };
+ assert_eq!(element.data_type(), &ArrowDataType::Utf8);
+
+ let ArrowDataType::Map(entries, _) = schema.field(2).data_type() else {
+ panic!("map field should map to an Arrow Map");
+ };
+ let ArrowDataType::Struct(map_fields) = entries.data_type() else {
+ panic!("map entries should map to an Arrow Struct");
+ };
+ assert_eq!(map_fields[0].data_type(), &ArrowDataType::Utf8);
+ assert_eq!(map_fields[1].data_type(), &ArrowDataType::Utf8);
+
+ let ArrowDataType::Struct(row_fields) = schema.field(3).data_type()
else {
+ panic!("row field should map to an Arrow Struct");
+ };
+ assert_eq!(row_fields[0].data_type(), &ArrowDataType::Utf8);
+ }
+
fn planned_file_names(scan: &PaimonTableScan) -> Vec<String> {
let mut names = scan
.planned_partitions()
@@ -1107,7 +1220,7 @@ mod tests {
let pts = batch
.column(0)
.as_any()
- .downcast_ref::<datafusion::arrow::array::StringArray>()
+ .downcast_ref::<datafusion::arrow::array::StringViewArray>()
.unwrap();
let ids = batch
.column(1)
diff --git a/crates/integrations/datafusion/src/update.rs
b/crates/integrations/datafusion/src/update.rs
index e78037a5..5362b09a 100644
--- a/crates/integrations/datafusion/src/update.rs
+++ b/crates/integrations/datafusion/src/update.rs
@@ -335,7 +335,7 @@ mod tests {
use super::*;
use std::sync::Arc;
- use datafusion::arrow::array::{Int32Array, StringArray, UInt64Array};
+ use datafusion::arrow::array::{Int32Array, StringViewArray, UInt64Array};
use datafusion::sql::sqlparser::dialect::GenericDialect;
use datafusion::sql::sqlparser::parser::Parser;
use paimon::catalog::{Catalog, Identifier};
@@ -412,7 +412,7 @@ mod tests {
let names = batch
.column(1)
.as_any()
- .downcast_ref::<StringArray>()
+ .downcast_ref::<StringViewArray>()
.unwrap();
let values = batch
.column(2)
diff --git a/crates/integrations/datafusion/tests/append_merge_into.rs
b/crates/integrations/datafusion/tests/append_merge_into.rs
index c9d66bb0..3cd3c97d 100644
--- a/crates/integrations/datafusion/tests/append_merge_into.rs
+++ b/crates/integrations/datafusion/tests/append_merge_into.rs
@@ -22,12 +22,12 @@
mod common;
-use arrow_array::{Array, Int32Array, StringArray};
+use arrow_array::{Array, Int32Array};
use paimon_datafusion::SQLContext;
use common::{
collect_int_int_str, collect_int_str, collect_three_ints,
create_sql_context, create_test_env,
- exec,
+ exec, string_value,
};
// ======================= Helpers =======================
@@ -278,11 +278,7 @@ async fn test_partial_insert_with_null() {
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
- let c = batch
- .column(2)
- .as_any()
- .downcast_ref::<StringArray>()
- .unwrap();
+ let c = batch.column(2);
for i in 0..batch.num_rows() {
rows.push((
a.value(i),
@@ -290,7 +286,7 @@ async fn test_partial_insert_with_null() {
if c.is_null(i) {
None
} else {
- Some(c.value(i).to_string())
+ Some(string_value(c.as_ref(), i).to_string())
},
));
}
diff --git a/crates/integrations/datafusion/tests/blob_tests.rs
b/crates/integrations/datafusion/tests/blob_tests.rs
index b844f65e..59058d9b 100644
--- a/crates/integrations/datafusion/tests/blob_tests.rs
+++ b/crates/integrations/datafusion/tests/blob_tests.rs
@@ -21,8 +21,8 @@
mod common;
-use arrow_array::{Array, BinaryArray, Int32Array, RecordBatch, StringArray};
-use common::{assert_sql_error, create_sql_context, create_test_env, exec};
+use arrow_array::{Array, BinaryArray, Int32Array, RecordBatch};
+use common::{assert_sql_error, create_sql_context, create_test_env, exec,
string_value};
use paimon::catalog::Identifier;
use paimon::spec::{BlobDescriptor, BlobViewStruct};
use paimon::table::BranchManager;
@@ -54,11 +54,7 @@ fn collect_id_name_picture(batches: &[RecordBatch]) ->
Vec<(i32, String, Option<
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
- let names = batch
- .column(1)
- .as_any()
- .downcast_ref::<StringArray>()
- .unwrap();
+ let names = batch.column(1);
let pics = batch
.column(2)
.as_any()
@@ -70,7 +66,11 @@ fn collect_id_name_picture(batches: &[RecordBatch]) ->
Vec<(i32, String, Option<
} else {
Some(pics.value(i).to_vec())
};
- rows.push((ids.value(i), names.value(i).to_string(), pic));
+ rows.push((
+ ids.value(i),
+ string_value(names.as_ref(), i).to_string(),
+ pic,
+ ));
}
}
rows.sort_by_key(|(id, _, _)| *id);
@@ -93,13 +93,9 @@ fn collect_id_name(batches: &[RecordBatch]) -> Vec<(i32,
String)> {
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
- let names = batch
- .column(1)
- .as_any()
- .downcast_ref::<StringArray>()
- .unwrap();
+ let names = batch.column(1);
for i in 0..batch.num_rows() {
- rows.push((ids.value(i), names.value(i).to_string()));
+ rows.push((ids.value(i), string_value(names.as_ref(),
i).to_string()));
}
}
rows.sort_by_key(|(id, _)| *id);
diff --git a/crates/integrations/datafusion/tests/common/mod.rs
b/crates/integrations/datafusion/tests/common/mod.rs
index c01201d1..23d9fb9c 100644
--- a/crates/integrations/datafusion/tests/common/mod.rs
+++ b/crates/integrations/datafusion/tests/common/mod.rs
@@ -19,13 +19,26 @@
use std::sync::Arc;
-use datafusion::arrow::array::{Int32Array, StringArray};
+use datafusion::arrow::array::{Int32Array, LargeStringArray, StringArray,
StringViewArray};
use paimon::{CatalogOptions, FileSystemCatalog, Options};
use paimon_datafusion::SQLContext;
use tempfile::TempDir;
use arrow_array::{Array, RecordBatch, UInt64Array};
+#[allow(dead_code)]
+pub fn string_value(array: &dyn Array, row: usize) -> &str {
+ if let Some(array) = array.as_any().downcast_ref::<StringArray>() {
+ array.value(row)
+ } else if let Some(array) =
array.as_any().downcast_ref::<LargeStringArray>() {
+ array.value(row)
+ } else if let Some(array) =
array.as_any().downcast_ref::<StringViewArray>() {
+ array.value(row)
+ } else {
+ panic!("expected a string array, got {}", array.data_type())
+ }
+}
+
pub fn create_test_env() -> (TempDir, Arc<FileSystemCatalog>) {
let temp_dir = TempDir::new().expect("Failed to create temp dir");
let warehouse = format!("file://{}", temp_dir.path().display());
@@ -76,12 +89,9 @@ pub fn collect_id_name_from_batches_in_order(batches:
&[RecordBatch]) -> Vec<(i3
.column_by_name("id")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.expect("id column");
- let names = batch
- .column_by_name("name")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .expect("name column");
+ let names = batch.column_by_name("name").expect("name column");
for i in 0..batch.num_rows() {
- rows.push((ids.value(i), names.value(i).to_string()));
+ rows.push((ids.value(i), string_value(names.as_ref(),
i).to_string()));
}
}
rows
@@ -168,13 +178,13 @@ pub fn collect_int_int_str(batches: &[RecordBatch]) ->
Vec<(i32, i32, String)> {
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
- let col2 = batch
- .column(2)
- .as_any()
- .downcast_ref::<StringArray>()
- .unwrap();
+ let col2 = batch.column(2);
for i in 0..batch.num_rows() {
- rows.push((col0.value(i), col1.value(i),
col2.value(i).to_string()));
+ rows.push((
+ col0.value(i),
+ col1.value(i),
+ string_value(col2.as_ref(), i).to_string(),
+ ));
}
}
rows.sort_by_key(|r| (r.0, r.1));
@@ -191,13 +201,9 @@ pub fn collect_int_str(batches: &[RecordBatch]) ->
Vec<(i32, String)> {
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
- let col1 = batch
- .column(1)
- .as_any()
- .downcast_ref::<StringArray>()
- .unwrap();
+ let col1 = batch.column(1);
for i in 0..batch.num_rows() {
- rows.push((col0.value(i), col1.value(i).to_string()));
+ rows.push((col0.value(i), string_value(col1.as_ref(),
i).to_string()));
}
}
rows.sort_by_key(|r| r.0);
@@ -242,18 +248,18 @@ pub fn collect_int_str_int(batches: &[RecordBatch]) ->
Vec<(i32, String, i32)> {
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
- let col1 = batch
- .column(1)
- .as_any()
- .downcast_ref::<StringArray>()
- .unwrap();
+ let col1 = batch.column(1);
let col2 = batch
.column(2)
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
for i in 0..batch.num_rows() {
- rows.push((col0.value(i), col1.value(i).to_string(),
col2.value(i)));
+ rows.push((
+ col0.value(i),
+ string_value(col1.as_ref(), i).to_string(),
+ col2.value(i),
+ ));
}
}
rows.sort_by_key(|r| r.0);
diff --git a/crates/integrations/datafusion/tests/dynamic_bucket_tables.rs
b/crates/integrations/datafusion/tests/dynamic_bucket_tables.rs
index 97d7052b..34aea68f 100644
--- a/crates/integrations/datafusion/tests/dynamic_bucket_tables.rs
+++ b/crates/integrations/datafusion/tests/dynamic_bucket_tables.rs
@@ -21,8 +21,9 @@ mod common;
use common::{
collect_id_name, collect_id_value, create_sql_context, create_test_env,
setup_sql_context,
+ string_value,
};
-use datafusion::arrow::array::{Array, Int32Array, StringArray};
+use datafusion::arrow::array::{Array, Int32Array};
use paimon::catalog::Identifier;
use paimon::spec::{IndexManifest, IndexManifestEntry};
use paimon::{Catalog, CatalogOptions, DataSplit, FileSystemCatalog, Options,
SnapshotManager};
@@ -97,10 +98,7 @@ async fn collect_partial_update_rows(
.column_by_name("v_int")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
- let strs = batch
- .column_by_name("v_str")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let strs = batch.column_by_name("v_str").unwrap();
for i in 0..batch.num_rows() {
rows.push((
ids.value(i),
@@ -112,7 +110,7 @@ async fn collect_partial_update_rows(
if strs.is_null(i) {
None
} else {
- Some(strs.value(i).to_string())
+ Some(string_value(strs.as_ref(), i).to_string())
},
));
}
@@ -136,14 +134,8 @@ async fn collect_aggregation_rows(
.column_by_name("amount")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
- let tags = batch
- .column_by_name("tag")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
- let notes = batch
- .column_by_name("note")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let tags = batch.column_by_name("tag").unwrap();
+ let notes = batch.column_by_name("note").unwrap();
for i in 0..batch.num_rows() {
rows.push((
ids.value(i),
@@ -155,12 +147,12 @@ async fn collect_aggregation_rows(
if tags.is_null(i) {
None
} else {
- Some(tags.value(i).to_string())
+ Some(string_value(tags.as_ref(), i).to_string())
},
if notes.is_null(i) {
None
} else {
- Some(notes.value(i).to_string())
+ Some(string_value(notes.as_ref(), i).to_string())
},
));
}
@@ -573,10 +565,7 @@ async fn test_pk_dynamic_bucket_partitioned() {
let mut rows = Vec::new();
for batch in &batches {
- let dts = batch
- .column_by_name("dt")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let dts = batch.column_by_name("dt").unwrap();
let ids = batch
.column_by_name("id")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
@@ -586,7 +575,11 @@ async fn test_pk_dynamic_bucket_partitioned() {
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
for i in 0..batch.num_rows() {
- rows.push((dts.value(i).to_string(), ids.value(i), vals.value(i)));
+ rows.push((
+ string_value(dts.as_ref(), i).to_string(),
+ ids.value(i),
+ vals.value(i),
+ ));
}
}
rows.sort_by(|a, b| a.0.cmp(&b.0).then(a.1.cmp(&b.1)));
@@ -655,10 +648,7 @@ async fn
test_pk_dynamic_bucket_partitioned_partial_update() {
let mut rows = Vec::new();
for batch in &batches {
- let dts = batch
- .column_by_name("dt")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let dts = batch.column_by_name("dt").unwrap();
let ids = batch
.column_by_name("id")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
@@ -667,13 +657,10 @@ async fn
test_pk_dynamic_bucket_partitioned_partial_update() {
.column_by_name("v_int")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
- let strs = batch
- .column_by_name("v_str")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let strs = batch.column_by_name("v_str").unwrap();
for i in 0..batch.num_rows() {
rows.push((
- dts.value(i).to_string(),
+ string_value(dts.as_ref(), i).to_string(),
ids.value(i),
if ints.is_null(i) {
None
@@ -683,7 +670,7 @@ async fn
test_pk_dynamic_bucket_partitioned_partial_update() {
if strs.is_null(i) {
None
} else {
- Some(strs.value(i).to_string())
+ Some(string_value(strs.as_ref(), i).to_string())
},
));
}
@@ -1066,12 +1053,9 @@ async fn
test_read_spark_dynamic_bucket_and_compare_index() {
.column_by_name("id")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
- let names = batch
- .column_by_name("name")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let names = batch.column_by_name("name").unwrap();
for i in 0..batch.num_rows() {
- all_rows.push((ids.value(i), names.value(i).to_string()));
+ all_rows.push((ids.value(i), string_value(names.as_ref(),
i).to_string()));
}
}
all_rows.sort_by_key(|(id, _)| *id);
diff --git a/crates/integrations/datafusion/tests/merge_into_tests.rs
b/crates/integrations/datafusion/tests/merge_into_tests.rs
index f3e93582..87fb50be 100644
--- a/crates/integrations/datafusion/tests/merge_into_tests.rs
+++ b/crates/integrations/datafusion/tests/merge_into_tests.rs
@@ -21,9 +21,12 @@
//! join on `_ROW_ID`, and error path validation.
//! Reference: Java Paimon's `RowTrackingTestBase`.
+mod common;
+
use std::sync::Arc;
-use arrow_array::{Int32Array, Int64Array, StringArray};
+use arrow_array::{Int32Array, Int64Array};
+use common::string_value;
use paimon::catalog::Identifier;
use paimon::table::SnapshotManager;
use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options};
@@ -80,18 +83,18 @@ async fn collect_rows_3col(sql_context: &SQLContext, sql:
&str) -> Vec<(i32, Str
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
- let names = batch
- .column(1)
- .as_any()
- .downcast_ref::<StringArray>()
- .unwrap();
+ let names = batch.column(1);
let values = batch
.column(2)
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
for i in 0..batch.num_rows() {
- rows.push((ids.value(i), names.value(i).to_string(),
values.value(i)));
+ rows.push((
+ ids.value(i),
+ string_value(names.as_ref(), i).to_string(),
+ values.value(i),
+ ));
}
}
rows
@@ -1241,26 +1244,18 @@ async fn
test_merge_insert_reordered_columns_on_partitioned_table() {
let mut rows = Vec::new();
for batch in &batches {
- let dts = batch
- .column(0)
- .as_any()
- .downcast_ref::<StringArray>()
- .unwrap();
+ let dts = batch.column(0);
let ids = batch
.column(1)
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
- let names = batch
- .column(2)
- .as_any()
- .downcast_ref::<StringArray>()
- .unwrap();
+ let names = batch.column(2);
for i in 0..batch.num_rows() {
rows.push((
- dts.value(i).to_string(),
+ string_value(dts.as_ref(), i).to_string(),
ids.value(i),
- names.value(i).to_string(),
+ string_value(names.as_ref(), i).to_string(),
));
}
}
diff --git a/crates/integrations/datafusion/tests/mosaic_tables.rs
b/crates/integrations/datafusion/tests/mosaic_tables.rs
index 426515d1..7ada5e3c 100644
--- a/crates/integrations/datafusion/tests/mosaic_tables.rs
+++ b/crates/integrations/datafusion/tests/mosaic_tables.rs
@@ -17,14 +17,18 @@
//! Mosaic file format read compatibility tests.
+mod common;
+
use std::path::Path;
use std::sync::Arc;
-use datafusion::arrow::array::{Int32Array, Int64Array, StringArray};
+use datafusion::arrow::array::{Int32Array, Int64Array};
use datafusion::arrow::record_batch::RecordBatch;
use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options};
use paimon_datafusion::SQLContext;
+use common::string_value;
+
const FIXTURE_TABLE: &str = "test_mosaic_read";
fn extract_test_warehouse() -> (tempfile::TempDir, String) {
@@ -74,10 +78,7 @@ fn collect_id_name_score(batches: &[RecordBatch]) ->
Vec<(i32, String, i64)> {
.column_by_name("id")
.and_then(|column| column.as_any().downcast_ref::<Int32Array>())
.expect("id column");
- let names = batch
- .column_by_name("name")
- .and_then(|column| column.as_any().downcast_ref::<StringArray>())
- .expect("name column");
+ let names = batch.column_by_name("name").expect("name column");
let scores = batch
.column_by_name("score")
.and_then(|column| column.as_any().downcast_ref::<Int64Array>())
@@ -86,7 +87,7 @@ fn collect_id_name_score(batches: &[RecordBatch]) ->
Vec<(i32, String, i64)> {
for row in 0..batch.num_rows() {
rows.push((
ids.value(row),
- names.value(row).to_string(),
+ string_value(names.as_ref(), row).to_string(),
scores.value(row),
));
}
@@ -97,11 +98,7 @@ fn collect_id_name_score(batches: &[RecordBatch]) ->
Vec<(i32, String, i64)> {
fn collect_name_id(batches: &[RecordBatch]) -> Vec<(String, i32)> {
let mut rows = Vec::new();
for batch in batches {
- let names = batch
- .column(0)
- .as_any()
- .downcast_ref::<StringArray>()
- .expect("first column should be name");
+ let names = batch.column(0);
let ids = batch
.column(1)
.as_any()
@@ -109,7 +106,10 @@ fn collect_name_id(batches: &[RecordBatch]) ->
Vec<(String, i32)> {
.expect("second column should be id");
for row in 0..batch.num_rows() {
- rows.push((names.value(row).to_string(), ids.value(row)));
+ rows.push((
+ string_value(names.as_ref(), row).to_string(),
+ ids.value(row),
+ ));
}
}
rows
diff --git a/crates/integrations/datafusion/tests/pk_tables.rs
b/crates/integrations/datafusion/tests/pk_tables.rs
index 66d5de44..4b0840c0 100644
--- a/crates/integrations/datafusion/tests/pk_tables.rs
+++ b/crates/integrations/datafusion/tests/pk_tables.rs
@@ -30,9 +30,9 @@ mod common;
use common::{
collect_id_name, collect_id_value, collect_int_int_str,
create_sql_context, create_test_env,
- row_count, setup_sql_context,
+ row_count, setup_sql_context, string_value,
};
-use datafusion::arrow::array::{Array, Int32Array, Int64Array, StringArray};
+use datafusion::arrow::array::{Array, Int32Array, Int64Array};
use paimon::catalog::Identifier;
use paimon::Catalog;
@@ -146,10 +146,7 @@ async fn test_pk_partial_update_fixed_bucket_e2e() {
.column_by_name("v_int")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
- let strs = batch
- .column_by_name("v_str")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let strs = batch.column_by_name("v_str").unwrap();
for i in 0..batch.num_rows() {
rows.push((
ids.value(i),
@@ -161,7 +158,7 @@ async fn test_pk_partial_update_fixed_bucket_e2e() {
if strs.is_null(i) {
None
} else {
- Some(strs.value(i).to_string())
+ Some(string_value(strs.as_ref(), i).to_string())
},
));
}
@@ -474,23 +471,17 @@ async fn test_pk_partitioned_dedup_across_commits() {
let mut rows = Vec::new();
for batch in &batches {
- let dts = batch
- .column_by_name("dt")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let dts = batch.column_by_name("dt").unwrap();
let ids = batch
.column_by_name("id")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
- let names = batch
- .column_by_name("name")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let names = batch.column_by_name("name").unwrap();
for i in 0..batch.num_rows() {
rows.push((
- dts.value(i).to_string(),
+ string_value(dts.as_ref(), i).to_string(),
ids.value(i),
- names.value(i).to_string(),
+ string_value(names.as_ref(), i).to_string(),
));
}
}
@@ -646,13 +637,9 @@ async fn test_pk_column_projection() {
let mut names = Vec::new();
for batch in &batches {
- let arr = batch
- .column(0)
- .as_any()
- .downcast_ref::<StringArray>()
- .unwrap();
+ let arr = batch.column(0);
for i in 0..batch.num_rows() {
- names.push(arr.value(i).to_string());
+ names.push(string_value(arr.as_ref(), i).to_string());
}
}
names.sort();
@@ -766,23 +753,17 @@ async fn test_pk_insert_overwrite_partitioned() {
let mut rows = Vec::new();
for batch in &batches {
- let dts = batch
- .column_by_name("dt")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let dts = batch.column_by_name("dt").unwrap();
let ids = batch
.column_by_name("id")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
- let names = batch
- .column_by_name("name")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let names = batch.column_by_name("name").unwrap();
for i in 0..batch.num_rows() {
rows.push((
- dts.value(i).to_string(),
+ string_value(dts.as_ref(), i).to_string(),
ids.value(i),
- names.value(i).to_string(),
+ string_value(names.as_ref(), i).to_string(),
));
}
}
@@ -848,23 +829,17 @@ async fn test_pk_insert_overwrite_with_partition_clause()
{
let mut rows = Vec::new();
for batch in &batches {
- let dts = batch
- .column_by_name("dt")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let dts = batch.column_by_name("dt").unwrap();
let ids = batch
.column_by_name("id")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
- let names = batch
- .column_by_name("name")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let names = batch.column_by_name("name").unwrap();
for i in 0..batch.num_rows() {
rows.push((
- dts.value(i).to_string(),
+ string_value(dts.as_ref(), i).to_string(),
ids.value(i),
- names.value(i).to_string(),
+ string_value(names.as_ref(), i).to_string(),
));
}
}
@@ -934,28 +909,19 @@ async fn
test_pk_insert_overwrite_partial_partition_clause() {
let mut rows = Vec::new();
for batch in &batches {
- let dts = batch
- .column_by_name("dt")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
- let regions = batch
- .column_by_name("region")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let dts = batch.column_by_name("dt").unwrap();
+ let regions = batch.column_by_name("region").unwrap();
let ids = batch
.column_by_name("id")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
- let names = batch
- .column_by_name("name")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let names = batch.column_by_name("name").unwrap();
for i in 0..batch.num_rows() {
rows.push((
- dts.value(i).to_string(),
- regions.value(i).to_string(),
+ string_value(dts.as_ref(), i).to_string(),
+ string_value(regions.as_ref(), i).to_string(),
ids.value(i),
- names.value(i).to_string(),
+ string_value(names.as_ref(), i).to_string(),
));
}
}
@@ -1032,23 +998,17 @@ async fn test_pk_insert_overwrite_partition_truncate() {
let mut rows = Vec::new();
for batch in &batches {
- let dts = batch
- .column_by_name("dt")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let dts = batch.column_by_name("dt").unwrap();
let ids = batch
.column_by_name("id")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
- let names = batch
- .column_by_name("name")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let names = batch.column_by_name("name").unwrap();
for i in 0..batch.num_rows() {
rows.push((
- dts.value(i).to_string(),
+ string_value(dts.as_ref(), i).to_string(),
ids.value(i),
- names.value(i).to_string(),
+ string_value(names.as_ref(), i).to_string(),
));
}
}
@@ -1142,23 +1102,17 @@ async fn
test_pk_insert_overwrite_dynamic_partition_preserves_other_partitions()
let mut rows = Vec::new();
for batch in &batches {
- let dts = batch
- .column_by_name("dt")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let dts = batch.column_by_name("dt").unwrap();
let ids = batch
.column_by_name("id")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
- let names = batch
- .column_by_name("name")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let names = batch.column_by_name("name").unwrap();
for i in 0..batch.num_rows() {
rows.push((
- dts.value(i).to_string(),
+ string_value(dts.as_ref(), i).to_string(),
ids.value(i),
- names.value(i).to_string(),
+ string_value(names.as_ref(), i).to_string(),
));
}
}
@@ -1254,23 +1208,17 @@ async fn
test_pk_insert_overwrite_with_after_columns_reorder() {
let mut rows = Vec::new();
for batch in &batches {
- let dts = batch
- .column_by_name("dt")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let dts = batch.column_by_name("dt").unwrap();
let ids = batch
.column_by_name("id")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
- let names = batch
- .column_by_name("name")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let names = batch.column_by_name("name").unwrap();
for i in 0..batch.num_rows() {
rows.push((
- dts.value(i).to_string(),
+ string_value(dts.as_ref(), i).to_string(),
ids.value(i),
- names.value(i).to_string(),
+ string_value(names.as_ref(), i).to_string(),
));
}
}
@@ -1332,10 +1280,7 @@ async fn test_pk_composite_key() {
let mut rows = Vec::new();
for batch in &batches {
- let regions = batch
- .column_by_name("region")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let regions = batch.column_by_name("region").unwrap();
let ids = batch
.column_by_name("id")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
@@ -1345,7 +1290,11 @@ async fn test_pk_composite_key() {
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
for i in 0..batch.num_rows() {
- rows.push((regions.value(i).to_string(), ids.value(i),
vals.value(i)));
+ rows.push((
+ string_value(regions.as_ref(), i).to_string(),
+ ids.value(i),
+ vals.value(i),
+ ));
}
}
@@ -1492,10 +1441,7 @@ async fn test_pk_partitioned_multi_bucket() {
let mut rows = Vec::new();
for batch in &batches {
- let dts = batch
- .column_by_name("dt")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let dts = batch.column_by_name("dt").unwrap();
let ids = batch
.column_by_name("id")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
@@ -1505,7 +1451,11 @@ async fn test_pk_partitioned_multi_bucket() {
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
for i in 0..batch.num_rows() {
- rows.push((dts.value(i).to_string(), ids.value(i), vals.value(i)));
+ rows.push((
+ string_value(dts.as_ref(), i).to_string(),
+ ids.value(i),
+ vals.value(i),
+ ));
}
}
@@ -1606,16 +1556,13 @@ async fn test_pk_string_key() {
let mut rows = Vec::new();
for batch in &batches {
- let codes = batch
- .column_by_name("code")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
- let names = batch
- .column_by_name("name")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let codes = batch.column_by_name("code").unwrap();
+ let names = batch.column_by_name("name").unwrap();
for i in 0..batch.num_rows() {
- rows.push((codes.value(i).to_string(),
names.value(i).to_string()));
+ rows.push((
+ string_value(codes.as_ref(), i).to_string(),
+ string_value(names.as_ref(), i).to_string(),
+ ));
}
}
@@ -1680,10 +1627,7 @@ async fn test_pk_multiple_value_columns() {
.column_by_name("col_a")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
- let bs = batch
- .column_by_name("col_b")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let bs = batch.column_by_name("col_b").unwrap();
let cs = batch
.column_by_name("col_c")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
@@ -1692,7 +1636,7 @@ async fn test_pk_multiple_value_columns() {
rows.push((
ids.value(i),
as_.value(i),
- bs.value(i).to_string(),
+ string_value(bs.as_ref(), i).to_string(),
cs.value(i),
));
}
@@ -2343,10 +2287,7 @@ async fn
test_pk_aggregation_sum_and_listagg_fixed_multi_bucket_e2e() {
.column_by_name("amount")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
- let tags = batch
- .column_by_name("tag")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let tags = batch.column_by_name("tag").unwrap();
for i in 0..batch.num_rows() {
rows.push((
ids.value(i),
@@ -2358,7 +2299,7 @@ async fn
test_pk_aggregation_sum_and_listagg_fixed_multi_bucket_e2e() {
if tags.is_null(i) {
None
} else {
- Some(tags.value(i).to_string())
+ Some(string_value(tags.as_ref(), i).to_string())
},
));
}
@@ -2437,13 +2378,10 @@ async fn test_pk_aggregation_default_function() {
.column_by_name("a")
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
- let b = batch
- .column_by_name("b")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let b = batch.column_by_name("b").unwrap();
assert_eq!(id.value(0), 1);
assert_eq!(a.value(0), 99); // latest non-null int across the three commits
- assert_eq!(b.value(0), "new"); // latest non-null string
+ assert_eq!(string_value(b.as_ref(), 0), "new"); // latest non-null string
}
/// Mixed aggregators in a single table: sum / max / bool_or /
first_non_null_value.
@@ -2504,14 +2442,11 @@ async fn test_pk_aggregation_mixed_aggregators() {
.column_by_name("ok")
.and_then(|c| c.as_any().downcast_ref::<BooleanArray>())
.unwrap();
- let first_seen = batch
- .column_by_name("first_seen")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .unwrap();
+ let first_seen = batch.column_by_name("first_seen").unwrap();
assert_eq!(total.value(0), 18); // 10 + 5 + 3
assert_eq!(peak.value(0), 8); // max(5, 8, 7)
assert!(ok.value(0)); // bool_or = true if any is true
- assert_eq!(first_seen.value(0), "a"); // first non-null wins
+ assert_eq!(string_value(first_seen.as_ref(), 0), "a"); // first non-null
wins
}
/// `sequence.field` forces the named column to `last_value`, even when a
diff --git a/crates/integrations/datafusion/tests/read_tables.rs
b/crates/integrations/datafusion/tests/read_tables.rs
index 5ee87099..66d75d5b 100644
--- a/crates/integrations/datafusion/tests/read_tables.rs
+++ b/crates/integrations/datafusion/tests/read_tables.rs
@@ -20,7 +20,8 @@ use std::sync::Arc;
mod common;
-use datafusion::arrow::array::{Array, Int32Array, StringArray};
+use common::string_value;
+use datafusion::arrow::array::{Array, Int32Array};
use datafusion::arrow::record_batch::RecordBatch;
use datafusion::arrow::util::display::array_value_to_string;
use datafusion::catalog::CatalogProvider;
@@ -164,13 +165,12 @@ fn extract_id_name_rows(
.expect("Expected Int32Array for id column");
let name_array = batch
.column_by_name("name")
- .and_then(|column| column.as_any().downcast_ref::<StringArray>())
- .expect("Expected StringArray for name column");
+ .expect("Expected string array for name column");
for row_index in 0..batch.num_rows() {
rows.push((
id_array.value(row_index),
- name_array.value(row_index).to_string(),
+ string_value(name_array.as_ref(), row_index).to_string(),
));
}
}
@@ -1081,20 +1081,22 @@ async fn test_data_evolution_drop_column_null_fill() {
.expect("Expected Int32Array for id");
let name_array = batch
.column_by_name("name")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .expect("Expected StringArray for name");
+ .expect("Expected string array for name");
let extra_array = batch
.column_by_name("extra")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .expect("Expected StringArray for extra");
+ .expect("Expected string array for extra");
for i in 0..batch.num_rows() {
let extra = if extra_array.is_null(i) {
None
} else {
- Some(extra_array.value(i).to_string())
+ Some(string_value(extra_array.as_ref(), i).to_string())
};
- rows.push((id_array.value(i), name_array.value(i).to_string(),
extra));
+ rows.push((
+ id_array.value(i),
+ string_value(name_array.as_ref(), i).to_string(),
+ extra,
+ ));
}
}
rows.sort_by_key(|(id, _, _)| *id);
@@ -1447,12 +1449,14 @@ async fn
test_case_insensitive_column_not_supported_via_sql() {
mod fulltext_tests {
use std::sync::Arc;
- use datafusion::arrow::array::{Int32Array, StringArray};
+ use datafusion::arrow::array::Int32Array;
use paimon::catalog::Identifier;
use paimon::table::BranchManager;
use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options};
use paimon_datafusion::{register_full_text_search, SQLContext};
+ use super::common::string_value;
+
/// Extract the bundled tar.gz into a temp dir and return (tempdir,
warehouse_path).
fn extract_test_warehouse() -> (tempfile::TempDir, String) {
let archive_path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
@@ -1496,10 +1500,12 @@ mod fulltext_tests {
.expect("Expected Int32Array for id");
let content_array = batch
.column_by_name("content")
- .and_then(|c| c.as_any().downcast_ref::<StringArray>())
- .expect("Expected StringArray for content");
+ .expect("Expected string array for content");
for i in 0..batch.num_rows() {
- rows.push((id_array.value(i),
content_array.value(i).to_string()));
+ rows.push((
+ id_array.value(i),
+ string_value(content_array.as_ref(), i).to_string(),
+ ));
}
}
rows.sort_by_key(|(id, _)| *id);
@@ -1619,6 +1625,8 @@ mod vector_search_tests {
use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options};
use paimon_datafusion::{register_vector_search, SQLContext};
+ use super::common::string_value;
+
fn extract_test_warehouse(archive_name: &str) -> (tempfile::TempDir,
String) {
let archive_path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
.join("testdata")
@@ -1791,11 +1799,7 @@ mod vector_search_tests {
for batch in batches {
let index_type_array = batch
.column_by_name("index_type")
- .and_then(|c| {
- c.as_any()
-
.downcast_ref::<datafusion::arrow::array::StringArray>()
- })
- .expect("Expected StringArray for index_type");
+ .expect("Expected string array for index_type");
let row_count_array = batch
.column_by_name("row_count")
.and_then(|c| {
@@ -1819,19 +1823,15 @@ mod vector_search_tests {
.expect("Expected Int64Array for row_range_end");
let index_field_name_array = batch
.column_by_name("index_field_name")
- .and_then(|c| {
- c.as_any()
-
.downcast_ref::<datafusion::arrow::array::StringArray>()
- })
- .expect("Expected StringArray for index_field_name");
+ .expect("Expected string array for index_field_name");
for row_index in 0..batch.num_rows() {
rows.push((
- index_type_array.value(row_index).to_string(),
+ string_value(index_type_array.as_ref(),
row_index).to_string(),
row_count_array.value(row_index),
row_range_start_array.value(row_index),
row_range_end_array.value(row_index),
- index_field_name_array.value(row_index).to_string(),
+ string_value(index_field_name_array.as_ref(),
row_index).to_string(),
));
}
}
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index 1573342c..c15ead0d 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -1964,6 +1964,92 @@ mod tests {
assert_eq!(mask, expected);
}
+ #[test]
+ fn test_evaluate_string_view_comparison_families() {
+ use crate::spec::VarCharType;
+ use arrow_array::{ArrayRef, BooleanArray, StringViewArray};
+
+ let column: ArrayRef = Arc::new(StringViewArray::from(vec![
+ Some("a"),
+ Some("b"),
+ Some("c"),
+ None,
+ ]));
+ let data_type = DataType::VarChar(VarCharType::default());
+ let cases = [
+ (
+ super::PredicateOperator::Eq,
+ vec![Datum::String("b".to_string())],
+ vec![false, true, false, false],
+ ),
+ (
+ super::PredicateOperator::NotEq,
+ vec![Datum::String("b".to_string())],
+ vec![true, false, true, false],
+ ),
+ (
+ super::PredicateOperator::Lt,
+ vec![Datum::String("b".to_string())],
+ vec![true, false, false, false],
+ ),
+ (
+ super::PredicateOperator::LtEq,
+ vec![Datum::String("b".to_string())],
+ vec![true, true, false, false],
+ ),
+ (
+ super::PredicateOperator::Gt,
+ vec![Datum::String("b".to_string())],
+ vec![false, false, true, false],
+ ),
+ (
+ super::PredicateOperator::GtEq,
+ vec![Datum::String("b".to_string())],
+ vec![false, true, true, false],
+ ),
+ (
+ super::PredicateOperator::In,
+ vec![
+ Datum::String("a".to_string()),
+ Datum::String("c".to_string()),
+ ],
+ vec![true, false, true, false],
+ ),
+ (
+ super::PredicateOperator::NotIn,
+ vec![
+ Datum::String("a".to_string()),
+ Datum::String("c".to_string()),
+ ],
+ vec![false, true, false, false],
+ ),
+ (
+ super::PredicateOperator::Between,
+ vec![
+ Datum::String("a".to_string()),
+ Datum::String("b".to_string()),
+ ],
+ vec![true, true, false, false],
+ ),
+ (
+ super::PredicateOperator::NotBetween,
+ vec![
+ Datum::String("a".to_string()),
+ Datum::String("b".to_string()),
+ ],
+ vec![false, false, true, false],
+ ),
+ ];
+
+ for (op, literals, expected) in cases {
+ let mask = crate::arrow::residual::evaluate_exact_leaf_predicate(
+ &column, &data_type, op, &literals,
+ )
+ .unwrap_or_else(|error| panic!("{op:?} should evaluate: {error}"));
+ assert_eq!(mask, BooleanArray::from(expected), "operator {op:?}");
+ }
+ }
+
#[test]
fn test_evaluate_like_pattern_with_underscore_and_percent() {
use arrow_array::StringArray;
diff --git a/crates/paimon/src/arrow/residual.rs
b/crates/paimon/src/arrow/residual.rs
index b767d2aa..1a6ff5a1 100644
--- a/crates/paimon/src/arrow/residual.rs
+++ b/crates/paimon/src/arrow/residual.rs
@@ -571,6 +571,8 @@ fn evaluate_column_predicate(
scalar: &Scalar<ArrayRef>,
op: PredicateOperator,
) -> Result<BooleanArray, ArrowError> {
+ let scalar = string_scalar_for_column(scalar, column.data_type())?;
+
// Binary ordering must match Paimon's Datum::Bytes semantics (Java
signed-byte
// order, 0xFF < 0x00), which Arrow's unsigned byte comparison does not.
Route
// ordering ops on Binary/VarBinary columns through the signed comparator.
@@ -585,28 +587,25 @@ fn evaluate_column_predicate(
| PredicateOperator::Gt
| PredicateOperator::GtEq
) {
- return evaluate_binary_ordering_predicate(column, scalar, op);
+ return evaluate_binary_ordering_predicate(column, &scalar, op);
}
match op {
- PredicateOperator::Eq => arrow_eq(column, scalar),
- PredicateOperator::NotEq => arrow_neq(column, scalar),
- PredicateOperator::Lt => arrow_lt(column, scalar),
- PredicateOperator::LtEq => arrow_lt_eq(column, scalar),
- PredicateOperator::Gt => arrow_gt(column, scalar),
- PredicateOperator::GtEq => arrow_gt_eq(column, scalar),
+ PredicateOperator::Eq => arrow_eq(column, &scalar),
+ PredicateOperator::NotEq => arrow_neq(column, &scalar),
+ PredicateOperator::Lt => arrow_lt(column, &scalar),
+ PredicateOperator::LtEq => arrow_lt_eq(column, &scalar),
+ PredicateOperator::Gt => arrow_gt(column, &scalar),
+ PredicateOperator::GtEq => arrow_gt_eq(column, &scalar),
PredicateOperator::StartsWith
| PredicateOperator::EndsWith
| PredicateOperator::Contains
- | PredicateOperator::Like => {
- let pattern = pattern_scalar_for_string_kernel(scalar,
column.data_type())?;
- match op {
- PredicateOperator::StartsWith => arrow_starts_with(column,
&pattern),
- PredicateOperator::EndsWith => arrow_ends_with(column,
&pattern),
- PredicateOperator::Contains => arrow_contains(column,
&pattern),
- PredicateOperator::Like => arrow_like(column, &pattern),
- _ => unreachable!(),
- }
- }
+ | PredicateOperator::Like => match op {
+ PredicateOperator::StartsWith => arrow_starts_with(column,
&scalar),
+ PredicateOperator::EndsWith => arrow_ends_with(column, &scalar),
+ PredicateOperator::Contains => arrow_contains(column, &scalar),
+ PredicateOperator::Like => arrow_like(column, &scalar),
+ _ => unreachable!(),
+ },
PredicateOperator::IsNull
| PredicateOperator::IsNotNull
| PredicateOperator::In
@@ -665,11 +664,11 @@ fn evaluate_binary_ordering_predicate(
Ok(mask)
}
-/// `arrow_string::like::*` kernels reject mismatched string types — Utf8
column
-/// against Utf8 pattern is fine, but a LargeUtf8 / Utf8View column needs a
-/// pattern of the same flavour. The shared scalar built upstream is always
-/// `StringArray` (Utf8); promote it to match the column when needed.
-fn pattern_scalar_for_string_kernel(
+/// Arrow comparison and pattern kernels reject mismatched string types. The
+/// shared scalar built from Paimon's logical Char/VarChar type is Utf8, while
a
+/// decoded file column may be Utf8, LargeUtf8, or Utf8View. Promote the scalar
+/// to the actual column representation before invoking any string kernel.
+fn string_scalar_for_column(
scalar: &Scalar<ArrayRef>,
column_type: &arrow_schema::DataType,
) -> Result<Scalar<ArrayRef>, ArrowError> {
