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 604ae83 feat: support case-insensitive column matching on reads
(read-time parameter) (#496)
604ae83 is described below
commit 604ae83d269c240dbaac394bfd4c7ab27ac25d6c
Author: Junrui Lee <[email protected]>
AuthorDate: Mon Jul 13 13:13:48 2026 +0800
feat: support case-insensitive column matching on reads (read-time
parameter) (#496)
Add ReadBuilder::with_case_sensitive(bool) (default true) and
PredicateBuilder::new_with_case_sensitive(fields, bool), mirroring Java's
RowType.getFieldIndex(name, caseSensitive). When disabled, projection and
predicate column names resolve by ASCII case-folding, error on ambiguity,
and the canonical schema name is stored in the predicate leaf.
Projection resolution is deferred to scan/read build time so
with_projection and with_case_sensitive are order-independent;
with_projection
does best-effort early validation, rejecting only columns that cannot match
under any case sensitivity. Default stays case-sensitive.
---
bindings/c/src/table.rs | 374 ++++++++++++++++--
bindings/c/src/types.rs | 1 +
.../python/python/pypaimon_rust/datafusion.pyi | 11 +
bindings/python/src/predicate.rs | 150 +++++--
bindings/python/src/read.rs | 38 +-
crates/integration_tests/tests/read_tables.rs | 10 +-
.../integrations/datafusion/src/filter_pushdown.rs | 154 +++++++-
.../datafusion/src/full_text_search.rs | 1 +
.../integrations/datafusion/src/hybrid_search.rs | 1 +
.../datafusion/src/physical_plan/scan.rs | 10 +
crates/integrations/datafusion/src/table/mod.rs | 26 +-
.../datafusion/src/variant_pushdown.rs | 11 +-
.../integrations/datafusion/src/vector_search.rs | 1 +
.../integrations/datafusion/tests/read_tables.rs | 43 ++
crates/paimon/src/spec/predicate.rs | 89 ++++-
crates/paimon/src/table/format_read_builder.rs | 46 ++-
crates/paimon/src/table/read_builder.rs | 438 ++++++++++++++++++++-
17 files changed, 1277 insertions(+), 127 deletions(-)
diff --git a/bindings/c/src/table.rs b/bindings/c/src/table.rs
index 5c1622d..0940c5b 100644
--- a/bindings/c/src/table.rs
+++ b/bindings/c/src/table.rs
@@ -85,6 +85,7 @@ pub unsafe extern "C" fn paimon_table_new_read_builder(
table: table_ref.clone(),
projected_columns: None,
filter: None,
+ case_sensitive: true,
};
paimon_result_read_builder {
read_builder: box_read_builder_state(state),
@@ -111,8 +112,13 @@ pub unsafe extern "C" fn paimon_read_builder_free(rb: *mut
paimon_read_builder)
/// Set column projection for a ReadBuilder.
///
/// The `columns` parameter is a null-terminated array of null-terminated C
strings.
-/// Output order follows the caller-specified order. Unknown or duplicate names
-/// are validated immediately; an empty list is a valid zero-column projection.
+/// Output order follows the caller-specified order. An empty list is a valid
+/// zero-column projection. An obvious typo — a name that matches no field
under
+/// any case sensitivity — is rejected by this call. Case-dependent resolution
+/// (a name that matches only case-insensitively, or a case-fold ambiguity) is
+/// deferred to `paimon_read_builder_new_read`, which uses the case sensitivity
+/// effective then, so this stays order-independent with
+/// `paimon_read_builder_with_case_sensitive`.
///
/// # Safety
/// `rb` must be a valid pointer from `paimon_table_new_read_builder`, or null
(returns error).
@@ -148,6 +154,12 @@ pub unsafe extern "C" fn
paimon_read_builder_with_projection(
ptr = ptr.add(1);
}
+ // Best-effort early validation for obvious typos (columns that cannot
match
+ // under any case sensitivity). Core `with_projection` performs this
+ // case-independent check and otherwise stores names for lazy resolution,
so
+ // this stays order-independent with
`paimon_read_builder_with_case_sensitive`;
+ // case-dependent resolution/ambiguity errors surface later from
+ // `paimon_read_builder_new_read`.
let col_refs: Vec<&str> = col_names.iter().map(String::as_str).collect();
if let Err(e) = state.table.new_read_builder().with_projection(&col_refs) {
return paimon_error::from_paimon(e);
@@ -157,6 +169,32 @@ pub unsafe extern "C" fn
paimon_read_builder_with_projection(
std::ptr::null_mut()
}
+/// Set whether column-name matching for **projection** is case-sensitive for
+/// this ReadBuilder. Defaults to `true` (exact match). When `false`, projected
+/// column names are matched by ASCII case-folding and an ambiguous
+/// (case-colliding) request errors.
+///
+/// This does **not** affect predicate resolution: a predicate is resolved when
+/// it is constructed, so its case sensitivity is chosen by which constructor
+/// you call — `paimon_predicate_*` (case-sensitive) or the additive
+/// `paimon_predicate_*_with_case_sensitive` variant — independently of this
+/// setting.
+///
+/// # Safety
+/// `rb` must be a valid pointer from `paimon_table_new_read_builder`, or null
(returns error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_read_builder_with_case_sensitive(
+ rb: *mut paimon_read_builder,
+ case_sensitive: bool,
+) -> *mut paimon_error {
+ if let Err(e) = check_non_null(rb, "rb") {
+ return e;
+ }
+ let state = &mut *((*rb).inner as *mut ReadBuilderState);
+ state.case_sensitive = case_sensitive;
+ std::ptr::null_mut()
+}
+
/// Set a filter predicate for scan planning.
///
/// The predicate is consumed (ownership transferred to the read builder).
@@ -229,6 +267,7 @@ pub unsafe extern "C" fn paimon_read_builder_new_read(
}
let state = &*((*rb).inner as *const ReadBuilderState);
let mut rb_rust = state.table.new_read_builder();
+ rb_rust.with_case_sensitive(state.case_sensitive);
// Apply projection if set
if let Some(ref columns) = state.projected_columns {
@@ -613,6 +652,7 @@ fn coerce_integer_datum(
datum: Datum,
fields: &[DataField],
column: &str,
+ case_sensitive: bool,
) -> Result<Datum, *mut paimon_error> {
let val = match &datum {
Datum::TinyInt(v) => *v as i64,
@@ -622,8 +662,22 @@ fn coerce_integer_datum(
_ => return Ok(datum),
};
- let Some(field) = fields.iter().find(|f| f.name() == column) else {
- // Column not found; let PredicateBuilder produce the proper error.
+ // Resolve the column with the same case sensitivity as PredicateBuilder.
+ // A non-unique (absent or ambiguous) match is left uncoerced so the
+ // PredicateBuilder produces the proper not-found / ambiguous error.
+ let field = if case_sensitive {
+ fields.iter().find(|f| f.name() == column)
+ } else {
+ let mut hits = fields
+ .iter()
+ .filter(|f| f.name().eq_ignore_ascii_case(column));
+ match (hits.next(), hits.next()) {
+ (Some(f), None) => Some(f),
+ _ => None,
+ }
+ };
+ let Some(field) = field else {
+ // Column not found / ambiguous; let PredicateBuilder produce the
error.
return Ok(datum);
};
@@ -668,6 +722,7 @@ unsafe fn build_leaf_predicate_datum(
table: *const paimon_table,
column: *const std::ffi::c_char,
datum: &paimon_datum,
+ case_sensitive: bool,
build_fn: impl FnOnce(&PredicateBuilder, &str, Datum) ->
paimon::Result<Predicate>,
) -> paimon_result_predicate {
if let Err(e) = check_non_null(table, "table") {
@@ -699,7 +754,7 @@ unsafe fn build_leaf_predicate_datum(
let table_ref = &*((*table).inner as *const Table);
let fields = table_ref.schema().fields();
- let d = match coerce_integer_datum(d, fields, &col_name) {
+ let d = match coerce_integer_datum(d, fields, &col_name, case_sensitive) {
Ok(d) => d,
Err(e) => {
return paimon_result_predicate {
@@ -709,7 +764,7 @@ unsafe fn build_leaf_predicate_datum(
}
};
- let pb = PredicateBuilder::new(fields);
+ let pb = PredicateBuilder::new_with_case_sensitive(fields, case_sensitive);
match build_fn(&pb, &col_name, d) {
Ok(pred) => {
let inner = Box::into_raw(Box::new(pred)) as *mut c_void;
@@ -729,6 +784,7 @@ unsafe fn build_leaf_predicate_datum(
unsafe fn build_leaf_predicate(
table: *const paimon_table,
column: *const std::ffi::c_char,
+ case_sensitive: bool,
build_fn: impl FnOnce(&PredicateBuilder, &str) ->
paimon::Result<Predicate>,
) -> paimon_result_predicate {
if let Err(e) = check_non_null(table, "table") {
@@ -747,7 +803,7 @@ unsafe fn build_leaf_predicate(
}
};
let table_ref = &*((*table).inner as *const Table);
- let pb = PredicateBuilder::new(table_ref.schema().fields());
+ let pb =
PredicateBuilder::new_with_case_sensitive(table_ref.schema().fields(),
case_sensitive);
match build_fn(&pb, &col_name) {
Ok(pred) => {
let inner = Box::into_raw(Box::new(pred)) as *mut c_void;
@@ -763,7 +819,10 @@ unsafe fn build_leaf_predicate(
}
}
-/// Create an equality predicate: `column = datum`.
+/// Create an equality predicate: `column = datum` (case-sensitive column
match).
+///
+/// For case-insensitive column matching use
+/// `paimon_predicate_equal_with_case_sensitive`.
///
/// # Safety
/// `table` and `column` must be valid pointers.
@@ -773,10 +832,26 @@ pub unsafe extern "C" fn paimon_predicate_equal(
column: *const std::ffi::c_char,
datum: paimon_datum,
) -> paimon_result_predicate {
- build_leaf_predicate_datum(table, column, &datum, |pb, col, d|
pb.equal(col, d))
+ build_leaf_predicate_datum(table, column, &datum, true, |pb, col, d|
pb.equal(col, d))
}
-/// Create a not-equal predicate: `column != datum`.
+/// Create an equality predicate with configurable column-name case
sensitivity.
+///
+/// # Safety
+/// `table` and `column` must be valid pointers.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_predicate_equal_with_case_sensitive(
+ table: *const paimon_table,
+ column: *const std::ffi::c_char,
+ datum: paimon_datum,
+ case_sensitive: bool,
+) -> paimon_result_predicate {
+ build_leaf_predicate_datum(table, column, &datum, case_sensitive, |pb,
col, d| {
+ pb.equal(col, d)
+ })
+}
+
+/// Create a not-equal predicate: `column != datum` (case-sensitive column
match).
///
/// # Safety
/// `table` and `column` must be valid pointers.
@@ -786,10 +861,28 @@ pub unsafe extern "C" fn paimon_predicate_not_equal(
column: *const std::ffi::c_char,
datum: paimon_datum,
) -> paimon_result_predicate {
- build_leaf_predicate_datum(table, column, &datum, |pb, col, d|
pb.not_equal(col, d))
+ build_leaf_predicate_datum(table, column, &datum, true, |pb, col, d| {
+ pb.not_equal(col, d)
+ })
}
-/// Create a less-than predicate: `column < datum`.
+/// Create a not-equal predicate with configurable column-name case
sensitivity.
+///
+/// # Safety
+/// `table` and `column` must be valid pointers.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_predicate_not_equal_with_case_sensitive(
+ table: *const paimon_table,
+ column: *const std::ffi::c_char,
+ datum: paimon_datum,
+ case_sensitive: bool,
+) -> paimon_result_predicate {
+ build_leaf_predicate_datum(table, column, &datum, case_sensitive, |pb,
col, d| {
+ pb.not_equal(col, d)
+ })
+}
+
+/// Create a less-than predicate: `column < datum` (case-sensitive column
match).
///
/// # Safety
/// `table` and `column` must be valid pointers.
@@ -799,10 +892,28 @@ pub unsafe extern "C" fn paimon_predicate_less_than(
column: *const std::ffi::c_char,
datum: paimon_datum,
) -> paimon_result_predicate {
- build_leaf_predicate_datum(table, column, &datum, |pb, col, d|
pb.less_than(col, d))
+ build_leaf_predicate_datum(table, column, &datum, true, |pb, col, d| {
+ pb.less_than(col, d)
+ })
}
-/// Create a less-or-equal predicate: `column <= datum`.
+/// Create a less-than predicate with configurable column-name case
sensitivity.
+///
+/// # Safety
+/// `table` and `column` must be valid pointers.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_predicate_less_than_with_case_sensitive(
+ table: *const paimon_table,
+ column: *const std::ffi::c_char,
+ datum: paimon_datum,
+ case_sensitive: bool,
+) -> paimon_result_predicate {
+ build_leaf_predicate_datum(table, column, &datum, case_sensitive, |pb,
col, d| {
+ pb.less_than(col, d)
+ })
+}
+
+/// Create a less-or-equal predicate: `column <= datum` (case-sensitive column
match).
///
/// # Safety
/// `table` and `column` must be valid pointers.
@@ -812,10 +923,28 @@ pub unsafe extern "C" fn paimon_predicate_less_or_equal(
column: *const std::ffi::c_char,
datum: paimon_datum,
) -> paimon_result_predicate {
- build_leaf_predicate_datum(table, column, &datum, |pb, col, d|
pb.less_or_equal(col, d))
+ build_leaf_predicate_datum(table, column, &datum, true, |pb, col, d| {
+ pb.less_or_equal(col, d)
+ })
+}
+
+/// Create a less-or-equal predicate with configurable column-name case
sensitivity.
+///
+/// # Safety
+/// `table` and `column` must be valid pointers.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_predicate_less_or_equal_with_case_sensitive(
+ table: *const paimon_table,
+ column: *const std::ffi::c_char,
+ datum: paimon_datum,
+ case_sensitive: bool,
+) -> paimon_result_predicate {
+ build_leaf_predicate_datum(table, column, &datum, case_sensitive, |pb,
col, d| {
+ pb.less_or_equal(col, d)
+ })
}
-/// Create a greater-than predicate: `column > datum`.
+/// Create a greater-than predicate: `column > datum` (case-sensitive column
match).
///
/// # Safety
/// `table` and `column` must be valid pointers.
@@ -825,10 +954,28 @@ pub unsafe extern "C" fn paimon_predicate_greater_than(
column: *const std::ffi::c_char,
datum: paimon_datum,
) -> paimon_result_predicate {
- build_leaf_predicate_datum(table, column, &datum, |pb, col, d|
pb.greater_than(col, d))
+ build_leaf_predicate_datum(table, column, &datum, true, |pb, col, d| {
+ pb.greater_than(col, d)
+ })
+}
+
+/// Create a greater-than predicate with configurable column-name case
sensitivity.
+///
+/// # Safety
+/// `table` and `column` must be valid pointers.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_predicate_greater_than_with_case_sensitive(
+ table: *const paimon_table,
+ column: *const std::ffi::c_char,
+ datum: paimon_datum,
+ case_sensitive: bool,
+) -> paimon_result_predicate {
+ build_leaf_predicate_datum(table, column, &datum, case_sensitive, |pb,
col, d| {
+ pb.greater_than(col, d)
+ })
}
-/// Create a greater-or-equal predicate: `column >= datum`.
+/// Create a greater-or-equal predicate: `column >= datum` (case-sensitive
column match).
///
/// # Safety
/// `table` and `column` must be valid pointers.
@@ -838,12 +985,28 @@ pub unsafe extern "C" fn
paimon_predicate_greater_or_equal(
column: *const std::ffi::c_char,
datum: paimon_datum,
) -> paimon_result_predicate {
- build_leaf_predicate_datum(table, column, &datum, |pb, col, d| {
+ build_leaf_predicate_datum(table, column, &datum, true, |pb, col, d| {
+ pb.greater_or_equal(col, d)
+ })
+}
+
+/// Create a greater-or-equal predicate with configurable column-name case
sensitivity.
+///
+/// # Safety
+/// `table` and `column` must be valid pointers.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_predicate_greater_or_equal_with_case_sensitive(
+ table: *const paimon_table,
+ column: *const std::ffi::c_char,
+ datum: paimon_datum,
+ case_sensitive: bool,
+) -> paimon_result_predicate {
+ build_leaf_predicate_datum(table, column, &datum, case_sensitive, |pb,
col, d| {
pb.greater_or_equal(col, d)
})
}
-/// Create an IS NULL predicate.
+/// Create an IS NULL predicate (case-sensitive column match).
///
/// # Safety
/// `table` and `column` must be valid pointers.
@@ -852,10 +1015,23 @@ pub unsafe extern "C" fn paimon_predicate_is_null(
table: *const paimon_table,
column: *const std::ffi::c_char,
) -> paimon_result_predicate {
- build_leaf_predicate(table, column, |pb, col| pb.is_null(col))
+ build_leaf_predicate(table, column, true, |pb, col| pb.is_null(col))
}
-/// Create an IS NOT NULL predicate.
+/// Create an IS NULL predicate with configurable column-name case sensitivity.
+///
+/// # Safety
+/// `table` and `column` must be valid pointers.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_predicate_is_null_with_case_sensitive(
+ table: *const paimon_table,
+ column: *const std::ffi::c_char,
+ case_sensitive: bool,
+) -> paimon_result_predicate {
+ build_leaf_predicate(table, column, case_sensitive, |pb, col|
pb.is_null(col))
+}
+
+/// Create an IS NOT NULL predicate (case-sensitive column match).
///
/// # Safety
/// `table` and `column` must be valid pointers.
@@ -864,10 +1040,23 @@ pub unsafe extern "C" fn paimon_predicate_is_not_null(
table: *const paimon_table,
column: *const std::ffi::c_char,
) -> paimon_result_predicate {
- build_leaf_predicate(table, column, |pb, col| pb.is_not_null(col))
+ build_leaf_predicate(table, column, true, |pb, col| pb.is_not_null(col))
}
-/// Create an IN predicate: `column IN (datum1, datum2, ...)`.
+/// Create an IS NOT NULL predicate with configurable column-name case
sensitivity.
+///
+/// # Safety
+/// `table` and `column` must be valid pointers.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_predicate_is_not_null_with_case_sensitive(
+ table: *const paimon_table,
+ column: *const std::ffi::c_char,
+ case_sensitive: bool,
+) -> paimon_result_predicate {
+ build_leaf_predicate(table, column, case_sensitive, |pb, col|
pb.is_not_null(col))
+}
+
+/// Create an IN predicate: `column IN (datum1, datum2, ...)` (case-sensitive
column match).
///
/// # Safety
/// `table`, `column`, and `datums` must be valid pointers. `datums_len` must
be the length.
@@ -878,12 +1067,39 @@ pub unsafe extern "C" fn paimon_predicate_is_in(
datums: *const paimon_datum,
datums_len: usize,
) -> paimon_result_predicate {
- build_leaf_predicate_datums(table, column, datums, datums_len, |pb, col,
values| {
- pb.is_in(col, values)
- })
+ build_leaf_predicate_datums(
+ table,
+ column,
+ datums,
+ datums_len,
+ true,
+ |pb, col, values| pb.is_in(col, values),
+ )
}
-/// Create a NOT IN predicate: `column NOT IN (datum1, datum2, ...)`.
+/// Create an IN predicate with configurable column-name case sensitivity.
+///
+/// # Safety
+/// `table`, `column`, and `datums` must be valid pointers. `datums_len` must
be the length.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_predicate_is_in_with_case_sensitive(
+ table: *const paimon_table,
+ column: *const std::ffi::c_char,
+ datums: *const paimon_datum,
+ datums_len: usize,
+ case_sensitive: bool,
+) -> paimon_result_predicate {
+ build_leaf_predicate_datums(
+ table,
+ column,
+ datums,
+ datums_len,
+ case_sensitive,
+ |pb, col, values| pb.is_in(col, values),
+ )
+}
+
+/// Create a NOT IN predicate: `column NOT IN (datum1, datum2, ...)`
(case-sensitive column match).
///
/// # Safety
/// `table`, `column`, and `datums` must be valid pointers. `datums_len` must
be the length.
@@ -894,9 +1110,36 @@ pub unsafe extern "C" fn paimon_predicate_is_not_in(
datums: *const paimon_datum,
datums_len: usize,
) -> paimon_result_predicate {
- build_leaf_predicate_datums(table, column, datums, datums_len, |pb, col,
values| {
- pb.is_not_in(col, values)
- })
+ build_leaf_predicate_datums(
+ table,
+ column,
+ datums,
+ datums_len,
+ true,
+ |pb, col, values| pb.is_not_in(col, values),
+ )
+}
+
+/// Create a NOT IN predicate with configurable column-name case sensitivity.
+///
+/// # Safety
+/// `table`, `column`, and `datums` must be valid pointers. `datums_len` must
be the length.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_predicate_is_not_in_with_case_sensitive(
+ table: *const paimon_table,
+ column: *const std::ffi::c_char,
+ datums: *const paimon_datum,
+ datums_len: usize,
+ case_sensitive: bool,
+) -> paimon_result_predicate {
+ build_leaf_predicate_datums(
+ table,
+ column,
+ datums,
+ datums_len,
+ case_sensitive,
+ |pb, col, values| pb.is_not_in(col, values),
+ )
}
/// Helper to build an IN/NOT IN predicate with a datum array.
@@ -905,6 +1148,7 @@ unsafe fn build_leaf_predicate_datums(
column: *const std::ffi::c_char,
datums: *const paimon_datum,
datums_len: usize,
+ case_sensitive: bool,
build_fn: impl FnOnce(&PredicateBuilder, &str, Vec<Datum>) ->
paimon::Result<Predicate>,
) -> paimon_result_predicate {
if let Err(e) = check_non_null(table, "table") {
@@ -954,7 +1198,7 @@ unsafe fn build_leaf_predicate_datums(
let values: Result<Vec<Datum>, _> = values
.into_iter()
- .map(|d| coerce_integer_datum(d, fields, &col_name))
+ .map(|d| coerce_integer_datum(d, fields, &col_name, case_sensitive))
.collect();
let values = match values {
Ok(v) => v,
@@ -966,7 +1210,7 @@ unsafe fn build_leaf_predicate_datums(
}
};
- let pb = PredicateBuilder::new(fields);
+ let pb = PredicateBuilder::new_with_case_sensitive(fields, case_sensitive);
match build_fn(&pb, &col_name, values) {
Ok(pred) => {
let inner = Box::into_raw(Box::new(pred)) as *mut c_void;
@@ -1039,3 +1283,67 @@ pub unsafe extern "C" fn paimon_predicate_free(p: *mut
paimon_predicate) {
}
}
}
+
+// --- C ABI signature guards -------------------------------------------------
+//
+// The `paimon_predicate_*` constructors are called across the FFI boundary
with
+// fixed argument counts: the Go binding prepares a libffi call interface (CIF)
+// per symbol (see `bindings/go/predicate.go`), and external consumers can link
+// against the generated headers (e.g. Doris integrations). Adding a parameter
to
+// one of these existing symbols silently breaks every such caller — the extra
+// argument is read from an undefined register/stack slot at the ABI boundary.
+//
+// These compile-time assertions pin the existing signatures. To add behavior
+// (e.g. case-insensitive column matching), introduce a new
+// `paimon_predicate_*_with_case_sensitive` symbol instead of changing one of
+// these; touching a signature here will fail to compile.
+const _: unsafe extern "C" fn(
+ *const paimon_table,
+ *const std::ffi::c_char,
+ paimon_datum,
+) -> paimon_result_predicate = paimon_predicate_equal;
+const _: unsafe extern "C" fn(
+ *const paimon_table,
+ *const std::ffi::c_char,
+ paimon_datum,
+) -> paimon_result_predicate = paimon_predicate_not_equal;
+const _: unsafe extern "C" fn(
+ *const paimon_table,
+ *const std::ffi::c_char,
+ paimon_datum,
+) -> paimon_result_predicate = paimon_predicate_less_than;
+const _: unsafe extern "C" fn(
+ *const paimon_table,
+ *const std::ffi::c_char,
+ paimon_datum,
+) -> paimon_result_predicate = paimon_predicate_less_or_equal;
+const _: unsafe extern "C" fn(
+ *const paimon_table,
+ *const std::ffi::c_char,
+ paimon_datum,
+) -> paimon_result_predicate = paimon_predicate_greater_than;
+const _: unsafe extern "C" fn(
+ *const paimon_table,
+ *const std::ffi::c_char,
+ paimon_datum,
+) -> paimon_result_predicate = paimon_predicate_greater_or_equal;
+const _: unsafe extern "C" fn(
+ *const paimon_table,
+ *const std::ffi::c_char,
+) -> paimon_result_predicate = paimon_predicate_is_null;
+const _: unsafe extern "C" fn(
+ *const paimon_table,
+ *const std::ffi::c_char,
+) -> paimon_result_predicate = paimon_predicate_is_not_null;
+const _: unsafe extern "C" fn(
+ *const paimon_table,
+ *const std::ffi::c_char,
+ *const paimon_datum,
+ usize,
+) -> paimon_result_predicate = paimon_predicate_is_in;
+const _: unsafe extern "C" fn(
+ *const paimon_table,
+ *const std::ffi::c_char,
+ *const paimon_datum,
+ usize,
+) -> paimon_result_predicate = paimon_predicate_is_not_in;
diff --git a/bindings/c/src/types.rs b/bindings/c/src/types.rs
index 4df28de..c7b059a 100644
--- a/bindings/c/src/types.rs
+++ b/bindings/c/src/types.rs
@@ -83,6 +83,7 @@ pub(crate) struct ReadBuilderState {
pub table: Table,
pub projected_columns: Option<Vec<String>>,
pub filter: Option<Predicate>,
+ pub case_sensitive: bool,
}
/// Internal state for TableScan that stores table and filter.
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 6b66d1f..b0d9e4b 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -54,6 +54,17 @@ class TableRead:
class ReadBuilder:
def with_projection(self, columns: List[str]) -> "ReadBuilder": ...
+ def with_case_sensitive(self, case_sensitive: bool) -> "ReadBuilder":
+ """
+ Set whether column-name matching (projection and predicate column
+ resolution) is case-sensitive. Defaults to ``True`` (exact match).
+
+ Projection resolution is lazy, so this is order-independent with
+ ``with_projection``. Predicates built via ``with_filter`` capture case
+ sensitivity when they are constructed, so ``with_case_sensitive`` must
be
+ set before ``with_filter`` for the filter to honor it.
+ """
+ ...
def with_limit(self, limit: int) -> "ReadBuilder": ...
def with_filter(self, predicate: dict) -> "ReadBuilder": ...
def new_scan(self) -> TableScan: ...
diff --git a/bindings/python/src/predicate.rs b/bindings/python/src/predicate.rs
index 08b2f5c..e7efba4 100644
--- a/bindings/python/src/predicate.rs
+++ b/bindings/python/src/predicate.rs
@@ -362,6 +362,7 @@ const METHOD_NOT_SUPPORTED: &[&str] = &["not"];
pub(crate) fn dict_to_predicate(
node: &Bound<'_, PyDict>,
fields: &[DataField],
+ case_sensitive: bool,
) -> PyResult<Predicate> {
let method: String = node
.get_item("method")?
@@ -387,7 +388,7 @@ pub(crate) fn dict_to_predicate(
.cast::<PyDict>()
.map_err(|_| PyValueError::new_err("each child must be a
dict"))?;
// Unsupported child propagates → no partial pushdown.
- preds.push(dict_to_predicate(child_dict, fields)?);
+ preds.push(dict_to_predicate(child_dict, fields,
case_sensitive)?);
}
Ok(if method == "and" {
Predicate::and(preds)
@@ -398,16 +399,48 @@ pub(crate) fn dict_to_predicate(
m if METHOD_NOT_SUPPORTED.contains(&m) =>
Err(PyNotImplementedError::new_err(format!(
"predicate operator '{m}' is not supported for Rust pushdown"
))),
- _ => leaf_to_predicate(&method, node, fields),
+ _ => leaf_to_predicate(&method, node, fields, case_sensitive),
}
}
+/// Resolve a leaf's field name to its schema [`DataType`] under the given case
+/// sensitivity.
+///
+/// Case-sensitive: exact match. Case-insensitive: ASCII-fold and require a
+/// unique match. Absent → `ValueError("Column '{field}' not found in
schema")`;
+/// ambiguous (2+ fields collide under folding) → `ValueError` naming the
clash.
+fn resolve_leaf_field_type(
+ fields: &[DataField],
+ field: &str,
+ case_sensitive: bool,
+) -> PyResult<DataType> {
+ let not_found = || PyValueError::new_err(format!("Column '{field}' not
found in schema"));
+ if case_sensitive {
+ return fields
+ .iter()
+ .find(|f| f.name() == field)
+ .map(|f| f.data_type().clone())
+ .ok_or_else(not_found);
+ }
+ let mut matches = fields
+ .iter()
+ .filter(|f| f.name().eq_ignore_ascii_case(field));
+ let first = matches.next().ok_or_else(not_found)?;
+ if matches.next().is_some() {
+ return Err(PyValueError::new_err(format!(
+ "Ambiguous column '{field}': multiple schema fields match
case-insensitively"
+ )));
+ }
+ Ok(first.data_type().clone())
+}
+
/// Convert a single leaf dict (already known not to be `and`/`or`) into a
/// [`Predicate`], resolving the field type from the schema.
fn leaf_to_predicate(
method: &str,
node: &Bound<'_, PyDict>,
fields: &[DataField],
+ case_sensitive: bool,
) -> PyResult<Predicate> {
let field: String = node
.get_item("field")?
@@ -415,14 +448,10 @@ fn leaf_to_predicate(
.extract()?;
// Resolve field DataType from schema (authoritative) for literal
conversion.
- let data_type = fields
- .iter()
- .find(|f| f.name() == field)
- .map(|f| f.data_type().clone())
- .ok_or_else(|| PyValueError::new_err(format!("Column '{field}' not
found in schema")))?;
+ let data_type = resolve_leaf_field_type(fields, &field, case_sensitive)?;
let literals_obj = node.get_item("literals")?;
- let pb = PredicateBuilder::new(fields);
+ let pb = PredicateBuilder::new_with_case_sensitive(fields, case_sensitive);
// Convert literals (DataType-driven), wrapping NotImplemented type
messages
// with field context.
@@ -607,7 +636,7 @@ mod tests {
Python::attach(|py| {
let fields = test_fields();
let dict = leaf_dict(py, "equal", "id", &[1]);
- let pred = dict_to_predicate(&dict, &fields).unwrap();
+ let pred = dict_to_predicate(&dict, &fields, true).unwrap();
match &pred {
Predicate::Leaf {
column,
@@ -629,7 +658,7 @@ mod tests {
Python::attach(|py| {
let fields = test_fields();
let dict = leaf_dict(py, "not", "id", &[1]);
- let err = dict_to_predicate(&dict, &fields).unwrap_err();
+ let err = dict_to_predicate(&dict, &fields, true).unwrap_err();
assert!(err.is_instance_of::<PyNotImplementedError>(py));
});
}
@@ -666,7 +695,7 @@ mod tests {
Python::attach(|py| {
let fields = test_fields();
let dict = str_leaf_dict(py, "startsWith", "name", &["ab"]);
- let pred = dict_to_predicate(&dict, &fields).unwrap();
+ let pred = dict_to_predicate(&dict, &fields, true).unwrap();
expect_leaf_op(&pred, PredicateOperator::StartsWith);
});
}
@@ -676,7 +705,7 @@ mod tests {
Python::attach(|py| {
let fields = test_fields();
let dict = str_leaf_dict(py, "endsWith", "name", &["ab"]);
- let pred = dict_to_predicate(&dict, &fields).unwrap();
+ let pred = dict_to_predicate(&dict, &fields, true).unwrap();
expect_leaf_op(&pred, PredicateOperator::EndsWith);
});
}
@@ -686,17 +715,68 @@ mod tests {
Python::attach(|py| {
let fields = test_fields();
let dict = str_leaf_dict(py, "contains", "name", &["ab"]);
- let pred = dict_to_predicate(&dict, &fields).unwrap();
+ let pred = dict_to_predicate(&dict, &fields, true).unwrap();
expect_leaf_op(&pred, PredicateOperator::Contains);
});
}
+ /// Fields whose string column is spelled `Name` (mixed case).
+ fn mixed_case_fields() -> Vec<DataField> {
+ vec![
+ DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(
+ 1,
+ "Name".to_string(),
+ DataType::VarChar(VarCharType::default()),
+ ),
+ ]
+ }
+
+ #[test]
+ fn case_insensitive_leaf_resolves_differently_cased_field() {
+ Python::attach(|py| {
+ let fields = mixed_case_fields();
+ // Request uses `name`; the schema field is `Name`.
+ let dict = str_leaf_dict(py, "equal", "name", &["bob"]);
+
+ // Case-sensitive (default): `name` != `Name` → column not found.
+ let err = dict_to_predicate(&dict, &fields, true).unwrap_err();
+ assert!(err.is_instance_of::<PyValueError>(py));
+
+ // Case-insensitive: resolves to the canonical `Name`.
+ let pred = dict_to_predicate(&dict, &fields, false).unwrap();
+ match &pred {
+ Predicate::Leaf { column, op, .. } => {
+ assert_eq!(
+ column, "Name",
+ "canonical schema name is stored in the leaf"
+ );
+ assert_eq!(*op, PredicateOperator::Eq);
+ }
+ other => panic!("expected Leaf, got {other:?}"),
+ }
+ });
+ }
+
+ #[test]
+ fn case_insensitive_ambiguous_field_raises_value_error() {
+ Python::attach(|py| {
+ let fields = vec![
+ DataField::new(0, "Col".to_string(),
DataType::Int(IntType::new())),
+ DataField::new(1, "col".to_string(),
DataType::Int(IntType::new())),
+ ];
+ let dict = leaf_dict(py, "equal", "COL", &[1]);
+ let err = dict_to_predicate(&dict, &fields, false).unwrap_err();
+ assert!(err.is_instance_of::<PyValueError>(py));
+ });
+ }
+
#[test]
fn like_prefix_pattern_optimizes_to_starts_with() {
Python::attach(|py| {
let fields = test_fields();
let dict = str_leaf_dict(py, "like", "name", &["ab%"]);
- let pred = dict_to_predicate(&dict, &fields).unwrap();
+ let pred = dict_to_predicate(&dict, &fields, true).unwrap();
expect_leaf_op(&pred, PredicateOperator::StartsWith);
});
}
@@ -706,7 +786,7 @@ mod tests {
Python::attach(|py| {
let fields = test_fields();
let dict = str_leaf_dict(py, "like", "name", &["a%b%c"]);
- let pred = dict_to_predicate(&dict, &fields).unwrap();
+ let pred = dict_to_predicate(&dict, &fields, true).unwrap();
expect_leaf_op(&pred, PredicateOperator::Like);
});
}
@@ -716,7 +796,7 @@ mod tests {
Python::attach(|py| {
let fields = test_fields();
let dict = str_leaf_dict(py, "like", "name", &["100\\%%", "\\"]);
- let pred = dict_to_predicate(&dict, &fields).unwrap();
+ let pred = dict_to_predicate(&dict, &fields, true).unwrap();
// Escaped-wildcard patterns are not rewritten by the core's LIKE
// optimization; they stay as a residual Like leaf.
expect_leaf_op(&pred, PredicateOperator::Like);
@@ -728,7 +808,7 @@ mod tests {
Python::attach(|py| {
let fields = test_fields();
let dict = str_leaf_dict(py, "like", "name", &["100!%%", "!"]);
- let err = dict_to_predicate(&dict, &fields).unwrap_err();
+ let err = dict_to_predicate(&dict, &fields, true).unwrap_err();
assert!(err.is_instance_of::<PyValueError>(py));
});
}
@@ -738,7 +818,7 @@ mod tests {
Python::attach(|py| {
let fields = test_fields();
let dict = str_leaf_dict(py, "like", "name", &["a%", "ab"]);
- let err = dict_to_predicate(&dict, &fields).unwrap_err();
+ let err = dict_to_predicate(&dict, &fields, true).unwrap_err();
assert!(err.is_instance_of::<PyValueError>(py));
});
}
@@ -748,7 +828,7 @@ mod tests {
Python::attach(|py| {
let fields = test_fields();
let dict = str_leaf_dict(py, "like", "name", &["a%", "\\", "x"]);
- let err = dict_to_predicate(&dict, &fields).unwrap_err();
+ let err = dict_to_predicate(&dict, &fields, true).unwrap_err();
assert!(err.is_instance_of::<PyValueError>(py));
});
}
@@ -759,7 +839,7 @@ mod tests {
let fields = test_fields();
for method in ["startsWith", "endsWith", "contains"] {
let dict = str_leaf_dict(py, method, "name", &[""]);
- let pred = dict_to_predicate(&dict, &fields).unwrap();
+ let pred = dict_to_predicate(&dict, &fields, true).unwrap();
expect_leaf_op(&pred, PredicateOperator::IsNotNull);
}
});
@@ -771,7 +851,7 @@ mod tests {
let fields = test_fields();
for method in ["startsWith", "endsWith", "contains", "like"] {
let dict = str_leaf_dict(py, method, "id", &["a"]);
- let err = dict_to_predicate(&dict, &fields).unwrap_err();
+ let err = dict_to_predicate(&dict, &fields, true).unwrap_err();
assert!(err.is_instance_of::<PyValueError>(py), "{method}");
}
});
@@ -783,12 +863,12 @@ mod tests {
let fields = test_fields();
for method in ["startsWith", "endsWith", "contains", "like"] {
let dict = str_leaf_dict(py, method, "name", &[]);
- let err = dict_to_predicate(&dict, &fields).unwrap_err();
+ let err = dict_to_predicate(&dict, &fields, true).unwrap_err();
assert!(err.is_instance_of::<PyValueError>(py), "{method}
zero");
}
for method in ["startsWith", "endsWith", "contains"] {
let dict = str_leaf_dict(py, method, "name", &["a", "b"]);
- let err = dict_to_predicate(&dict, &fields).unwrap_err();
+ let err = dict_to_predicate(&dict, &fields, true).unwrap_err();
assert!(err.is_instance_of::<PyValueError>(py), "{method}
two");
}
});
@@ -802,7 +882,7 @@ mod tests {
// leaf path: field resolution happens first, so an unknown field
is
// a ValueError (not NotImplementedError as before).
let dict = str_leaf_dict(py, "like", "nope", &["x"]);
- let err = dict_to_predicate(&dict, &fields).unwrap_err();
+ let err = dict_to_predicate(&dict, &fields, true).unwrap_err();
assert!(err.is_instance_of::<PyValueError>(py));
});
}
@@ -816,7 +896,7 @@ mod tests {
let dict = PyDict::new(py);
dict.set_item("method", "not").unwrap();
dict.set_item("children", PyList::empty(py)).unwrap();
- let err = dict_to_predicate(&dict, &fields).unwrap_err();
+ let err = dict_to_predicate(&dict, &fields, true).unwrap_err();
assert!(err.is_instance_of::<PyNotImplementedError>(py));
});
}
@@ -826,7 +906,7 @@ mod tests {
Python::attach(|py| {
let fields = test_fields();
let dict = leaf_dict(py, "equal", "nope", &[1]);
- let err = dict_to_predicate(&dict, &fields).unwrap_err();
+ let err = dict_to_predicate(&dict, &fields, true).unwrap_err();
assert!(err.is_instance_of::<PyValueError>(py));
});
}
@@ -838,7 +918,7 @@ mod tests {
let dict = PyDict::new(py);
dict.set_item("method", "and").unwrap();
dict.set_item("children", PyList::empty(py)).unwrap();
- let err = dict_to_predicate(&dict, &fields).unwrap_err();
+ let err = dict_to_predicate(&dict, &fields, true).unwrap_err();
assert!(err.is_instance_of::<PyValueError>(py));
});
}
@@ -855,7 +935,7 @@ mod tests {
let dict = PyDict::new(py);
dict.set_item("method", "and").unwrap();
dict.set_item("children", children).unwrap();
- let err = dict_to_predicate(&dict, &fields).unwrap_err();
+ let err = dict_to_predicate(&dict, &fields, true).unwrap_err();
// No partial pushdown: the unsupported child propagates.
assert!(err.is_instance_of::<PyNotImplementedError>(py));
});
@@ -873,7 +953,7 @@ mod tests {
let dict = PyDict::new(py);
dict.set_item("method", "and").unwrap();
dict.set_item("children", children).unwrap();
- let pred = dict_to_predicate(&dict, &fields).unwrap();
+ let pred = dict_to_predicate(&dict, &fields, true).unwrap();
match &pred {
Predicate::And(ch) => assert_eq!(ch.len(), 2),
other => panic!("expected And, got {other:?}"),
@@ -893,7 +973,7 @@ mod tests {
let dict = PyDict::new(py);
dict.set_item("method", "or").unwrap();
dict.set_item("children", children).unwrap();
- let pred = dict_to_predicate(&dict, &fields).unwrap();
+ let pred = dict_to_predicate(&dict, &fields, true).unwrap();
match &pred {
Predicate::Or(ch) => assert_eq!(ch.len(), 2),
other => panic!("expected Or, got {other:?}"),
@@ -911,7 +991,7 @@ mod tests {
let lits = PyList::empty(py);
lits.append(py.None()).unwrap();
d.set_item("literals", lits).unwrap();
- let err = dict_to_predicate(&d, &fields).unwrap_err();
+ let err = dict_to_predicate(&d, &fields, true).unwrap_err();
assert!(err.is_instance_of::<PyValueError>(py));
});
}
@@ -921,7 +1001,7 @@ mod tests {
Python::attach(|py| {
let fields = test_fields();
let dict = leaf_dict(py, "equal", "id", &[1, 2]);
- let err = dict_to_predicate(&dict, &fields).unwrap_err();
+ let err = dict_to_predicate(&dict, &fields, true).unwrap_err();
assert!(err.is_instance_of::<PyValueError>(py));
});
}
@@ -932,7 +1012,7 @@ mod tests {
let fields = test_fields();
for method in ["isNull", "isNotNull"] {
let dict = leaf_dict(py, method, "name", &[1]);
- let err = dict_to_predicate(&dict, &fields).unwrap_err();
+ let err = dict_to_predicate(&dict, &fields, true).unwrap_err();
assert!(err.is_instance_of::<PyValueError>(py));
}
});
@@ -944,12 +1024,12 @@ mod tests {
let fields = test_fields();
// Empty literals list is accepted.
let with_empty = leaf_dict(py, "isNull", "name", &[]);
- assert!(dict_to_predicate(&with_empty, &fields).is_ok());
+ assert!(dict_to_predicate(&with_empty, &fields, true).is_ok());
// Missing literals key is accepted.
let no_lits = PyDict::new(py);
no_lits.set_item("method", "isNotNull").unwrap();
no_lits.set_item("field", "name").unwrap();
- assert!(dict_to_predicate(&no_lits, &fields).is_ok());
+ assert!(dict_to_predicate(&no_lits, &fields, true).is_ok());
});
}
diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs
index b5d4008..49217a4 100644
--- a/bindings/python/src/read.rs
+++ b/bindings/python/src/read.rs
@@ -69,7 +69,9 @@ fn apply_read_config(
projection: &Option<Vec<String>>,
limit: Option<usize>,
filter: &Option<Predicate>,
+ case_sensitive: bool,
) -> PyResult<()> {
+ builder.with_case_sensitive(case_sensitive);
if let Some(projection) = projection {
let cols: Vec<&str> = projection.iter().map(String::as_str).collect();
builder.with_projection(&cols).map_err(to_py_err)?;
@@ -108,6 +110,7 @@ pub struct PyReadBuilder {
projection: Option<Vec<String>>,
limit: Option<usize>,
filter: Option<Predicate>,
+ case_sensitive: bool,
}
impl PyReadBuilder {
@@ -117,6 +120,7 @@ impl PyReadBuilder {
projection: None,
limit: None,
filter: None,
+ case_sensitive: true,
}
}
@@ -158,6 +162,7 @@ impl PyReadBuilder {
projection: None,
limit: None,
filter: None,
+ case_sensitive: true,
})
}
}
@@ -174,6 +179,17 @@ impl PyReadBuilder {
slf
}
+ /// Set whether column-name matching (projection and predicate resolution)
is
+ /// case-sensitive. Defaults to `true` (exact match); pass `false` to match
+ /// column names case-insensitively (ASCII case-folding).
+ fn with_case_sensitive(
+ mut slf: PyRefMut<'_, Self>,
+ case_sensitive: bool,
+ ) -> PyRefMut<'_, Self> {
+ slf.case_sensitive = case_sensitive;
+ slf
+ }
+
/// Convert a lightweight dict predicate into a Rust [`Predicate`] and
store
/// it for pushdown. Conversion happens immediately, so conversion errors
/// (unknown field, type mismatch, unsupported operator/type) surface at
call
@@ -182,7 +198,7 @@ impl PyReadBuilder {
mut slf: PyRefMut<'py, Self>,
predicate: &Bound<'_, PyDict>,
) -> PyResult<PyRefMut<'py, Self>> {
- let filter = dict_to_predicate(predicate,
slf.table.schema().fields())?;
+ let filter = dict_to_predicate(predicate, slf.table.schema().fields(),
slf.case_sensitive)?;
slf.filter = Some(filter);
Ok(slf)
}
@@ -193,6 +209,7 @@ impl PyReadBuilder {
projection: self.projection.clone(),
limit: self.limit,
filter: self.filter.clone(),
+ case_sensitive: self.case_sensitive,
}
}
@@ -202,6 +219,7 @@ impl PyReadBuilder {
projection: self.projection.clone(),
limit: self.limit,
filter: self.filter.clone(),
+ case_sensitive: self.case_sensitive,
}
}
}
@@ -212,6 +230,7 @@ pub struct PyTableScan {
projection: Option<Vec<String>>,
limit: Option<usize>,
filter: Option<Predicate>,
+ case_sensitive: bool,
}
#[pymethods]
@@ -221,7 +240,13 @@ impl PyTableScan {
let splits = py.detach(|| {
rt.block_on(async {
let mut builder = self.table.new_read_builder();
- apply_read_config(&mut builder, &self.projection, self.limit,
&self.filter)?;
+ apply_read_config(
+ &mut builder,
+ &self.projection,
+ self.limit,
+ &self.filter,
+ self.case_sensitive,
+ )?;
let plan = builder.new_scan().plan().await.map_err(to_py_err)?;
Ok::<_, PyErr>(plan.splits().to_vec())
})
@@ -236,6 +261,7 @@ pub struct PyTableRead {
projection: Option<Vec<String>>,
limit: Option<usize>,
filter: Option<Predicate>,
+ case_sensitive: bool,
}
#[pymethods]
@@ -247,7 +273,13 @@ impl PyTableRead {
let batches = py.detach(|| {
rt.block_on(async {
let mut builder = self.table.new_read_builder();
- apply_read_config(&mut builder, &self.projection, self.limit,
&self.filter)?;
+ apply_read_config(
+ &mut builder,
+ &self.projection,
+ self.limit,
+ &self.filter,
+ self.case_sensitive,
+ )?;
// Validate config (e.g. projection) before the empty-splits
fast
// path so an invalid projection fails consistently regardless
of
// how many splits are passed.
diff --git a/crates/integration_tests/tests/read_tables.rs
b/crates/integration_tests/tests/read_tables.rs
index 11a390f..f867f7c 100644
--- a/crates/integration_tests/tests/read_tables.rs
+++ b/crates/integration_tests/tests/read_tables.rs
@@ -727,10 +727,16 @@ async fn test_read_projection_duplicate_column() {
let catalog = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "simple_log_table").await;
+ // `id` exists, so with_projection passes its best-effort early check; the
+ // duplicate is a case-dependent outcome resolved lazily, so it surfaces
when
+ // the read is built.
let mut read_builder = table.new_read_builder();
- let err = read_builder
+ read_builder
.with_projection(&["id", "id"])
- .expect_err("Duplicate projection should fail");
+ .expect("with_projection defers duplicate detection to read build");
+ let err = read_builder
+ .new_read()
+ .expect_err("Duplicate projection should fail at new_read");
assert!(
matches!(&err, Error::ConfigInvalid { message } if
message.contains("Duplicate projection column 'id'")),
diff --git a/crates/integrations/datafusion/src/filter_pushdown.rs
b/crates/integrations/datafusion/src/filter_pushdown.rs
index 1739a99..cef12aa 100644
--- a/crates/integrations/datafusion/src/filter_pushdown.rs
+++ b/crates/integrations/datafusion/src/filter_pushdown.rs
@@ -40,8 +40,12 @@ struct TranslatedPredicate {
requires_residual: bool,
}
-fn analyze_filter(filter: &Expr, fields: &[DataField]) -> SingleFilterAnalysis
{
- let translator = FilterTranslator::new(fields);
+fn analyze_filter(
+ filter: &Expr,
+ fields: &[DataField],
+ case_sensitive: bool,
+) -> SingleFilterAnalysis {
+ let translator = FilterTranslator::new(fields, case_sensitive);
if let Some(translated) = translator.translate(filter) {
return SingleFilterAnalysis {
translated_predicates: vec![translated.predicate],
@@ -59,12 +63,16 @@ fn analyze_filter(filter: &Expr, fields: &[DataField]) ->
SingleFilterAnalysis {
}
}
-pub(crate) fn analyze_filters(filters: &[Expr], fields: &[DataField]) ->
FilterPushdownAnalysis {
+pub(crate) fn analyze_filters(
+ filters: &[Expr],
+ fields: &[DataField],
+ case_sensitive: bool,
+) -> FilterPushdownAnalysis {
let mut translated_predicates = Vec::new();
let mut requires_residual = false;
for filter in filters {
- let analysis = analyze_filter(filter, fields);
+ let analysis = analyze_filter(filter, fields, case_sensitive);
translated_predicates.extend(analysis.translated_predicates);
requires_residual |= analysis.requires_residual;
}
@@ -81,18 +89,29 @@ pub(crate) fn analyze_filters(filters: &[Expr], fields:
&[DataField]) -> FilterP
#[cfg(test)]
pub(crate) fn build_pushed_predicate(filters: &[Expr], fields: &[DataField])
-> Option<Predicate> {
- analyze_filters(filters, fields).pushed_predicate
+ analyze_filters(filters, fields, true).pushed_predicate
}
pub(crate) fn classify_filter_pushdown<F>(
filter: &Expr,
fields: &[DataField],
+ case_sensitive: bool,
is_exact_filter_pushdown: F,
) -> TableProviderFilterPushDown
where
F: Fn(&Predicate) -> bool,
{
- let translator = FilterTranslator::new(fields);
+ // `FilterTranslator` still supports case-insensitive column resolution for
+ // direct ReadBuilder API callers (and its own unit tests), but the
DataFusion
+ // TableProvider/SQL path always passes `case_sensitive = true`: the
planner
+ // resolves columns against the schema before `scan`, so SQL reads are
+ // case-sensitive. Reporting `Exact` tells DataFusion to drop its residual
+ // filter, so it must only be returned when column resolution is
unambiguous.
+ // Under case-sensitive resolution a reference matches exactly one field,
so
+ // ASCII case-folding collisions elsewhere in the schema (e.g. an unrelated
+ // `Name`/`name` pair) never make a resolved filter ambiguous and must not
+ // downgrade its classification.
+ let translator = FilterTranslator::new(fields, case_sensitive);
if let Some(translated) = translator.translate(filter) {
if translated.requires_residual {
TableProviderFilterPushDown::Inexact
@@ -129,13 +148,15 @@ fn split_conjunction(expr: &Expr) -> Vec<&Expr> {
struct FilterTranslator<'a> {
fields: &'a [DataField],
predicate_builder: PredicateBuilder,
+ case_sensitive: bool,
}
impl<'a> FilterTranslator<'a> {
- fn new(fields: &'a [DataField]) -> Self {
+ fn new(fields: &'a [DataField], case_sensitive: bool) -> Self {
Self {
fields,
- predicate_builder: PredicateBuilder::new(fields),
+ predicate_builder:
PredicateBuilder::new_with_case_sensitive(fields, case_sensitive),
+ case_sensitive,
}
}
@@ -352,7 +373,21 @@ impl<'a> FilterTranslator<'a> {
return None;
};
- self.fields.iter().find(|field| field.name() == name)
+ if self.case_sensitive {
+ return self.fields.iter().find(|field| field.name() == name);
+ }
+ // Case-insensitive: ASCII-fold and require a unique match. An
ambiguous
+ // (2+) collision returns None so the filter is left as a residual for
+ // DataFusion to apply exactly — safe, just not pushed.
+ let mut matches = self
+ .fields
+ .iter()
+ .filter(|field| field.name().eq_ignore_ascii_case(name));
+ let first = matches.next()?;
+ if matches.next().is_some() {
+ return None;
+ }
+ Some(first)
}
}
@@ -706,7 +741,34 @@ mod tests {
let filter =
Expr::Column(Column::from_name("dt")).eq(lit("2024-01-01"));
assert_eq!(
- classify_filter_pushdown(&filter, &fields,
is_exact_filter_pushdown),
+ classify_filter_pushdown(&filter, &fields, true,
is_exact_filter_pushdown),
+ TableProviderFilterPushDown::Exact
+ );
+ }
+
+ #[test]
+ fn test_classify_exact_for_case_colliding_unrelated_schema() {
+ use paimon::spec::{DataField, DataType, IntType};
+ // The SQL path is case-sensitive, so an unrelated `Name`/`name` pair
+ // that only collides under ASCII case-folding must not affect the
+ // classification of a filter on a different column: the partition
+ // column `dt` resolves to exactly one field and stays `Exact`.
+ let mut fields = test_fields();
+ let next_id = fields.len() as i32;
+ fields.push(DataField::new(
+ next_id,
+ "Name".to_string(),
+ DataType::Int(IntType::new()),
+ ));
+ fields.push(DataField::new(
+ next_id + 1,
+ "name".to_string(),
+ DataType::Int(IntType::new()),
+ ));
+ let filter =
Expr::Column(Column::from_name("dt")).eq(lit("2024-01-01"));
+
+ assert_eq!(
+ classify_filter_pushdown(&filter, &fields, true,
is_exact_filter_pushdown),
TableProviderFilterPushDown::Exact
);
}
@@ -715,7 +777,7 @@ mod tests {
fn
test_analyze_filters_for_supported_data_filter_has_no_untranslated_residual() {
let fields = test_fields();
let filters = vec![Expr::Column(Column::from_name("id")).gt(lit(10))];
- let analysis = analyze_filters(&filters, &fields);
+ let analysis = analyze_filters(&filters, &fields, true);
assert_eq!(
analysis
@@ -735,7 +797,7 @@ mod tests {
.and(Expr::Not(Box::new(
Expr::Column(Column::from_name("hr")).eq(lit(10)),
)))];
- let analysis = analyze_filters(&filters, &fields);
+ let analysis = analyze_filters(&filters, &fields, true);
assert_eq!(
analysis
@@ -753,7 +815,7 @@ mod tests {
let filters = vec![Expr::Not(Box::new(
Expr::Column(Column::from_name("dt")).eq(lit("2024-01-01")),
))];
- let analysis = analyze_filters(&filters, &fields);
+ let analysis = analyze_filters(&filters, &fields, true);
assert_eq!(
analysis
@@ -765,6 +827,65 @@ mod tests {
assert!(analysis.requires_residual);
}
+ /// Fields whose only string column is spelled `Name` (mixed case), used to
+ /// prove case-insensitive column resolution in pushdown.
+ fn mixed_case_fields() -> Vec<DataField> {
+ vec![
+ DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(
+ 1,
+ "Name".to_string(),
+ DataType::VarChar(VarCharType::string_type()),
+ ),
+ ]
+ }
+
+ #[test]
+ fn test_case_insensitive_pushdown_translates_to_canonical_name() {
+ let fields = mixed_case_fields();
+ // Request uses the lowercase spelling `name`; schema field is `Name`.
+ let filters =
vec![Expr::Column(Column::from_name("name")).eq(lit("bob"))];
+
+ // Case-sensitive (default): no match, so nothing is pushed.
+ assert!(
+ analyze_filters(&filters, &fields, true)
+ .pushed_predicate
+ .is_none(),
+ "exact matching must not resolve a differently-cased column"
+ );
+
+ // Case-insensitive: resolves to the canonical `Name` and pushes.
+ let analysis = analyze_filters(&filters, &fields, false);
+ assert_eq!(
+ analysis
+ .pushed_predicate
+ .expect("case-insensitive filter should push")
+ .to_string(),
+ "Name = 'bob'"
+ );
+ assert!(
+ !analysis.requires_residual,
+ "a translated equality is exact, not residual-only"
+ );
+ }
+
+ #[test]
+ fn test_case_insensitive_pushdown_ambiguous_falls_open() {
+ // Two fields collide under ASCII folding: resolution is ambiguous, so
the
+ // filter is left as a residual (not pushed) rather than picking one.
+ let fields = vec![
+ DataField::new(0, "Col".to_string(),
DataType::Int(IntType::new())),
+ DataField::new(1, "col".to_string(),
DataType::Int(IntType::new())),
+ ];
+ let filters = vec![Expr::Column(Column::from_name("COL")).eq(lit(1))];
+ assert!(
+ analyze_filters(&filters, &fields, false)
+ .pushed_predicate
+ .is_none(),
+ "ambiguous case-insensitive column must not be pushed"
+ );
+ }
+
#[test]
fn test_translate_reversed_partition_comparison() {
let fields = test_fields();
@@ -821,7 +942,7 @@ mod tests {
let filter = Expr::Column(Column::from_name("id")).gt(lit(10));
assert_eq!(
- classify_filter_pushdown(&filter, &fields,
is_exact_filter_pushdown),
+ classify_filter_pushdown(&filter, &fields, true,
is_exact_filter_pushdown),
TableProviderFilterPushDown::Inexact
);
}
@@ -847,7 +968,7 @@ mod tests {
.and(Expr::Column(Column::from_name("id")).gt(lit(10)));
assert_eq!(
- classify_filter_pushdown(&filter, &fields,
is_exact_filter_pushdown),
+ classify_filter_pushdown(&filter, &fields, true,
is_exact_filter_pushdown),
TableProviderFilterPushDown::Inexact
);
}
@@ -875,7 +996,7 @@ mod tests {
));
assert_eq!(
- classify_filter_pushdown(&filter, &fields,
is_exact_filter_pushdown),
+ classify_filter_pushdown(&filter, &fields, true,
is_exact_filter_pushdown),
TableProviderFilterPushDown::Inexact
);
}
@@ -1046,6 +1167,7 @@ mod tests {
classify_filter_pushdown(
&like_filter("a%", true, false),
&fields,
+ true,
is_exact_filter_pushdown
),
TableProviderFilterPushDown::Inexact
diff --git a/crates/integrations/datafusion/src/full_text_search.rs
b/crates/integrations/datafusion/src/full_text_search.rs
index 27c8db1..66a9c77 100644
--- a/crates/integrations/datafusion/src/full_text_search.rs
+++ b/crates/integrations/datafusion/src/full_text_search.rs
@@ -192,6 +192,7 @@ impl TableProvider for FullTextSearchTableProvider {
limit,
target_partitions: target,
filter_exact: false, // FTS scan does not support exact filter
pushdown
+ case_sensitive: true,
}
.build()
}
diff --git a/crates/integrations/datafusion/src/hybrid_search.rs
b/crates/integrations/datafusion/src/hybrid_search.rs
index 2d5f997..f5c0e7a 100644
--- a/crates/integrations/datafusion/src/hybrid_search.rs
+++ b/crates/integrations/datafusion/src/hybrid_search.rs
@@ -201,6 +201,7 @@ impl TableProvider for HybridSearchTableProvider {
limit,
target_partitions:
state.config_options().execution.target_partitions,
filter_exact: false,
+ case_sensitive: true,
}
.build()
}
diff --git a/crates/integrations/datafusion/src/physical_plan/scan.rs
b/crates/integrations/datafusion/src/physical_plan/scan.rs
index a0d9f3c..55d28f3 100644
--- a/crates/integrations/datafusion/src/physical_plan/scan.rs
+++ b/crates/integrations/datafusion/src/physical_plan/scan.rs
@@ -60,6 +60,9 @@ pub struct PaimonTableScan {
scan_trace: Option<ScanTrace>,
/// Human-readable Variant extraction summary for explain output.
pushed_variants: Option<String>,
+ /// Column-name case sensitivity carried from planning to execution so the
+ /// read path resolves names the same way the scan was planned.
+ case_sensitive: bool,
}
impl PaimonTableScan {
@@ -74,6 +77,7 @@ impl PaimonTableScan {
filter_exact: bool,
scan_trace: Option<ScanTrace>,
pushed_variants: Option<String>,
+ case_sensitive: bool,
) -> Self {
let plan_properties = Arc::new(PlanProperties::new(
EquivalenceProperties::new(schema.clone()),
@@ -91,6 +95,7 @@ impl PaimonTableScan {
filter_exact,
scan_trace,
pushed_variants,
+ case_sensitive,
}
}
@@ -154,10 +159,12 @@ impl ExecutionPlan for PaimonTableScan {
let schema = self.schema();
let read_type = self.read_type.clone();
let pushed_predicate = self.pushed_predicate.clone();
+ let case_sensitive = self.case_sensitive;
let fut = async move {
let mut read_builder = table.new_read_builder();
+ read_builder.with_case_sensitive(case_sensitive);
read_builder.with_read_type(read_type);
if let Some(filter) = pushed_predicate {
read_builder.with_filter(filter);
@@ -315,6 +322,7 @@ mod tests {
false,
None,
None,
+ true,
);
assert_eq!(scan.properties().output_partitioning().partition_count(),
1);
}
@@ -337,6 +345,7 @@ mod tests {
false,
None,
None,
+ true,
);
assert_eq!(scan.properties().output_partitioning().partition_count(),
3);
}
@@ -420,6 +429,7 @@ mod tests {
false,
None,
None,
+ true,
);
let ctx = SessionContext::new();
diff --git a/crates/integrations/datafusion/src/table/mod.rs
b/crates/integrations/datafusion/src/table/mod.rs
index 9e333ac..cc152f3 100644
--- a/crates/integrations/datafusion/src/table/mod.rs
+++ b/crates/integrations/datafusion/src/table/mod.rs
@@ -283,6 +283,9 @@ pub(crate) struct PaimonScanBuilder<'a> {
pub(crate) limit: Option<usize>,
pub(crate) target_partitions: usize,
pub(crate) filter_exact: bool,
+ /// Column-name case sensitivity, carried into the physical scan so
execute()
+ /// resolves names the same way planning did.
+ pub(crate) case_sensitive: bool,
}
impl PaimonScanBuilder<'_> {
@@ -324,6 +327,7 @@ impl PaimonScanBuilder<'_> {
self.filter_exact,
self.scan_trace,
None,
+ self.case_sensitive,
)))
}
}
@@ -349,9 +353,23 @@ impl TableProvider for PaimonTableProvider {
filters: &[Expr],
limit: Option<usize>,
) -> DFResult<Arc<dyn ExecutionPlan>> {
+ // Column-name matching is case-sensitive on the DataFusion path.
+ //
+ // DataFusion resolves projection/filter columns against the provider
+ // schema (`schema()`, which exposes the original field casing) during
+ // logical planning, *before* `scan` is called.
`enable_ident_normalization`
+ // only lowercases unquoted identifiers at parse time; it does not make
+ // schema resolution case-insensitive. So a genuine case mismatch
+ // (`SELECT name` against a `Name` field) already fails at planning and
+ // never reaches this code — see the negative test in
`tests/read_tables.rs`.
+ // Case-insensitive column matching is therefore offered only through
the
+ // direct ReadBuilder API (core / C / Python), not via SQL.
+ let case_sensitive = true;
// Plan splits eagerly so we know partition count upfront.
- let filter_analysis = analyze_filters(filters,
self.table.schema().fields());
+ let filter_analysis =
+ analyze_filters(filters, self.table.schema().fields(),
case_sensitive);
let mut read_builder = self.table.new_read_builder();
+ read_builder.with_case_sensitive(case_sensitive);
if let Some(indices) = projection {
let read_fields = datafusion_read_fields(&self.table);
let read_type = indices
@@ -392,6 +410,7 @@ impl TableProvider for PaimonTableProvider {
limit: pushed_limit,
target_partitions: target,
filter_exact,
+ case_sensitive,
}
.build()
}
@@ -426,12 +445,15 @@ impl TableProvider for PaimonTableProvider {
filters: &[&Expr],
) -> DFResult<Vec<TableProviderFilterPushDown>> {
let fields = self.table.schema().fields();
+ // SQL reads resolve columns case-sensitively (see `scan`), so classify
+ // pushdown the same way.
+ let case_sensitive = true;
let read_builder = self.table.new_read_builder();
Ok(filters
.iter()
.map(|filter| {
- classify_filter_pushdown(filter, fields, |predicate| {
+ classify_filter_pushdown(filter, fields, case_sensitive,
|predicate| {
read_builder.is_exact_filter_pushdown(predicate)
})
})
diff --git a/crates/integrations/datafusion/src/variant_pushdown.rs
b/crates/integrations/datafusion/src/variant_pushdown.rs
index c33339f..a5bb348 100644
--- a/crates/integrations/datafusion/src/variant_pushdown.rs
+++ b/crates/integrations/datafusion/src/variant_pushdown.rs
@@ -208,8 +208,16 @@ impl ExtensionPlanner for
VariantExtractionExtensionPlanner {
return internal_err!("PaimonVariantExtractionScan physical
planning expects no input");
}
- let filter_analysis = analyze_filters(&node.filters,
node.table.schema().fields());
+ // Column-name matching is case-sensitive on the DataFusion path,
mirroring
+ // `TableProvider::scan`: DataFusion resolves columns against the
provider
+ // schema before this planner runs, so a genuine case mismatch fails at
+ // planning and never reaches here. Case-insensitive matching is
offered
+ // only through the direct ReadBuilder API (core / C / Python), not
via SQL.
+ let case_sensitive = true;
+ let filter_analysis =
+ analyze_filters(&node.filters, node.table.schema().fields(),
case_sensitive);
let mut read_builder = node.table.new_read_builder();
+ read_builder.with_case_sensitive(case_sensitive);
read_builder.with_read_type(node.read_type.clone());
if let Some(filter) = filter_analysis.pushed_predicate.clone() {
read_builder.with_filter(filter);
@@ -250,6 +258,7 @@ impl ExtensionPlanner for VariantExtractionExtensionPlanner
{
filter_exact,
Some(scan_trace),
Some(node.pushed_variants.clone()),
+ case_sensitive,
))))
}
}
diff --git a/crates/integrations/datafusion/src/vector_search.rs
b/crates/integrations/datafusion/src/vector_search.rs
index d101cce..897b385 100644
--- a/crates/integrations/datafusion/src/vector_search.rs
+++ b/crates/integrations/datafusion/src/vector_search.rs
@@ -249,6 +249,7 @@ impl TableProvider for VectorSearchTableProvider {
limit,
target_partitions: target,
filter_exact: false,
+ case_sensitive: true,
}
.build()
}
diff --git a/crates/integrations/datafusion/tests/read_tables.rs
b/crates/integrations/datafusion/tests/read_tables.rs
index c0a5e6d..df3ed0f 100644
--- a/crates/integrations/datafusion/tests/read_tables.rs
+++ b/crates/integrations/datafusion/tests/read_tables.rs
@@ -1398,6 +1398,49 @@ async fn test_filter_row_id_from_data_evolution_table() {
}
}
+/// Pins that the DataFusion SQL path is always case-sensitive for column
names.
+///
+/// Paimon exposes case-insensitive column matching only through the direct
+/// ReadBuilder API (core / C / Python). It cannot be offered via SQL:
DataFusion
+/// resolves projection/filter columns against the provider schema during
logical
+/// planning, *before* `TableProvider::scan` runs. `enable_ident_normalization`
+/// only lowercases unquoted identifiers at parse time; it does not make schema
+/// resolution case-insensitive. So a differently-cased reference fails at
+/// planning and never reaches the scan-level resolution. This test documents
+/// that boundary so any future attempt to claim SQL-level case-insensitivity
has
+/// to confront the planner-side resolution first.
+#[tokio::test]
+async fn test_case_insensitive_column_not_supported_via_sql() {
+ let (_tmp, sql_context) = common::setup_sql_context().await;
+ // Quote the identifier so DDL parsing preserves the mixed-case field name.
+ sql_context
+ .sql("CREATE TABLE paimon.test_db.mixed_case_cols (id INT, \"Name\"
STRING)")
+ .await
+ .expect("CREATE TABLE should succeed")
+ .collect()
+ .await
+ .expect("CREATE TABLE should collect");
+
+ // The exact-case (quoted) column resolves and the query plans fine.
+ sql_context
+ .sql("SELECT \"Name\" FROM paimon.test_db.mixed_case_cols")
+ .await
+ .expect("exact-case column reference should resolve");
+
+ // An unquoted `name` is normalized to lowercase and fails during logical
+ // planning against the `Name` schema field — the scan-level
case-insensitive
+ // resolution is never reached, confirming SQL reads stay case-sensitive.
+ let err = sql_context
+ .sql("SELECT name FROM paimon.test_db.mixed_case_cols")
+ .await
+ .expect_err("case-mismatched column must fail at planning, not resolve
case-insensitively");
+ let msg = err.to_string().to_lowercase();
+ assert!(
+ msg.contains("name"),
+ "expected a column-resolution error referencing `name`, got: {err}"
+ );
+}
+
// ======================= Full-Text Search Tests =======================
#[cfg(feature = "fulltext")]
diff --git a/crates/paimon/src/spec/predicate.rs
b/crates/paimon/src/spec/predicate.rs
index badabb8..55dae84 100644
--- a/crates/paimon/src/spec/predicate.rs
+++ b/crates/paimon/src/spec/predicate.rs
@@ -615,14 +615,24 @@ impl fmt::Display for Predicate {
pub struct PredicateBuilder {
field_names: Vec<String>,
field_types: Vec<DataType>,
+ case_sensitive: bool,
}
impl PredicateBuilder {
- /// Create a new builder from schema fields. Infallible.
+ /// Create a new builder from schema fields, matching column names
+ /// case-sensitively. Infallible.
pub fn new(fields: &[DataField]) -> Self {
+ Self::new_with_case_sensitive(fields, true)
+ }
+
+ /// Create a builder with explicit case sensitivity. When `case_sensitive`
+ /// is `false`, column names are resolved by ASCII case-folding and an
+ /// ambiguous (case-colliding) schema errors. Infallible.
+ pub fn new_with_case_sensitive(fields: &[DataField], case_sensitive: bool)
-> Self {
Self {
field_names: fields.iter().map(|f| f.name().to_string()).collect(),
field_types: fields.iter().map(|f|
f.data_type().clone()).collect(),
+ case_sensitive,
}
}
@@ -801,13 +811,13 @@ impl PredicateBuilder {
/// Resolve field name to index + type, validate literals, and build a
leaf predicate.
fn leaf(&self, field: &str, op: PredicateOperator, literals: Vec<Datum>)
-> Result<Predicate> {
- let (index, data_type) = self.resolve_field(field)?;
+ let (index, canonical, data_type) = self.resolve_field(field)?;
Self::validate_literal_count(op, &literals)?;
for lit in &literals {
validate_datum_matches_type(lit, &data_type)?;
}
Ok(Predicate::Leaf {
- column: field.to_string(),
+ column: canonical,
index,
data_type,
op,
@@ -816,11 +826,41 @@ impl PredicateBuilder {
}
/// Look up a field name, returning its (index, DataType) or an error.
- fn resolve_field(&self, field: &str) -> Result<(usize, DataType)> {
- self.field_names
- .iter()
- .position(|n| n == field)
- .map(|idx| (idx, self.field_types[idx].clone()))
+ /// Resolve a column name to `(index, canonical_name, data_type)`.
+ ///
+ /// The canonical name is the schema's stored field name for the matched
+ /// index. It is stored in the resulting [`Predicate::Leaf`] so downstream
+ /// name-based lookups (e.g. global-index pruning) match the schema even
+ /// when the caller passed a different casing. Case-insensitive matching
+ /// errors on an ambiguous (case-colliding) schema.
+ fn resolve_field(&self, field: &str) -> Result<(usize, String, DataType)> {
+ let index = if self.case_sensitive {
+ self.field_names.iter().position(|n| n == field)
+ } else {
+ let mut hits = self
+ .field_names
+ .iter()
+ .enumerate()
+ .filter(|(_, n)| n.eq_ignore_ascii_case(field));
+ let first = hits.next();
+ if first.is_some() && hits.next().is_some() {
+ return Err(Error::ConfigInvalid {
+ message: format!(
+ "Ambiguous column '{field}' matches multiple schema
fields case-insensitively: {:?}",
+ self.field_names
+ ),
+ });
+ }
+ first.map(|(i, _)| i)
+ };
+ index
+ .map(|idx| {
+ (
+ idx,
+ self.field_names[idx].clone(),
+ self.field_types[idx].clone(),
+ )
+ })
.ok_or_else(|| Error::ConfigInvalid {
message: format!(
"Column '{}' not found in schema fields {:?}",
@@ -1280,6 +1320,39 @@ mod tests {
// ======================== PredicateBuilder basics
========================
+ #[test]
+ fn test_builder_case_insensitive_matches_wrong_case() {
+ let pb = PredicateBuilder::new_with_case_sensitive(&test_fields(),
false);
+ assert!(pb.equal("ID", Datum::Int(1)).is_ok());
+ }
+
+ #[test]
+ fn test_builder_case_sensitive_default_rejects_wrong_case() {
+ let pb = PredicateBuilder::new(&test_fields());
+ assert!(pb.equal("ID", Datum::Int(1)).is_err());
+ }
+
+ #[test]
+ fn test_builder_case_insensitive_stores_canonical_column_name() {
+ // The leaf must carry the canonical schema name, not the caller's
+ // casing, so downstream name-based lookups (e.g. index pruning) match.
+ let pb = PredicateBuilder::new_with_case_sensitive(&test_fields(),
false);
+ match pb.equal("ID", Datum::Int(1)).unwrap() {
+ Predicate::Leaf { column, .. } => assert_eq!(column, "id"),
+ other => panic!("expected Leaf, got {other:?}"),
+ }
+ }
+
+ #[test]
+ fn test_builder_case_insensitive_ambiguous_errors() {
+ let fields = vec![
+ DataField::new(0, "Col".to_string(),
DataType::Int(IntType::new())),
+ DataField::new(1, "col".to_string(),
DataType::Int(IntType::new())),
+ ];
+ let pb = PredicateBuilder::new_with_case_sensitive(&fields, false);
+ assert!(pb.equal("col", Datum::Int(1)).is_err());
+ }
+
#[test]
fn test_builder_equal() {
let pb = PredicateBuilder::new(&test_fields());
diff --git a/crates/paimon/src/table/format_read_builder.rs
b/crates/paimon/src/table/format_read_builder.rs
index f44d367..4fa03b6 100644
--- a/crates/paimon/src/table/format_read_builder.rs
+++ b/crates/paimon/src/table/format_read_builder.rs
@@ -18,8 +18,8 @@
//! Read builder for Java-compatible `type=format-table` metadata.
use super::partition_filter::PartitionFilter;
-use super::read_builder::resolve_projected_fields;
use super::read_builder::split_scan_predicates;
+use super::read_builder::{resolve_projected_fields,
validate_projection_possible};
use super::{Table, TableRead, TableScan};
use crate::spec::{CoreOptions, DataField, Predicate};
use crate::table::source::RowRange;
@@ -29,9 +29,14 @@ use crate::Result;
pub(crate) struct FormatReadBuilder<'a> {
table: &'a Table,
read_type: Option<Vec<DataField>>,
+ /// Deferred projection column names, resolved lazily at read build time so
+ /// projection and case sensitivity are order-independent. Mutually
exclusive
+ /// with `read_type`.
+ projection_names: Option<Vec<String>>,
partition_filter: Option<PartitionFilter>,
data_predicates: Vec<Predicate>,
limit: Option<usize>,
+ case_sensitive: bool,
}
impl<'a> FormatReadBuilder<'a> {
@@ -39,9 +44,11 @@ impl<'a> FormatReadBuilder<'a> {
Self {
table,
read_type: None,
+ projection_names: None,
partition_filter: None,
data_predicates: Vec::new(),
limit: None,
+ case_sensitive: true,
}
}
@@ -50,17 +57,26 @@ impl<'a> FormatReadBuilder<'a> {
}
pub(crate) fn with_projection(&mut self, columns: &[&str]) -> Result<&mut
Self> {
- let projection_names = columns.iter().map(|c|
(*c).to_string()).collect::<Vec<_>>();
- self.read_type = Some(resolve_projected_fields(
+ let projection_names: Vec<String> = columns.iter().map(|c|
(*c).to_string()).collect();
+ validate_projection_possible(
self.table.identifier().full_name(),
self.table.schema().fields(),
&projection_names,
- )?);
+ )?;
+ self.projection_names = Some(projection_names);
+ self.read_type = None;
Ok(self)
}
+ /// Set whether column-name matching is case-sensitive. Defaults to `true`.
+ pub(crate) fn with_case_sensitive(&mut self, case_sensitive: bool) -> &mut
Self {
+ self.case_sensitive = case_sensitive;
+ self
+ }
+
pub(crate) fn with_read_type(&mut self, read_type: Vec<DataField>) -> &mut
Self {
self.read_type = Some(read_type);
+ self.projection_names = None;
self
}
@@ -99,9 +115,9 @@ impl<'a> FormatReadBuilder<'a> {
pub(crate) fn new_read(&self) -> Result<TableRead<'a>> {
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
- let read_type = match &self.read_type {
+ let read_type = match self.resolve_read_type()? {
None => self.table.schema().fields().to_vec(),
- Some(fields) => fields.clone(),
+ Some(fields) => fields,
};
Ok(TableRead::new_format(
self.table,
@@ -110,4 +126,22 @@ impl<'a> FormatReadBuilder<'a> {
self.limit,
))
}
+
+ /// Resolve the effective read type, deferring projection name resolution
to
+ /// the case sensitivity effective at build time (order-independent with
+ /// `with_case_sensitive`).
+ fn resolve_read_type(&self) -> Result<Option<Vec<DataField>>> {
+ if let Some(read_type) = &self.read_type {
+ return Ok(Some(read_type.clone()));
+ }
+ if let Some(names) = &self.projection_names {
+ return Ok(Some(resolve_projected_fields(
+ self.table.identifier().full_name(),
+ self.table.schema().fields(),
+ names,
+ self.case_sensitive,
+ )?));
+ }
+ Ok(None)
+ }
}
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index a0fc6d7..6046ab0 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/crates/paimon/src/table/read_builder.rs
@@ -128,8 +128,14 @@ impl<'a> ReadBuilder<'a> {
}
/// Set column projection by name. Output order follows the
caller-specified order.
- /// Unknown or duplicate names cause this method to fail; an empty list is
a valid
- /// zero-column projection.
+ /// An empty list is a valid zero-column projection.
+ ///
+ /// Name resolution is deferred to scan/read build time (order-independent
with
+ /// [`with_case_sensitive`](Self::with_case_sensitive)). As a convenience,
a
+ /// column that cannot match under any case sensitivity (an obvious typo)
is
+ /// rejected here; case-dependent outcomes — a name that matches only
+ /// case-insensitively, or a case-fold duplicate/ambiguity — surface from
+ /// [`new_read`](Self::new_read) using the effective case sensitivity.
pub fn with_projection(&mut self, columns: &[&str]) -> Result<&mut Self> {
match &mut self.0 {
ReadBuilderKind::Paimon(builder) => {
@@ -142,6 +148,32 @@ impl<'a> ReadBuilder<'a> {
Ok(self)
}
+ /// Set whether column-name matching (projection and predicate column
+ /// resolution) is case-sensitive. Defaults to `true` (exact match). When
set
+ /// to `false`, names are matched by ASCII case-folding and an ambiguous
+ /// (case-colliding) request errors. This mirrors the per-read case
+ /// sensitivity engines like Spark drive from `spark.sql.caseSensitive`,
+ /// rather than being a table property.
+ ///
+ /// Projection resolution is lazy, so this affects a projection set via
+ /// [`with_projection`](Self::with_projection) regardless of call order
(the
+ /// projected names are resolved at scan/read build time using the flag
+ /// effective then). Predicates built via `PredicateBuilder` capture case
+ /// sensitivity at their own construction time, so this flag does not
+ /// retroactively change a predicate already passed to
+ /// [`with_filter`](Self::with_filter).
+ pub fn with_case_sensitive(&mut self, case_sensitive: bool) -> &mut Self {
+ match &mut self.0 {
+ ReadBuilderKind::Paimon(builder) => {
+ builder.with_case_sensitive(case_sensitive);
+ }
+ ReadBuilderKind::Format(builder) => {
+ builder.with_case_sensitive(case_sensitive);
+ }
+ }
+ self
+ }
+
/// Set the full read type, including nested field pruning or
connector-defined
/// logical read types such as Variant extractions.
pub fn with_read_type(&mut self, read_type: Vec<DataField>) -> &mut Self {
@@ -250,9 +282,15 @@ impl<'a> ReadBuilder<'a> {
struct PaimonReadBuilder<'a> {
table: &'a Table,
read_type: Option<Vec<DataField>>,
+ /// Deferred projection column names. Resolved to `read_type` lazily at
+ /// scan/read build time so `with_projection` and `with_case_sensitive`
+ /// can be called in any order. Mutually exclusive with `read_type`
+ /// ("last projection setter wins").
+ projection_names: Option<Vec<String>>,
filter: NormalizedFilter,
limit: Option<usize>,
row_ranges: Option<Vec<RowRange>>,
+ case_sensitive: bool,
}
impl<'a> PaimonReadBuilder<'a> {
@@ -260,25 +298,48 @@ impl<'a> PaimonReadBuilder<'a> {
Self {
table,
read_type: None,
+ projection_names: None,
filter: NormalizedFilter::default(),
limit: None,
row_ranges: None,
+ case_sensitive: true,
}
}
/// Set column projection by name. Output order follows the
caller-specified order.
- /// Unknown or duplicate names cause this method to fail; an empty list is
a valid
- /// zero-column projection.
+ /// An empty list is a valid zero-column projection.
+ ///
+ /// Name resolution is deferred: the names are stored and resolved against
the
+ /// schema at scan/read build time using the case sensitivity effective
then,
+ /// so `with_projection` and `with_case_sensitive` are order-independent.
As a
+ /// convenience, columns that cannot match under any case sensitivity (a
clear
+ /// typo) are rejected here; case-dependent errors (case-only matches,
+ /// duplicates, or ambiguity) surface from [`new_read`](Self::new_read).
pub fn with_projection(&mut self, columns: &[&str]) -> Result<&mut Self> {
- let projection_names = columns.iter().map(|c|
(*c).to_string()).collect::<Vec<_>>();
- self.read_type =
Some(self.resolve_projected_fields(&projection_names)?);
+ let projection_names: Vec<String> = columns.iter().map(|c|
(*c).to_string()).collect();
+ validate_projection_possible(
+ self.table.identifier().full_name(),
+ self.table.schema.fields(),
+ &projection_names,
+ )?;
+ self.projection_names = Some(projection_names);
+ // A names projection supersedes any previously set read type.
+ self.read_type = None;
Ok(self)
}
+ /// Set whether column-name matching is case-sensitive. Defaults to `true`.
+ pub fn with_case_sensitive(&mut self, case_sensitive: bool) -> &mut Self {
+ self.case_sensitive = case_sensitive;
+ self
+ }
+
/// Set the full read type, including nested field pruning or
connector-defined
/// logical read types such as Variant extractions.
pub fn with_read_type(&mut self, read_type: Vec<DataField>) -> &mut Self {
self.read_type = Some(read_type);
+ // An explicit read type supersedes any pending names projection.
+ self.projection_names = None;
self
}
@@ -360,10 +421,20 @@ impl<'a> PaimonReadBuilder<'a> {
}
/// Create a table scan. Call [TableScan::plan] to get splits.
+ ///
+ /// Projection names are resolved here on a best-effort basis: the resolved
+ /// read type only feeds the data-evolution `projected_read_field_ids`
+ /// planning optimization, and a scan does not surface `Result`. If a
pending
+ /// names projection fails to resolve (unknown/ambiguous/duplicate column),
+ /// the scan conservatively plans with no projection pushdown (correct,
just
+ /// less selective); the same projection resolution runs in `new_read`,
which
+ /// does return `Result` and surfaces the error to the caller. This keeps
the
+ /// scan infallible without silently discarding an error that a read would
hit.
pub fn new_scan(&self) -> TableScan<'a> {
let partition_filter =
self.filter.partition_predicate.clone().map(|pred| {
PartitionFilter::from_predicate(pred,
&self.table.schema().partition_fields())
});
+ let read_type = self.resolve_read_type().unwrap_or(None);
TableScan::new(
self.table,
partition_filter,
@@ -372,7 +443,7 @@ impl<'a> PaimonReadBuilder<'a> {
self.limit,
self.row_ranges.clone(),
)
-
.with_projected_read_field_ids(projected_read_field_ids(&self.read_type))
+ .with_projected_read_field_ids(projected_read_field_ids(&read_type))
}
/// Create a table read for consuming splits (e.g. from a scan plan).
@@ -380,9 +451,9 @@ impl<'a> PaimonReadBuilder<'a> {
// Fail closed at read construction so bindings that short-circuit
before
// `to_arrow` (e.g. an empty-splits fast path) can't bypass the guard.
CoreOptions::new(self.table.schema.options()).ensure_read_authorized()?;
- let read_type = match &self.read_type {
+ let read_type = match self.resolve_read_type()? {
None => self.table.schema.fields().to_vec(),
- Some(fields) => fields.clone(),
+ Some(fields) => fields,
};
// Pass the FULL data predicate through (including `And`/`Or`/`Not`).
@@ -395,6 +466,19 @@ impl<'a> PaimonReadBuilder<'a> {
))
}
+ /// Resolve the effective read type, deferring projection name resolution
to
+ /// the case sensitivity effective at build time (order-independent with
+ /// `with_case_sensitive`).
+ fn resolve_read_type(&self) -> Result<Option<Vec<DataField>>> {
+ if let Some(read_type) = &self.read_type {
+ return Ok(Some(read_type.clone()));
+ }
+ if let Some(names) = &self.projection_names {
+ return Ok(Some(self.resolve_projected_fields(names)?));
+ }
+ Ok(None)
+ }
+
pub(super) fn resolve_projected_fields(
&self,
projection_names: &[String],
@@ -403,27 +487,86 @@ impl<'a> PaimonReadBuilder<'a> {
self.table.identifier().full_name(),
self.table.schema.fields(),
projection_names,
+ self.case_sensitive,
)
}
}
+/// Best-effort early validation for `with_projection`: reject only columns
that
+/// cannot match the schema under *any* case sensitivity, so an obvious typo
+/// (e.g. `foo`) fails fast at call time. Case-dependent outcomes (a name that
+/// matches only case-insensitively, or a case-fold ambiguity) are left to the
+/// final resolution in `new_scan`/`new_read`, which uses the effective
+/// `case_sensitive` — keeping `with_projection` and `with_case_sensitive`
+/// order-independent.
+pub(super) fn validate_projection_possible(
+ full_name: String,
+ fields: &[DataField],
+ projection_names: &[String],
+) -> Result<()> {
+ // Fold the schema names once (O(fields)) so validation is O(fields +
+ // projections) rather than scanning the whole schema per projected name.
+ let folded_names: HashSet<String> = fields
+ .iter()
+ .map(|f| f.name().to_ascii_lowercase())
+ .collect();
+ for name in projection_names {
+ if name == crate::spec::ROW_ID_FIELD_NAME {
+ continue;
+ }
+ if !folded_names.contains(&name.to_ascii_lowercase()) {
+ return Err(Error::ColumnNotExist {
+ full_name: full_name.clone(),
+ column: name.clone(),
+ });
+ }
+ }
+ Ok(())
+}
+
pub(super) fn resolve_projected_fields(
full_name: String,
fields: &[DataField],
projection_names: &[String],
+ case_sensitive: bool,
) -> Result<Vec<DataField>> {
if projection_names.is_empty() {
return Ok(Vec::new());
}
- let field_map: HashMap<&str, &DataField> =
- fields.iter().map(|field| (field.name(), field)).collect();
+ // Build the name index once (O(fields)) so resolution is O(fields +
+ // projections) rather than scanning the whole schema per projected name.
+ // Case-sensitive: exact name -> field. Case-insensitive: ASCII-folded name
+ // -> the unique field, or `None` when two or more fields collide under
+ // folding (ambiguous, mirroring Spark's `AMBIGUOUS` behavior).
+ let sensitive_index: HashMap<&str, &DataField> = if case_sensitive {
+ fields.iter().map(|f| (f.name(), f)).collect()
+ } else {
+ HashMap::new()
+ };
+ let mut folded_index: HashMap<String, Option<&DataField>> = HashMap::new();
+ if !case_sensitive {
+ for f in fields {
+ folded_index
+ .entry(f.name().to_ascii_lowercase())
+ .and_modify(|slot| *slot = None)
+ .or_insert(Some(f));
+ }
+ }
- let mut seen = HashSet::with_capacity(projection_names.len());
+ let mut seen: HashSet<String> =
HashSet::with_capacity(projection_names.len());
let mut resolved = Vec::with_capacity(projection_names.len());
for name in projection_names {
- if !seen.insert(name.as_str()) {
+ // Dedup under the same case sensitivity used for resolution: with
+ // `case-sensitive=false`, `["Name","name"]` must flag a duplicate
rather
+ // than resolve the same field twice.
+ let dedup_key = if case_sensitive {
+ name.clone()
+ } else {
+ name.to_ascii_lowercase()
+ };
+ if !seen.insert(dedup_key) {
return Err(Error::ConfigInvalid {
message: format!("Duplicate projection column '{name}' for
table {full_name}"),
});
@@ -438,13 +581,26 @@ pub(super) fn resolve_projected_fields(
continue;
}
- let field = field_map
- .get(name.as_str())
- .ok_or_else(|| Error::ColumnNotExist {
- full_name: full_name.clone(),
- column: name.clone(),
- })?;
- resolved.push((*field).clone());
+ let field = if case_sensitive {
+ sensitive_index.get(name.as_str()).copied()
+ } else {
+ match folded_index.get(&name.to_ascii_lowercase()) {
+ Some(Some(f)) => Some(*f),
+ Some(None) => {
+ return Err(Error::ConfigInvalid {
+ message: format!(
+ "Ambiguous projection column '{name}' for table
{full_name}: multiple fields match case-insensitively"
+ ),
+ });
+ }
+ None => None,
+ }
+ };
+ let field = field.ok_or_else(|| Error::ColumnNotExist {
+ full_name: full_name.clone(),
+ column: name.clone(),
+ })?;
+ resolved.push(field.clone());
}
Ok(resolved)
@@ -613,6 +769,8 @@ mod tests {
#[test]
fn test_with_projection_validates_unknown_projection() {
+ // A column that cannot match under any case sensitivity is an obvious
+ // typo and is rejected early by with_projection (possible-validation).
let table = simple_table();
let mut builder = ReadBuilder::new(&table);
let err = builder.with_projection(&["missing"]).unwrap_err();
@@ -628,9 +786,11 @@ mod tests {
#[test]
fn test_with_projection_validates_duplicate_projection() {
+ // Resolution is deferred: the duplicate error surfaces at new_read().
let table = simple_table();
let mut builder = ReadBuilder::new(&table);
- let err = builder.with_projection(&["id", "id"]).unwrap_err();
+ builder.with_projection(&["id", "id"]).unwrap();
+ let err = builder.new_read().unwrap_err();
assert!(matches!(
err,
@@ -639,6 +799,242 @@ mod tests {
));
}
+ fn mixed_case_table() -> Table {
+ let file_io = FileIOBuilder::new("file").build().unwrap();
+ let table_schema = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("Name",
DataType::VarChar(VarCharType::new(50).unwrap()))
+ .build()
+ .unwrap(),
+ );
+ Table::new(
+ file_io,
+ Identifier::new("default", "t"),
+ "/tmp/test-read-builder-ci".to_string(),
+ table_schema,
+ None,
+ )
+ }
+
+ #[test]
+ fn test_read_builder_default_case_sensitive_rejects_wrong_case() {
+ // Default (case-sensitive=true): a wrong-case projection must not
resolve.
+ // Resolution is deferred, so the error surfaces at new_read().
+ let table = mixed_case_table();
+ let mut builder = ReadBuilder::new(&table);
+ builder.with_projection(&["NAME"]).unwrap();
+ let err = builder.new_read().unwrap_err();
+ assert!(matches!(err, crate::Error::ColumnNotExist { .. }));
+ }
+
+ #[test]
+ fn test_read_builder_with_case_sensitive_false_resolves_to_canonical() {
+ // After with_case_sensitive(false), a wrong-case projection resolves
to
+ // the canonical schema field name.
+ let table = mixed_case_table();
+ let mut builder = ReadBuilder::new(&table);
+ builder
+ .with_case_sensitive(false)
+ .with_projection(&["nAmE"])
+ .unwrap();
+ let read_type = paimon_builder(&builder)
+ .resolve_read_type()
+ .unwrap()
+ .unwrap();
+ assert_eq!(read_type.len(), 1);
+ assert_eq!(read_type[0].name(), "Name");
+ }
+
+ #[test]
+ fn test_projection_then_case_sensitive_false_is_order_independent() {
+ // with_projection BEFORE with_case_sensitive(false): the wrong-case
name
+ // still resolves case-insensitively because resolution is deferred.
+ let table = mixed_case_table();
+ let mut builder = ReadBuilder::new(&table);
+ builder.with_projection(&["name"]).unwrap();
+ builder.with_case_sensitive(false);
+ let read = builder.new_read().unwrap();
+ assert_eq!(read.read_type().len(), 1);
+ assert_eq!(read.read_type()[0].name(), "Name");
+ }
+
+ #[test]
+ fn test_case_sensitive_false_then_projection_is_order_independent() {
+ // with_case_sensitive(false) BEFORE with_projection: same result.
+ let table = mixed_case_table();
+ let mut builder = ReadBuilder::new(&table);
+ builder.with_case_sensitive(false);
+ builder.with_projection(&["name"]).unwrap();
+ let read = builder.new_read().unwrap();
+ assert_eq!(read.read_type().len(), 1);
+ assert_eq!(read.read_type()[0].name(), "Name");
+ }
+
+ #[test]
+ fn test_default_case_sensitive_wrong_case_errors_at_new_read() {
+ // Default (no with_case_sensitive) + wrong-case projection errors at
read.
+ let table = mixed_case_table();
+ let mut builder = ReadBuilder::new(&table);
+ builder.with_projection(&["name"]).unwrap();
+ let err = builder.new_read().unwrap_err();
+ assert!(matches!(err, crate::Error::ColumnNotExist { .. }));
+ }
+
+ #[test]
+ fn test_new_scan_defers_projection_error_to_new_read() {
+ // Contract: new_scan is infallible — an unresolved projection
degrades to
+ // no projection pushdown (correct, less selective) rather than
erroring,
+ // while the same resolution surfaces the error from new_read.
+ let table = mixed_case_table();
+ let mut builder = ReadBuilder::new(&table);
+ builder.with_projection(&["name"]).unwrap(); // case-only match,
default sensitive
+ let _scan = builder.new_scan(); // must not panic / must succeed
+ let err = builder.new_read().unwrap_err();
+ assert!(matches!(err, crate::Error::ColumnNotExist { .. }));
+ }
+
+ #[test]
+ fn test_case_only_match_passes_early_validation() {
+ // Hybrid: a name matching only case-insensitively (`name` vs schema
+ // `Name`) is NOT rejected by with_projection — its outcome depends on
the
+ // final case sensitivity, so it is deferred. Only names that cannot
match
+ // under any case sensitivity fail early (covered by
+ // test_with_projection_validates_unknown_projection).
+ let table = mixed_case_table();
+ let mut builder = ReadBuilder::new(&table);
+ assert!(builder.with_projection(&["name"]).is_ok());
+ }
+
+ #[test]
+ fn test_read_type_after_projection_wins() {
+ // with_read_type after with_projection: the explicit read type wins
and
+ // the pending names projection is cleared.
+ let table = mixed_case_table();
+ let mut builder = ReadBuilder::new(&table);
+ builder.with_projection(&["id"]).unwrap();
+ builder.with_read_type(vec![DataField::new(
+ 1,
+ "Name".to_string(),
+ DataType::VarChar(VarCharType::new(50).unwrap()),
+ )]);
+ let read_type = paimon_builder(&builder)
+ .resolve_read_type()
+ .unwrap()
+ .unwrap();
+ assert_eq!(read_type.len(), 1);
+ assert_eq!(read_type[0].name(), "Name");
+ }
+
+ #[test]
+ fn test_projection_after_read_type_wins() {
+ // with_projection after with_read_type: the names projection wins and
the
+ // explicit read type is cleared.
+ let table = mixed_case_table();
+ let mut builder = ReadBuilder::new(&table);
+ builder.with_read_type(vec![DataField::new(
+ 1,
+ "Name".to_string(),
+ DataType::VarChar(VarCharType::new(50).unwrap()),
+ )]);
+ builder.with_projection(&["id"]).unwrap();
+ let read_type = paimon_builder(&builder)
+ .resolve_read_type()
+ .unwrap()
+ .unwrap();
+ assert_eq!(read_type.len(), 1);
+ assert_eq!(read_type[0].name(), "id");
+ }
+
+ fn ci_fields() -> Vec<DataField> {
+ vec![
+ DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(
+ 1,
+ "Name".to_string(),
+ DataType::VarChar(VarCharType::new(50).unwrap()),
+ ),
+ ]
+ }
+
+ #[test]
+ fn test_resolve_projection_case_sensitive_exact() {
+ // Default (case-sensitive): exact names resolve, wrong case does not.
+ let out = super::resolve_projected_fields(
+ "db.t".to_string(),
+ &ci_fields(),
+ &["Name".into()],
+ true,
+ )
+ .unwrap();
+ assert_eq!(out.len(), 1);
+ assert_eq!(out[0].name(), "Name");
+
+ let err = super::resolve_projected_fields(
+ "db.t".to_string(),
+ &ci_fields(),
+ &["NAME".into()],
+ true,
+ )
+ .unwrap_err();
+ assert!(matches!(err, crate::Error::ColumnNotExist { .. }));
+ }
+
+ #[test]
+ fn test_resolve_projection_case_insensitive_matches_and_keeps_canonical() {
+ // Case-insensitive: wrong-case request resolves to the canonical
field.
+ let out = super::resolve_projected_fields(
+ "db.t".to_string(),
+ &ci_fields(),
+ &["nAmE".into()],
+ false,
+ )
+ .unwrap();
+ assert_eq!(out.len(), 1);
+ assert_eq!(out[0].name(), "Name", "canonical schema name is
preserved");
+ assert_eq!(out[0].id(), 1);
+ }
+
+ #[test]
+ fn test_resolve_projection_case_insensitive_ambiguous_errors() {
+ let fields = vec![
+ DataField::new(0, "Col".to_string(),
DataType::Int(IntType::new())),
+ DataField::new(1, "col".to_string(),
DataType::Int(IntType::new())),
+ ];
+ let err =
+ super::resolve_projected_fields("db.t".to_string(), &fields,
&["COL".into()], false)
+ .unwrap_err();
+ assert!(matches!(err, crate::Error::ConfigInvalid { .. }));
+ }
+
+ #[test]
+ fn test_resolve_projection_case_insensitive_dedups_by_folded_name() {
+ // With case-insensitive matching, `["Name","name"]` both resolve to
the
+ // canonical `Name` field, so it must be flagged as a duplicate rather
+ // than returning the column twice.
+ let err = super::resolve_projected_fields(
+ "db.t".to_string(),
+ &ci_fields(),
+ &["Name".into(), "name".into()],
+ false,
+ )
+ .unwrap_err();
+ assert!(matches!(err, crate::Error::ConfigInvalid { message }
+ if message.contains("Duplicate projection column")));
+
+ // A single request still resolves cleanly.
+ let out = super::resolve_projected_fields(
+ "db.t".to_string(),
+ &ci_fields(),
+ &["Name".into()],
+ false,
+ )
+ .unwrap();
+ assert_eq!(out.len(), 1);
+ assert_eq!(out[0].name(), "Name");
+ }
+
#[test]
fn test_exact_filter_pushdown_is_true_for_partition_only_filter() {
let table = simple_table();
