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 1cd4527f fix(arrow): map BLOB to LargeBinary (#874)
1cd4527f is described below
commit 1cd4527ffd2b11ec7628176919d4acec86454310
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Sep 18 20:20:22 2026 +0800
fix(arrow): map BLOB to LargeBinary (#874)
---
bindings/go/blob_reader.go | 4 +-
.../datafusion/src/blob_descriptor_functions.rs | 8 +-
crates/integrations/datafusion/src/blob_view.rs | 8 +-
.../datafusion/tests/blob_descriptor_functions.rs | 6 +-
crates/integrations/datafusion/tests/blob_tests.rs | 14 +--
crates/paimon/src/arrow/format/avro.rs | 17 +++-
crates/paimon/src/arrow/format/blob.rs | 85 +++++++----------
crates/paimon/src/arrow/format/row.rs | 101 +++++++++++++++++----
crates/paimon/src/arrow/mod.rs | 34 ++++++-
crates/paimon/src/arrow/residual.rs | 28 +++++-
crates/paimon/src/arrow/shredding/variant.rs | 51 ++++++++---
crates/paimon/src/table/aggregator/numeric.rs | 4 +-
crates/paimon/src/table/blob_resolver.rs | 12 +--
crates/paimon/src/table/data_evolution_reader.rs | 63 ++++++-------
.../table/data_evolution_reader/blob_fallback.rs | 4 +-
.../src/table/dedicated_format_file_writer.rs | 4 +-
crates/paimon/src/table/table_write.rs | 8 +-
crates/paimon/tests/rest_catalog_test.rs | 8 +-
18 files changed, 292 insertions(+), 167 deletions(-)
diff --git a/bindings/go/blob_reader.go b/bindings/go/blob_reader.go
index 529ae9b0..5e2e7ee5 100644
--- a/bindings/go/blob_reader.go
+++ b/bindings/go/blob_reader.go
@@ -102,9 +102,9 @@ func StringBlobMapDescriptors(column arrow.Array, row int)
(map[string][]byte, e
if !ok {
return nil, fmt.Errorf("paimon: BLOB map keys are %T, want
*array.String", m.Keys())
}
- descriptors, ok := m.Items().(*array.Binary)
+ descriptors, ok := m.Items().(*array.LargeBinary)
if !ok {
- return nil, fmt.Errorf("paimon: BLOB map values are %T, want
*array.Binary", m.Items())
+ return nil, fmt.Errorf("paimon: BLOB map values are %T, want
*array.LargeBinary", m.Items())
}
start, end := m.ValueOffsets(row)
result := make(map[string][]byte, end-start)
diff --git a/crates/integrations/datafusion/src/blob_descriptor_functions.rs
b/crates/integrations/datafusion/src/blob_descriptor_functions.rs
index de000412..a49fffb2 100644
--- a/crates/integrations/datafusion/src/blob_descriptor_functions.rs
+++ b/crates/integrations/datafusion/src/blob_descriptor_functions.rs
@@ -18,7 +18,7 @@
use std::sync::Arc;
use datafusion::arrow::array::{
- Array, BinaryArray, BinaryBuilder, BinaryViewArray, LargeBinaryArray,
LargeStringArray,
+ Array, BinaryArray, BinaryViewArray, LargeBinaryArray, LargeBinaryBuilder,
LargeStringArray,
StringArray, StringBuilder, StringViewArray,
};
use datafusion::arrow::datatypes::DataType as ArrowDataType;
@@ -71,7 +71,7 @@ impl ScalarUDFImpl for PathToDescriptorFunc {
}
fn return_type(&self, _arg_types: &[ArrowDataType]) ->
DFResult<ArrowDataType> {
- Ok(ArrowDataType::Binary)
+ Ok(ArrowDataType::LargeBinary)
}
fn invoke_with_args(&self, args: ScalarFunctionArgs) ->
DFResult<ColumnarValue> {
@@ -147,7 +147,7 @@ fn path_scalar(value: ScalarValue) ->
DFResult<ColumnarValue> {
ScalarValue::Null => None,
other => return unexpected_type(PATH_TO_DESCRIPTOR,
&other.data_type()),
};
- Ok(ColumnarValue::Scalar(ScalarValue::Binary(
+ Ok(ColumnarValue::Scalar(ScalarValue::LargeBinary(
path.as_deref().map(serialize_path),
)))
}
@@ -166,7 +166,7 @@ fn path_array(input: &dyn Array) -> DFResult<ColumnarValue>
{
}
fn descriptor_array_from_paths<'a>(paths: impl Iterator<Item = Option<&'a
str>>) -> ColumnarValue {
- let mut builder = BinaryBuilder::new();
+ let mut builder = LargeBinaryBuilder::new();
for path in paths {
match path {
Some(path) => builder.append_value(serialize_path(path)),
diff --git a/crates/integrations/datafusion/src/blob_view.rs
b/crates/integrations/datafusion/src/blob_view.rs
index a91d4db1..b2fed25f 100644
--- a/crates/integrations/datafusion/src/blob_view.rs
+++ b/crates/integrations/datafusion/src/blob_view.rs
@@ -20,7 +20,7 @@ use std::hash::{Hash, Hasher};
use std::sync::Arc;
use datafusion::arrow::array::{
- Array, BinaryBuilder, Int16Array, Int32Array, Int64Array, Int8Array,
LargeStringArray,
+ Array, Int16Array, Int32Array, Int64Array, Int8Array, LargeBinaryBuilder,
LargeStringArray,
StringArray, StringViewArray, UInt16Array, UInt32Array, UInt64Array,
UInt8Array,
};
use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, FieldRef};
@@ -133,13 +133,13 @@ impl ScalarUDFImpl for BlobViewFunc {
}
fn return_type(&self, _arg_types: &[ArrowDataType]) ->
DFResult<ArrowDataType> {
- Ok(ArrowDataType::Binary)
+ Ok(ArrowDataType::LargeBinary)
}
fn return_field_from_args(&self, _args: ReturnFieldArgs) ->
DFResult<FieldRef> {
Ok(Arc::new(Field::new(
FUNCTION_NAME,
- ArrowDataType::Binary,
+ ArrowDataType::LargeBinary,
true,
)))
}
@@ -156,7 +156,7 @@ impl ScalarUDFImpl for BlobViewFunc {
let fields = arrays[1].as_ref();
let row_ids = arrays[2].as_ref();
let mut field_id_cache = HashMap::new();
- let mut builder = BinaryBuilder::new();
+ let mut builder = LargeBinaryBuilder::new();
for row in 0..table_names.len() {
let Some(table_name) = string_at(table_names, row)? else {
diff --git a/crates/integrations/datafusion/tests/blob_descriptor_functions.rs
b/crates/integrations/datafusion/tests/blob_descriptor_functions.rs
index 029dcc4b..33f24db0 100644
--- a/crates/integrations/datafusion/tests/blob_descriptor_functions.rs
+++ b/crates/integrations/datafusion/tests/blob_descriptor_functions.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use datafusion::arrow::array::{Array, BinaryArray, StringArray};
+use datafusion::arrow::array::{Array, LargeBinaryArray, StringArray};
use paimon_datafusion::SQLContext;
const JAVA_V2_HEX: &str =
@@ -61,7 +61,7 @@ async fn
test_blob_descriptor_functions_are_registered_with_aliases_and_java_for
let descriptors = batch
.column(column)
.as_any()
- .downcast_ref::<BinaryArray>()
+ .downcast_ref::<LargeBinaryArray>()
.unwrap();
assert_eq!(to_hex(descriptors.value(0)), JAVA_V2_HEX);
}
@@ -100,7 +100,7 @@ async fn test_blob_descriptor_functions_propagate_nulls() {
let descriptors = batch
.column(1)
.as_any()
- .downcast_ref::<BinaryArray>()
+ .downcast_ref::<LargeBinaryArray>()
.unwrap();
let strings = batch
.column(2)
diff --git a/crates/integrations/datafusion/tests/blob_tests.rs
b/crates/integrations/datafusion/tests/blob_tests.rs
index 361e8ced..47364ead 100644
--- a/crates/integrations/datafusion/tests/blob_tests.rs
+++ b/crates/integrations/datafusion/tests/blob_tests.rs
@@ -21,7 +21,7 @@
mod common;
-use arrow_array::{Array, BinaryArray, Int32Array, RecordBatch};
+use arrow_array::{Array, Int32Array, LargeBinaryArray, RecordBatch};
use common::{assert_sql_error, create_sql_context, create_test_env, exec,
string_value};
use paimon::catalog::Identifier;
use paimon::spec::{BlobDescriptor, BlobViewStruct};
@@ -58,7 +58,7 @@ fn collect_id_name_picture(batches: &[RecordBatch]) ->
Vec<(i32, String, Option<
let pics = batch
.column(2)
.as_any()
- .downcast_ref::<BinaryArray>()
+ .downcast_ref::<LargeBinaryArray>()
.unwrap();
for i in 0..batch.num_rows() {
let pic = if pics.is_null(i) {
@@ -175,12 +175,12 @@ async fn test_blob_multiple_columns() {
let p1 = batch
.column(1)
.as_any()
- .downcast_ref::<BinaryArray>()
+ .downcast_ref::<LargeBinaryArray>()
.unwrap();
let p2 = batch
.column(2)
.as_any()
- .downcast_ref::<BinaryArray>()
+ .downcast_ref::<LargeBinaryArray>()
.unwrap();
assert_eq!(ids.value(0), 1);
assert_eq!(p1.value(0), b"AAA");
@@ -496,7 +496,7 @@ async fn test_blob_with_partition() {
let pics = batch
.column(1)
.as_any()
- .downcast_ref::<BinaryArray>()
+ .downcast_ref::<LargeBinaryArray>()
.unwrap();
for i in 0..batch.num_rows() {
rows.push((ids.value(i), pics.value(i).to_vec()));
@@ -1106,8 +1106,8 @@ async fn test_blob_view_preserves_branch_reference() {
let value = batches[0]
.column(0)
.as_any()
- .downcast_ref::<BinaryArray>()
- .expect("binary blob view")
+ .downcast_ref::<LargeBinaryArray>()
+ .expect("large binary blob view")
.value(0);
let view = BlobViewStruct::deserialize(value).expect("deserialize blob
view");
assert_eq!(view.identifier().full_name(), "test_db.src$branch_b1");
diff --git a/crates/paimon/src/arrow/format/avro.rs
b/crates/paimon/src/arrow/format/avro.rs
index 546d963e..4757f0a0 100644
--- a/crates/paimon/src/arrow/format/avro.rs
+++ b/crates/paimon/src/arrow/format/avro.rs
@@ -25,9 +25,9 @@ use apache_avro::types::Value;
use apache_avro::Reader;
use arrow_array::{
BinaryArray, BooleanArray, Date32Array, Decimal128Array, Float32Array,
Float64Array,
- Int16Array, Int32Array, Int64Array, Int8Array, ListArray, MapArray,
RecordBatch, StringArray,
- StructArray, Time32MillisecondArray, TimestampMicrosecondArray,
TimestampMillisecondArray,
- TimestampNanosecondArray,
+ Int16Array, Int32Array, Int64Array, Int8Array, LargeBinaryArray,
ListArray, MapArray,
+ RecordBatch, StringArray, StructArray, Time32MillisecondArray,
TimestampMicrosecondArray,
+ TimestampMillisecondArray, TimestampNanosecondArray,
};
use arrow_buffer::{BooleanBuffer, NullBuffer, OffsetBuffer, ScalarBuffer};
use arrow_schema::SchemaRef;
@@ -322,13 +322,20 @@ fn build_column(
.collect();
Arc::new(arr)
}
- DataType::Binary(_) | DataType::VarBinary(_) | DataType::Blob(_) => {
+ DataType::Binary(_) | DataType::VarBinary(_) => {
let values: Vec<Option<&[u8]>> = (0..num_rows)
.map(|i| get_field_at(&records[i],
idx).and_then(value_as_bytes))
.collect();
let arr: BinaryArray = values.into_iter().collect();
Arc::new(arr)
}
+ DataType::Blob(_) => {
+ let values: Vec<Option<&[u8]>> = (0..num_rows)
+ .map(|i| get_field_at(&records[i],
idx).and_then(value_as_bytes))
+ .collect();
+ let arr: LargeBinaryArray = values.into_iter().collect();
+ Arc::new(arr)
+ }
DataType::Date(_) => {
let arr: Date32Array = (0..num_rows)
.map(|i| {
@@ -1110,7 +1117,7 @@ mod tests {
vec![("b", av_null())],
]);
let col = build_column(&records, "b",
&DataType::Blob(BlobType::new()), 2).unwrap();
- let arr = col.as_any().downcast_ref::<BinaryArray>().unwrap();
+ let arr = col.as_any().downcast_ref::<LargeBinaryArray>().unwrap();
assert_eq!(arr.value(0), &[0xDE, 0xAD]);
assert!(arr.is_null(1));
}
diff --git a/crates/paimon/src/arrow/format/blob.rs
b/crates/paimon/src/arrow/format/blob.rs
index f03db782..79f2a0ca 100644
--- a/crates/paimon/src/arrow/format/blob.rs
+++ b/crates/paimon/src/arrow/format/blob.rs
@@ -21,11 +21,11 @@ use crate::io::{FileRead, FileWrite};
use crate::spec::{BlobDescriptor, DataField, DataType};
use crate::table::{ArrowRecordBatchStream, RowRange};
use crate::Error;
-use arrow_array::builder::{BinaryBuilder, ListBuilder};
+use arrow_array::builder::{LargeBinaryBuilder, ListBuilder};
use arrow_array::{
Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array,
Int16Array,
- Int32Array, Int64Array, Int8Array, MapArray, RecordBatch,
RecordBatchOptions, StringArray,
- StructArray, Time32MillisecondArray,
+ Int32Array, Int64Array, Int8Array, LargeBinaryArray, MapArray,
RecordBatch, RecordBatchOptions,
+ StringArray, StructArray, Time32MillisecondArray,
};
use arrow_buffer::{BooleanBuffer, NullBuffer, OffsetBuffer, ScalarBuffer};
use arrow_schema::DataType as ArrowDataType;
@@ -322,7 +322,7 @@ pub(crate) fn build_blob_batch(
target_schema: &Arc<arrow_schema::Schema>,
values: Vec<BlobReadValue>,
) -> crate::Result<RecordBatch> {
- let mut builder = BinaryBuilder::new();
+ let mut builder = LargeBinaryBuilder::new();
for value in values {
match value {
BlobReadValue::Value(bytes) =>
builder.append_value(bytes.as_ref()),
@@ -352,13 +352,13 @@ pub(crate) fn build_blob_array_batch(
other => {
return Err(Error::UnexpectedError {
message: format!(
- "Expected Array<Blob> to map to Arrow List<Binary>, got
{other:?}"
+ "Expected Array<Blob> to map to Arrow List<LargeBinary>,
got {other:?}"
),
source: None,
});
}
};
- let mut builder =
ListBuilder::new(BinaryBuilder::new()).with_field(element_field);
+ let mut builder =
ListBuilder::new(LargeBinaryBuilder::new()).with_field(element_field);
for value in values {
match value {
BlobReadValue::Array(elements) => {
@@ -407,7 +407,6 @@ pub(crate) fn build_blob_map_batch(
let mut keys = Vec::new();
let mut blobs = Vec::new();
- let mut blob_data_length = 0u64;
let mut offsets = vec![0i32];
let mut validity = Vec::with_capacity(values.len());
for value in values {
@@ -430,13 +429,6 @@ pub(crate) fn build_blob_map_batch(
source: None,
})?;
for (key, blob) in entries {
- if let Some(blob) = &blob {
- blob_data_length = checked_arrow_binary_data_length(
- blob_data_length,
- blob.len() as u64,
- "MAP<X, BLOB> batch value data",
- )?;
- }
keys.push(key);
blobs.push(blob);
}
@@ -456,7 +448,7 @@ pub(crate) fn build_blob_map_batch(
}
let key_array = decode_blob_map_keys(&keys, key_type)?;
- let value_array = Arc::new(BinaryArray::from_iter(
+ let value_array = Arc::new(LargeBinaryArray::from_iter(
blobs.iter().map(|value| value.as_deref()),
)) as ArrayRef;
let entries = StructArray::try_new(entry_fields.clone(), vec![key_array,
value_array], None)
@@ -926,19 +918,6 @@ fn parse_blob_array_layout(
})
}
-fn validate_inline_blob_array_data_length(layout: &BlobArrayLayout) ->
crate::Result<()> {
- let data_length = layout.element_data_range.end -
layout.element_data_range.start;
- if data_length > i32::MAX as u64 {
- return Err(Error::DataInvalid {
- message: format!(
- "ARRAY<BLOB> inline element data is too large for Arrow
Binary: {data_length} bytes"
- ),
- source: None,
- });
- }
- Ok(())
-}
-
fn decode_blob_array_metadata(
layout: BlobArrayLayout,
index_bytes: &[u8],
@@ -1020,8 +999,7 @@ async fn read_inline_blob_array_entry(
reader: &dyn FileRead,
payload_range: Range<u64>,
) -> crate::Result<BlobReadValue> {
- let preflight_layout = read_blob_array_layout(reader,
payload_range.clone()).await?;
- validate_inline_blob_array_data_length(&preflight_layout)?;
+ let _preflight_layout = read_blob_array_layout(reader,
payload_range.clone()).await?;
let payload = read_blob_entry(reader,
blob_entry_range(&payload_range)).await?;
let payload_length = validate_blob_array_payload_range(&payload_range)?;
@@ -1042,7 +1020,6 @@ async fn read_inline_blob_array_entry(
&payload[..BLOB_ARRAY_HEADER_SIZE as usize],
&payload[index_length_position..],
)?;
- validate_inline_blob_array_data_length(&layout)?;
let index_start = (layout.element_index_range.start - payload_range.start)
as usize;
let index_end = (layout.element_index_range.end - payload_range.start) as
usize;
let metadata = decode_blob_array_metadata(layout,
&payload[index_start..index_end])?;
@@ -1289,9 +1266,6 @@ async fn read_blob_map_entry(
source: None,
});
}
- if !descriptor_mode {
- checked_arrow_binary_data_length(0, total_value_length, "MAP<X, BLOB>
inline value data")?;
- }
if blob_map_key_uses_binary_offsets(key_type) {
checked_arrow_binary_data_length(0, key_data_length, "MAP<X, BLOB> key
data")?;
}
@@ -1789,9 +1763,9 @@ impl FormatFileWriter for BlobFormatWriter {
let col = batch
.column(0)
.as_any()
- .downcast_ref::<arrow_array::BinaryArray>()
+ .downcast_ref::<arrow_array::LargeBinaryArray>()
.ok_or_else(|| Error::DataInvalid {
- message: "BlobFormatWriter expects a single Binary
column".to_string(),
+ message: "BlobFormatWriter expects a single LargeBinary
column".to_string(),
source: None,
})?;
@@ -2243,20 +2217,19 @@ mod tests {
}
#[tokio::test]
- async fn
test_inline_blob_map_reader_rejects_oversized_data_before_entry_read() {
+ async fn test_inline_blob_map_reader_accepts_large_binary_value_data() {
let value_length = i32::MAX as u64 + 1;
- let (reader, payload_range) = sparse_blob_map_entry(&[1],
&[value_length as i64]);
+ let (reader, payload_range) = sparse_blob_map_entry(&[0],
&[value_length as i64]);
let key_type =
DataType::VarChar(VarCharType::new(VarCharType::MAX_LENGTH).unwrap());
- let error = read_blob_map_entry(&reader, payload_range.clone(), "",
false, &key_type)
+ let _error = read_blob_map_entry(&reader, payload_range.clone(), "",
false, &key_type)
.await
.unwrap_err();
assert!(
- !reader.ranges().contains(&blob_entry_range(&payload_range)),
- "oversized inline MAP<X, BLOB> must be rejected before reading the
complete entry"
+ reader.ranges().contains(&blob_entry_range(&payload_range)),
+ "LargeBinary MAP values must not be rejected by the i32 Arrow
Binary limit"
);
- assert_data_invalid(error, "too large");
}
#[tokio::test]
@@ -2302,10 +2275,13 @@ mod tests {
}
#[test]
- fn test_blob_map_batch_rejects_oversized_binary_data() {
- let error =
- checked_arrow_binary_data_length(i32::MAX as u64, 1, "MAP<X, BLOB>
batch value data")
- .unwrap_err();
+ fn test_blob_map_batch_rejects_oversized_binary_key_data() {
+ let error = checked_arrow_binary_data_length(
+ i32::MAX as u64,
+ 1,
+ "MAP<BINARY, BLOB> batch key data",
+ )
+ .unwrap_err();
assert_data_invalid(error, "too large");
}
@@ -2334,7 +2310,7 @@ mod tests {
}
#[tokio::test]
- async fn
test_inline_blob_array_reader_rejects_oversized_data_before_entry_read() {
+ async fn test_inline_blob_array_reader_accepts_large_binary_element_data()
{
let element_data_length = i32::MAX as u64 + 1;
let element_index = encode_delta_varints_write(&[element_data_length
as i64]);
let payload_length =
@@ -2355,15 +2331,14 @@ mod tests {
),
]);
- let error = read_inline_blob_array_entry(&reader,
payload_range.clone())
+ let _error = read_inline_blob_array_entry(&reader,
payload_range.clone())
.await
.unwrap_err();
assert!(
- !reader.ranges().contains(&blob_entry_range(&payload_range)),
- "oversized inline ARRAY<BLOB> must be rejected before reading the
complete entry"
+ reader.ranges().contains(&blob_entry_range(&payload_range)),
+ "LargeBinary ARRAY elements must not be rejected by the i32 Arrow
Binary limit"
);
- assert_data_invalid(error, "too large");
}
#[tokio::test]
@@ -3126,7 +3101,7 @@ mod tests {
let array = batch
.column(0)
.as_any()
- .downcast_ref::<arrow_array::BinaryArray>()
+ .downcast_ref::<arrow_array::LargeBinaryArray>()
.unwrap();
(0..array.len())
.map(|idx| (!array.is_null(idx)).then(||
array.value(idx).to_vec()))
@@ -3148,7 +3123,7 @@ mod tests {
let values = array.value(row_idx);
let values = values
.as_any()
- .downcast_ref::<arrow_array::BinaryArray>()
+ .downcast_ref::<arrow_array::LargeBinaryArray>()
.unwrap();
Some(
(0..values.len())
@@ -3167,7 +3142,7 @@ mod tests {
let values = array
.values()
.as_any()
- .downcast_ref::<BinaryArray>()
+ .downcast_ref::<LargeBinaryArray>()
.unwrap();
(0..array.len())
.map(|row| {
@@ -3277,6 +3252,7 @@ mod tests {
let lengths_start = payload_range.end - BLOB_MAP_INDEX_LENGTHS_SIZE;
let value_index_start = lengths_start - value_index.len() as u64;
let key_index_start = value_index_start - key_index.len() as u64;
+ let data_start = payload_range.start + BLOB_MAP_HEADER_SIZE;
let mut index_lengths = Vec::with_capacity(BLOB_MAP_INDEX_LENGTHS_SIZE
as usize);
index_lengths.extend_from_slice(&(key_index.len() as
i32).to_le_bytes());
index_lengths.extend_from_slice(&(value_index.len() as
i32).to_le_bytes());
@@ -3288,6 +3264,7 @@ mod tests {
(lengths_start..payload_range.end, Bytes::from(index_lengths)),
(key_index_start..value_index_start, Bytes::from(key_index)),
(value_index_start..lengths_start, Bytes::from(value_index)),
+ (data_start..data_start, Bytes::new()),
]);
(reader, payload_range)
}
diff --git a/crates/paimon/src/arrow/format/row.rs
b/crates/paimon/src/arrow/format/row.rs
index fa00773a..0c65645a 100644
--- a/crates/paimon/src/arrow/format/row.rs
+++ b/crates/paimon/src/arrow/format/row.rs
@@ -31,14 +31,14 @@ use crate::table::{ArrowRecordBatchStream, RowRange};
use crate::Error;
use arrow_array::builder::{
BinaryBuilder, BooleanBuilder, Date32Builder, Decimal128Builder,
Float32Builder,
- Float64Builder, Int16Builder, Int32Builder, Int64Builder, Int8Builder,
StringBuilder,
- Time32MillisecondBuilder, TimestampMicrosecondBuilder,
TimestampMillisecondBuilder,
- TimestampNanosecondBuilder,
+ Float64Builder, Int16Builder, Int32Builder, Int64Builder, Int8Builder,
LargeBinaryBuilder,
+ StringBuilder, Time32MillisecondBuilder, TimestampMicrosecondBuilder,
+ TimestampMillisecondBuilder, TimestampNanosecondBuilder,
};
use arrow_array::{
Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array,
Float32Array,
- Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, ListArray,
MapArray, RecordBatch,
- RecordBatchOptions, StringArray, StructArray, Time32MillisecondArray,
+ Float64Array, Int16Array, Int32Array, Int64Array, Int8Array,
LargeBinaryArray, ListArray,
+ MapArray, RecordBatch, RecordBatchOptions, StringArray, StructArray,
Time32MillisecondArray,
TimestampMicrosecondArray, TimestampMillisecondArray,
TimestampNanosecondArray,
};
use arrow_buffer::{BooleanBuffer, NullBuffer, OffsetBuffer, ScalarBuffer};
@@ -411,10 +411,8 @@ fn validate_arrow_type_for_row_field(
| (ArrowDataType::Float32, DataType::Float(_))
| (ArrowDataType::Float64, DataType::Double(_))
| (ArrowDataType::Utf8, DataType::Char(_) | DataType::VarChar(_))
- | (
- ArrowDataType::Binary,
- DataType::Binary(_) | DataType::VarBinary(_) | DataType::Blob(_),
- )
+ | (ArrowDataType::Binary, DataType::Binary(_) | DataType::VarBinary(_))
+ | (ArrowDataType::LargeBinary, DataType::Blob(_))
| (ArrowDataType::Date32, DataType::Date(_))
| (ArrowDataType::Time32(TimeUnit::Millisecond), DataType::Time(_)) =>
true,
(ArrowDataType::Struct(fields), DataType::Variant(_)) =>
is_variant_arrow_fields(fields),
@@ -671,10 +669,14 @@ fn write_field_value(
.value(row_idx)
.as_bytes(),
),
- DataType::Binary(_) | DataType::VarBinary(_) | DataType::Blob(_) =>
write_bytes(
+ DataType::Binary(_) | DataType::VarBinary(_) => write_bytes(
out,
downcast::<BinaryArray>(array, data_type)?.value(row_idx),
),
+ DataType::Blob(_) => write_bytes(
+ out,
+ downcast::<LargeBinaryArray>(array, data_type)?.value(row_idx),
+ ),
DataType::Variant(_) => {
let row = downcast::<StructArray>(array, data_type)?;
write_variant_struct(out, row, row_idx)?;
@@ -1168,6 +1170,7 @@ enum ColumnBuilder {
F64(Float64Builder),
String(StringBuilder),
Binary(BinaryBuilder),
+ LargeBinary(LargeBinaryBuilder),
Date(Date32Builder),
Time(Time32MillisecondBuilder),
TimestampMs(TimestampMillisecondBuilder),
@@ -1207,9 +1210,8 @@ impl ColumnBuilder {
DataType::Float(_) =>
Self::F32(Float32Builder::with_capacity(capacity)),
DataType::Double(_) =>
Self::F64(Float64Builder::with_capacity(capacity)),
DataType::Char(_) | DataType::VarChar(_) =>
Self::String(StringBuilder::new()),
- DataType::Binary(_) | DataType::VarBinary(_) | DataType::Blob(_)
=> {
- Self::Binary(BinaryBuilder::new())
- }
+ DataType::Binary(_) | DataType::VarBinary(_) =>
Self::Binary(BinaryBuilder::new()),
+ DataType::Blob(_) => Self::LargeBinary(LargeBinaryBuilder::new()),
DataType::Variant(_) => Self::Row {
fields: variant_arrow_fields(),
columns: vec![
@@ -1308,6 +1310,7 @@ impl ColumnBuilder {
Self::F64(b) => b.append_null(),
Self::String(b) => b.append_null(),
Self::Binary(b) => b.append_null(),
+ Self::LargeBinary(b) => b.append_null(),
Self::Date(b) => b.append_null(),
Self::Time(b) => b.append_null(),
Self::TimestampMs(b) => b.append_null(),
@@ -1363,9 +1366,10 @@ impl ColumnBuilder {
})?;
b.append_value(value);
}
- (Self::Binary(b), DataType::Binary(_) | DataType::VarBinary(_) |
DataType::Blob(_)) => {
+ (Self::Binary(b), DataType::Binary(_) | DataType::VarBinary(_)) =>
{
b.append_value(input.read_bytes()?);
}
+ (Self::LargeBinary(b), DataType::Blob(_)) =>
b.append_value(input.read_bytes()?),
(Self::Date(b), DataType::Date(_)) =>
b.append_value(input.read_i32()?),
(Self::Time(b), DataType::Time(_)) =>
b.append_value(input.read_i32()?),
(Self::TimestampMs(b), DataType::Timestamp(_) |
DataType::LocalZonedTimestamp(_)) => {
@@ -1492,6 +1496,7 @@ impl ColumnBuilder {
Self::F64(mut b) => Arc::new(b.finish()),
Self::String(mut b) => Arc::new(b.finish()),
Self::Binary(mut b) => Arc::new(b.finish()),
+ Self::LargeBinary(mut b) => Arc::new(b.finish()),
Self::Date(mut b) => Arc::new(b.finish()),
Self::Time(mut b) => Arc::new(b.finish()),
Self::TimestampMs(mut b) => Arc::new(b.finish()),
@@ -2335,9 +2340,9 @@ mod tests {
use crate::btree::test_util::BytesFileRead;
use crate::io::FileIOBuilder;
use crate::spec::{
- ArrayType, BigIntType, BooleanType, DataType, DateType, Datum,
DecimalType, DoubleType,
- FloatType, IntType, MapType, MultisetType, Predicate,
PredicateOperator, RowType, TimeType,
- TimestampType, VarBinaryType, VarCharType, VariantType,
+ ArrayType, BigIntType, BlobType, BooleanType, DataType, DateType,
Datum, DecimalType,
+ DoubleType, FloatType, IntType, MapType, MultisetType, Predicate,
PredicateOperator,
+ RowType, TimeType, TimestampType, VarBinaryType, VarCharType,
VariantType,
};
use crate::variant::GenericVariant;
use futures::TryStreamExt;
@@ -2636,6 +2641,68 @@ mod tests {
assert_eq!(names.value(1), "ccc");
}
+ #[tokio::test]
+ async fn row_writer_reader_preserves_binary_and_large_binary_types() {
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let path = "memory:/row-binary-types/data.row";
+ let fields = vec![
+ DataField::new(
+ 0,
+ "bytes".to_string(),
+ DataType::VarBinary(
+ VarBinaryType::try_new(true,
VarBinaryType::MAX_LENGTH).unwrap(),
+ ),
+ ),
+ DataField::new(1, "blob".to_string(),
DataType::Blob(BlobType::new())),
+ ];
+ let schema = build_target_arrow_schema(&fields).unwrap();
+ let output = file_io.new_output(path).unwrap();
+ let mut writer = RowFormatWriter::new(&output, schema.clone(),
fields.clone(), 1)
+ .await
+ .unwrap();
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(BinaryArray::from(vec![Some(b"bytes".as_slice()),
None])),
+ Arc::new(LargeBinaryArray::from(vec![Some(b"blob".as_slice()),
None])),
+ ],
+ )
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ Box::new(writer).close().await.unwrap();
+
+ let bytes = file_io.new_input(path).unwrap().read().await.unwrap();
+ let batches = RowFormatReader
+ .read_batch_stream(
+ Box::new(BytesFileRead(bytes.clone())),
+ bytes.len() as u64,
+ &fields,
+ None,
+ Some(8),
+ None,
+ )
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let batch = &batches[0];
+ let binary = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ let blob = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap();
+ assert_eq!(binary.value(0), b"bytes");
+ assert!(binary.is_null(1));
+ assert_eq!(blob.value(0), b"blob");
+ assert!(blob.is_null(1));
+ }
+
#[tokio::test]
async fn row_writer_reader_roundtrip_variant() {
let file_io = FileIOBuilder::new("memory").build().unwrap();
diff --git a/crates/paimon/src/arrow/mod.rs b/crates/paimon/src/arrow/mod.rs
index e30d83e5..1f017aee 100644
--- a/crates/paimon/src/arrow/mod.rs
+++ b/crates/paimon/src/arrow/mod.rs
@@ -52,9 +52,8 @@ pub fn paimon_type_to_arrow(dt: &PaimonDataType) ->
crate::Result<ArrowDataType>
PaimonDataType::Float(_) => ArrowDataType::Float32,
PaimonDataType::Double(_) => ArrowDataType::Float64,
PaimonDataType::VarChar(_) | PaimonDataType::Char(_) =>
ArrowDataType::Utf8,
- PaimonDataType::Binary(_) | PaimonDataType::VarBinary(_) |
PaimonDataType::Blob(_) => {
- ArrowDataType::Binary
- }
+ PaimonDataType::Binary(_) | PaimonDataType::VarBinary(_) =>
ArrowDataType::Binary,
+ PaimonDataType::Blob(_) => ArrowDataType::LargeBinary,
PaimonDataType::Variant(_) => variant_arrow_type(),
PaimonDataType::Date(_) => ArrowDataType::Date32,
PaimonDataType::Time(_) =>
ArrowDataType::Time32(TimeUnit::Millisecond),
@@ -419,9 +418,11 @@ mod tests {
#[test]
fn test_binary_types() {
+ let binary = PaimonDataType::Binary(BinaryType::new(16).unwrap());
let varbinary = PaimonDataType::VarBinary(
VarBinaryType::try_new(true, VarBinaryType::MAX_LENGTH).unwrap(),
);
+ assert_paimon_to_arrow(&binary, &ArrowDataType::Binary);
assert_paimon_to_arrow(&varbinary, &ArrowDataType::Binary);
for arrow in &[
@@ -434,14 +435,37 @@ mod tests {
}
#[test]
- fn test_blob_type_maps_one_way_to_arrow_binary() {
+ fn test_blob_type_maps_one_way_to_arrow_large_binary() {
let blob = PaimonDataType::Blob(BlobType::new());
let varbinary = PaimonDataType::VarBinary(
VarBinaryType::try_new(true, VarBinaryType::MAX_LENGTH).unwrap(),
);
- assert_paimon_to_arrow(&blob, &ArrowDataType::Binary);
+ assert_paimon_to_arrow(&blob, &ArrowDataType::LargeBinary);
assert_arrow_to_paimon(&ArrowDataType::Binary, true, &varbinary);
+
+ let array = PaimonDataType::Array(ArrayType::new(blob.clone()));
+ assert_paimon_to_arrow(
+ &array,
+ &ArrowDataType::List(Arc::new(ArrowField::new(
+ "element",
+ ArrowDataType::LargeBinary,
+ true,
+ ))),
+ );
+
+ let map = PaimonDataType::Map(MapType::new(
+ PaimonDataType::VarChar(VarCharType::string_type()),
+ blob,
+ ));
+ let map_arrow = paimon_type_to_arrow(&map).unwrap();
+ let ArrowDataType::Map(entries, _) = map_arrow else {
+ panic!("expected Arrow Map");
+ };
+ let ArrowDataType::Struct(fields) = entries.data_type() else {
+ panic!("expected Arrow Map entries Struct");
+ };
+ assert_eq!(fields[1].data_type(), &ArrowDataType::LargeBinary);
}
#[test]
diff --git a/crates/paimon/src/arrow/residual.rs
b/crates/paimon/src/arrow/residual.rs
index 5ddeb95d..50d716c4 100644
--- a/crates/paimon/src/arrow/residual.rs
+++ b/crates/paimon/src/arrow/residual.rs
@@ -50,8 +50,9 @@ use crate::Error;
use arrow_array::{
Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Datum as
ArrowDatum, Decimal128Array,
FixedSizeListArray, Float32Array, Float64Array, Int16Array, Int32Array,
Int64Array, Int8Array,
- LargeListArray, ListArray, RecordBatch, Scalar, StringArray,
Time32MillisecondArray,
- TimestampMicrosecondArray, TimestampMillisecondArray,
TimestampNanosecondArray,
+ LargeBinaryArray, LargeListArray, ListArray, RecordBatch, Scalar,
StringArray,
+ Time32MillisecondArray, TimestampMicrosecondArray,
TimestampMillisecondArray,
+ TimestampNanosecondArray,
};
use arrow_ord::cmp::{
eq as arrow_eq, gt as arrow_gt, gt_eq as arrow_gt_eq, lt as arrow_lt,
lt_eq as arrow_lt_eq,
@@ -1047,10 +1048,16 @@ pub(crate) fn literal_scalar_for_arrow_filter(
Datum::String(value) =>
Arc::new(StringArray::new_scalar(value.as_str()).into_inner()),
_ => return Ok(None),
},
- DataType::Binary(_) | DataType::VarBinary(_) | DataType::Blob(_) =>
match literal {
+ DataType::Binary(_) | DataType::VarBinary(_) => match literal {
Datum::Bytes(value) =>
Arc::new(BinaryArray::new_scalar(value.as_slice()).into_inner()),
_ => return Ok(None),
},
+ DataType::Blob(_) => match literal {
+ Datum::Bytes(value) => {
+
Arc::new(LargeBinaryArray::new_scalar(value.as_slice()).into_inner())
+ }
+ _ => return Ok(None),
+ },
DataType::Date(_) => match literal {
Datum::Date(value) =>
Arc::new(Date32Array::new_scalar(*value).into_inner()),
_ => return Ok(None),
@@ -1522,6 +1529,21 @@ mod tests {
);
}
+ #[test]
+ fn test_blob_literal_uses_large_binary_scalar() {
+ let scalar = literal_scalar_for_arrow_filter(
+ &Datum::Bytes(b"blob".to_vec()),
+ &DataType::Blob(crate::spec::BlobType::new()),
+ )
+ .unwrap()
+ .unwrap();
+
+ assert_eq!(
+ scalar.get().0.data_type(),
+ &arrow_schema::DataType::LargeBinary
+ );
+ }
+
#[test]
fn test_in_unconvertible_literal_still_errors() {
// An out-of-range literal must keep the general path's error behavior:
diff --git a/crates/paimon/src/arrow/shredding/variant.rs
b/crates/paimon/src/arrow/shredding/variant.rs
index 0f82c470..85475be5 100644
--- a/crates/paimon/src/arrow/shredding/variant.rs
+++ b/crates/paimon/src/arrow/shredding/variant.rs
@@ -31,9 +31,9 @@ use crate::variant::{
use crate::{Error, Result};
use arrow_array::{
new_null_array, Array, ArrayRef, BinaryArray, BooleanArray, Date32Array,
Decimal128Array,
- Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array,
ListArray,
- RecordBatch, StringArray, StructArray, TimestampMicrosecondArray,
TimestampMillisecondArray,
- TimestampNanosecondArray,
+ Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array,
LargeBinaryArray,
+ ListArray, RecordBatch, StringArray, StructArray,
TimestampMicrosecondArray,
+ TimestampMillisecondArray, TimestampNanosecondArray,
};
use arrow_buffer::{BooleanBuffer, NullBuffer, OffsetBuffer, ScalarBuffer};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Fields,
TimeUnit};
@@ -1002,7 +1002,7 @@ fn array_from_values(values: &[Option<ShreddedValue>],
data_type: &DataType) ->
})
.collect::<Vec<_>>(),
))),
- DataType::VarBinary(_) | DataType::Binary(_) | DataType::Blob(_) => {
+ DataType::VarBinary(_) | DataType::Binary(_) => {
let values = values
.iter()
.map(|value| match value {
@@ -1012,6 +1012,16 @@ fn array_from_values(values: &[Option<ShreddedValue>],
data_type: &DataType) ->
.collect::<Vec<_>>();
Ok(Arc::new(BinaryArray::from(values)))
}
+ DataType::Blob(_) => {
+ let values = values
+ .iter()
+ .map(|value| match value {
+ Some(ShreddedValue::Binary(v)) => Some(v.as_slice()),
+ _ => None,
+ })
+ .collect::<Vec<_>>();
+ Ok(Arc::new(LargeBinaryArray::from(values)))
+ }
DataType::Date(_) => Ok(Arc::new(Date32Array::from(
values
.iter()
@@ -1196,13 +1206,16 @@ fn value_at(array: &dyn Array, row: usize, data_type:
&DataType) -> Result<Optio
.value(row)
.to_string(),
)),
- DataType::VarBinary(_) | DataType::Binary(_) | DataType::Blob(_) => {
- Some(ShreddedValue::Binary(
- downcast_array::<BinaryArray>(array, "Binary")?
- .value(row)
- .to_vec(),
- ))
- }
+ DataType::VarBinary(_) | DataType::Binary(_) =>
Some(ShreddedValue::Binary(
+ downcast_array::<BinaryArray>(array, "Binary")?
+ .value(row)
+ .to_vec(),
+ )),
+ DataType::Blob(_) => Some(ShreddedValue::Binary(
+ downcast_array::<LargeBinaryArray>(array, "LargeBinary")?
+ .value(row)
+ .to_vec(),
+ )),
DataType::Date(_) => Some(ShreddedValue::Date32(
downcast_array::<Date32Array>(array, "Date")?.value(row),
)),
@@ -1323,7 +1336,7 @@ fn null_buffer(validities: Vec<bool>) -> NullBuffer {
mod tests {
use super::*;
use crate::arrow::variant_arrow_type;
- use crate::spec::{variant_extraction_row, IntType, VarCharType,
VariantType};
+ use crate::spec::{variant_extraction_row, BlobType, IntType, VarCharType,
VariantType};
fn variant_array_for_test(values: &[GenericVariant]) -> ArrayRef {
let value_items = values
@@ -1351,6 +1364,20 @@ mod tests {
)
}
+ #[test]
+ fn blob_typed_values_use_large_binary() {
+ let data_type = DataType::Blob(BlobType::new());
+ let values = vec![Some(ShreddedValue::Binary(b"blob".to_vec())), None];
+ let array = array_from_values(&values, &data_type).unwrap();
+
+ assert_eq!(array.data_type(), &ArrowDataType::LargeBinary);
+ assert_eq!(
+ value_at(array.as_ref(), 0, &data_type).unwrap(),
+ Some(ShreddedValue::Binary(b"blob".to_vec()))
+ );
+ assert_eq!(value_at(array.as_ref(), 1, &data_type).unwrap(), None);
+ }
+
#[test]
fn configured_schema_transforms_and_reassembles_variant_batch() {
let logical_fields = vec![
diff --git a/crates/paimon/src/table/aggregator/numeric.rs
b/crates/paimon/src/table/aggregator/numeric.rs
index f093827b..5546be3a 100644
--- a/crates/paimon/src/table/aggregator/numeric.rs
+++ b/crates/paimon/src/table/aggregator/numeric.rs
@@ -1388,8 +1388,8 @@ mod tests {
#[test]
fn test_min_rejects_unsupported_type() {
// Java's `TypeCheckUtils#isComparable` rejects exactly MAP, MULTISET,
ROW,
- // ARRAY, VECTOR, VARIANT and BLOB. BLOB is the interesting one here:
it
- // shares Arrow's `Binary` with BINARY and VARBINARY, which min/max
accepts.
+ // ARRAY, VECTOR, VARIANT and BLOB. BINARY and VARBINARY remain
comparable,
+ // while BLOB is intentionally rejected even though it is represented
as LargeBinary.
let err = MinAgg::new("v",
&DataType::Blob(crate::spec::BlobType::new())).unwrap_err();
assert!(matches!(err, crate::Error::ConfigInvalid { message } if
message.contains("min")));
}
diff --git a/crates/paimon/src/table/blob_resolver.rs
b/crates/paimon/src/table/blob_resolver.rs
index e66e4dcc..4d41aa93 100644
--- a/crates/paimon/src/table/blob_resolver.rs
+++ b/crates/paimon/src/table/blob_resolver.rs
@@ -19,8 +19,8 @@ use crate::arrow::format::blob::DEFAULT_BLOB_READ_PARALLELISM;
use crate::io::{FileIO, FileRead};
use crate::spec::BlobDescriptor;
use crate::Result;
-use arrow_array::builder::BinaryBuilder;
-use arrow_array::{Array, BinaryArray};
+use arrow_array::builder::LargeBinaryBuilder;
+use arrow_array::{Array, LargeBinaryArray};
use bytes::Bytes;
use futures::{stream, StreamExt, TryStreamExt};
use std::collections::HashMap;
@@ -122,7 +122,7 @@ impl BlobReader {
blob_error_with_context(error, &indices,
Some(&uri))
})?,
};
- let mut builder =
BinaryBuilder::with_capacity(entries.len(), 0);
+ let mut builder =
LargeBinaryBuilder::with_capacity(entries.len(), 0);
for (_, descriptor) in &entries {
builder.append_value(
BlobDescriptor::new(
@@ -405,10 +405,10 @@ impl BlobReadLimiter {
/// resolve it by reading the actual data from the referenced
URI+offset+length.
/// Raw data values are passed through unchanged.
pub(crate) async fn resolve_blob_column(
- col: &BinaryArray,
+ col: &LargeBinaryArray,
file_io: &FileIO,
limiter: BlobReadLimiter,
-) -> Result<BinaryArray> {
+) -> Result<LargeBinaryArray> {
let mut needs_resolve = false;
for i in 0..col.len() {
if !col.is_null(i) && BlobDescriptor::is_blob_descriptor(col.value(i))
{
@@ -538,7 +538,7 @@ pub(crate) async fn resolve_blob_column(
}
}
- let mut builder = BinaryBuilder::with_capacity(col.len(), value_capacity);
+ let mut builder = LargeBinaryBuilder::with_capacity(col.len(),
value_capacity);
for cell in cells {
match cell {
ResolvedBlobCell::Null => builder.append_null(),
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index a4480878..c162c7e0 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -35,7 +35,7 @@ use crate::table::schema_manager::SchemaManager;
use crate::table::source::any_range_overlaps_file;
use crate::table::{ArrowRecordBatchStream, RESTEnv, RowRange};
use crate::{DataSplit, Error};
-use arrow_array::{Array, BinaryArray, Int64Array, RecordBatch};
+use arrow_array::{Array, Int64Array, LargeBinaryArray, RecordBatch};
use async_stream::try_stream;
use futures::{StreamExt, TryStreamExt};
use roaring::RoaringBitmap;
@@ -903,8 +903,8 @@ async fn resolve_descriptor_columns_with<F, Fut>(
resolve: F,
) -> crate::Result<RecordBatch>
where
- F: Fn(BinaryArray) -> Fut,
- Fut: Future<Output = crate::Result<BinaryArray>>,
+ F: Fn(LargeBinaryArray) -> Fut,
+ Fut: Future<Output = crate::Result<LargeBinaryArray>>,
{
let schema = batch.schema();
let mut columns = batch.columns().to_vec();
@@ -915,7 +915,7 @@ where
if let Some(bin_col) = batch
.column(idx)
.as_any()
- .downcast_ref::<arrow_array::BinaryArray>()
+ .downcast_ref::<LargeBinaryArray>()
{
descriptor_columns.push((idx, bin_col.clone()));
}
@@ -927,14 +927,15 @@ where
}
let resolve = &resolve;
- let resolved_columns: Vec<(usize, BinaryArray)> =
futures::stream::iter(descriptor_columns)
- .map(move |(idx, column)| {
- let future = resolve(column);
- async move { future.await.map(|resolved| (idx, resolved)) }
- })
- .buffer_unordered(blob_parallelism)
- .try_collect()
- .await?;
+ let resolved_columns: Vec<(usize, LargeBinaryArray)> =
+ futures::stream::iter(descriptor_columns)
+ .map(move |(idx, column)| {
+ let future = resolve(column);
+ async move { future.await.map(|resolved| (idx, resolved)) }
+ })
+ .buffer_unordered(blob_parallelism)
+ .try_collect()
+ .await?;
for (idx, resolved) in resolved_columns {
columns[idx] = Arc::new(resolved);
}
@@ -989,7 +990,7 @@ fn collect_blob_view_structs(
if !blob_view_fields.contains(field.name()) {
continue;
}
- let col = binary_column(batch, idx, field.name())?;
+ let col = large_binary_column(batch, idx, field.name())?;
for row in 0..col.len() {
if col.is_null(row) {
continue;
@@ -1025,8 +1026,8 @@ fn replace_blob_view_columns(
continue;
}
- let col = binary_column(&batch, idx, field.name())?;
- let mut builder = arrow_array::builder::BinaryBuilder::new();
+ let col = large_binary_column(&batch, idx, field.name())?;
+ let mut builder = arrow_array::builder::LargeBinaryBuilder::new();
for row in 0..col.len() {
if col.is_null(row) {
builder.append_null();
@@ -1063,17 +1064,17 @@ fn replace_blob_view_columns(
})
}
-fn binary_column<'a>(
+fn large_binary_column<'a>(
batch: &'a RecordBatch,
idx: usize,
field_name: &str,
-) -> crate::Result<&'a BinaryArray> {
+) -> crate::Result<&'a LargeBinaryArray> {
batch
.column(idx)
.as_any()
- .downcast_ref::<BinaryArray>()
+ .downcast_ref::<LargeBinaryArray>()
.ok_or_else(|| Error::DataInvalid {
- message: format!("blob-view-field '{field_name}' requires a
BinaryArray column"),
+ message: format!("blob-view-field '{field_name}' requires a
LargeBinaryArray column"),
source: None,
})
}
@@ -1142,10 +1143,10 @@ impl BlobViewLookup {
let blob_col = batch
.column(0)
.as_any()
- .downcast_ref::<BinaryArray>()
+ .downcast_ref::<LargeBinaryArray>()
.ok_or_else(|| Error::DataInvalid {
message: format!(
- "Upstream blob field '{}' did not read as
BinaryArray",
+ "Upstream blob field '{}' did not read as
LargeBinaryArray",
field.name()
),
source: None,
@@ -2649,8 +2650,8 @@ mod tests {
CommitMessage, DataSplitBuilder, DeletionFile, Table, TableCommit,
TableRead,
};
use arrow_array::{
- Array, BinaryArray, FixedSizeListArray, Float32Array, Int32Array,
Int64Array, ListArray,
- RecordBatch,
+ Array, FixedSizeListArray, Float32Array, Int32Array, Int64Array,
LargeBinaryArray,
+ ListArray, RecordBatch,
};
use bytes::Bytes;
use futures::TryStreamExt;
@@ -2692,16 +2693,16 @@ mod tests {
#[tokio::test]
async fn test_descriptor_columns_resolve_concurrently_and_preserve_order()
{
let schema = Arc::new(arrow_schema::Schema::new(vec![
- arrow_schema::Field::new("blob_a", arrow_schema::DataType::Binary,
true),
+ arrow_schema::Field::new("blob_a",
arrow_schema::DataType::LargeBinary, true),
arrow_schema::Field::new("id", arrow_schema::DataType::Int32,
false),
- arrow_schema::Field::new("blob_b", arrow_schema::DataType::Binary,
true),
+ arrow_schema::Field::new("blob_b",
arrow_schema::DataType::LargeBinary, true),
]));
let batch = RecordBatch::try_new(
schema,
vec![
- Arc::new(BinaryArray::from(vec![Some(b"a".as_slice())])),
+ Arc::new(LargeBinaryArray::from(vec![Some(b"a".as_slice())])),
Arc::new(Int32Array::from(vec![7])),
- Arc::new(BinaryArray::from(vec![Some(b"b".as_slice())])),
+ Arc::new(LargeBinaryArray::from(vec![Some(b"b".as_slice())])),
],
)
.unwrap();
@@ -2730,7 +2731,7 @@ mod tests {
resolved
.column(0)
.as_any()
- .downcast_ref::<BinaryArray>()
+ .downcast_ref::<LargeBinaryArray>()
.unwrap()
.value(0),
b"a"
@@ -2739,7 +2740,7 @@ mod tests {
resolved
.column(2)
.as_any()
- .downcast_ref::<BinaryArray>()
+ .downcast_ref::<LargeBinaryArray>()
.unwrap()
.value(0),
b"b"
@@ -6530,7 +6531,7 @@ mod tests {
let array = batch
.column(idx)
.as_any()
- .downcast_ref::<BinaryArray>()
+ .downcast_ref::<LargeBinaryArray>()
.unwrap();
(0..array.len())
.map(|row| (!array.is_null(row)).then(||
array.value(row).to_vec()))
@@ -6558,7 +6559,7 @@ mod tests {
return None;
}
let values = array.value(row);
- let values =
values.as_any().downcast_ref::<BinaryArray>().unwrap();
+ let values =
values.as_any().downcast_ref::<LargeBinaryArray>().unwrap();
Some(
(0..values.len())
.map(|idx| {
diff --git a/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
b/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
index c0674c2c..adcab7d5 100644
--- a/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
+++ b/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
@@ -331,7 +331,7 @@ mod tests {
use super::*;
use crate::io::FileRead;
use crate::spec::{BlobType, DataType};
- use arrow_array::{Array, BinaryArray};
+ use arrow_array::{Array, LargeBinaryArray};
use bytes::Bytes;
use std::ops::Range;
use std::sync::atomic::{AtomicUsize, Ordering};
@@ -457,7 +457,7 @@ mod tests {
let values = batch
.column(0)
.as_any()
- .downcast_ref::<BinaryArray>()
+ .downcast_ref::<LargeBinaryArray>()
.unwrap();
assert_eq!(values.value(0), b"new-0");
diff --git a/crates/paimon/src/table/dedicated_format_file_writer.rs
b/crates/paimon/src/table/dedicated_format_file_writer.rs
index 6417b9cf..4f3ee315 100644
--- a/crates/paimon/src/table/dedicated_format_file_writer.rs
+++ b/crates/paimon/src/table/dedicated_format_file_writer.rs
@@ -268,11 +268,11 @@ impl AppendDedicatedFormatFileWriter {
let Some(col) = batch
.column(*column_index)
.as_any()
- .downcast_ref::<arrow_array::BinaryArray>()
+ .downcast_ref::<arrow_array::LargeBinaryArray>()
else {
return Err(crate::Error::DataInvalid {
message: format!(
- "blob-view-field '{field_name}' requires a BinaryArray
value column"
+ "blob-view-field '{field_name}' requires a
LargeBinaryArray value column"
),
source: None,
});
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 2d945184..4fc57a58 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -2282,13 +2282,13 @@ pub(in crate::table) mod tests {
let schema = Arc::new(ArrowSchema::new(vec![
ArrowField::new("id", ArrowDataType::Int32, false),
- ArrowField::new("payload", ArrowDataType::Binary, true),
+ ArrowField::new("payload", ArrowDataType::LargeBinary, true),
]));
let batch = RecordBatch::try_new(
schema,
vec![
Arc::new(Int32Array::from(vec![1, 2, 3])),
- Arc::new(arrow_array::BinaryArray::from(vec![
+ Arc::new(arrow_array::LargeBinaryArray::from(vec![
Some(b"hello" as &[u8]),
None,
Some(b"world"),
@@ -2354,12 +2354,12 @@ pub(in crate::table) mod tests {
);
let batch = RecordBatch::try_new(
Arc::new(ArrowSchema::new(vec![
- ArrowField::new("payload", ArrowDataType::Binary, true),
+ ArrowField::new("payload", ArrowDataType::LargeBinary, true),
ArrowField::new("a", ArrowDataType::Int32, false),
ArrowField::new("b", ArrowDataType::Int32, false),
])),
vec![
- Arc::new(arrow_array::BinaryArray::from(vec![Some(
+ Arc::new(arrow_array::LargeBinaryArray::from(vec![Some(
b"payload" as &[u8],
)])),
Arc::new(Int32Array::from(vec![100])),
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 7deb412f..bb617d05 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -23,7 +23,7 @@
use std::collections::HashMap;
use std::sync::Arc;
-use arrow_array::{Array, BinaryArray, Int32Array, Int64Array, RecordBatch,
StringArray};
+use arrow_array::{Array, Int32Array, Int64Array, LargeBinaryArray,
RecordBatch, StringArray};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as
ArrowSchema};
use axum::http::StatusCode;
use futures::TryStreamExt;
@@ -132,7 +132,7 @@ fn blob_batch(ids: Vec<i32>, names: Vec<&str>, pictures:
Vec<Vec<u8>>) -> Record
let schema = Arc::new(ArrowSchema::new(vec![
ArrowField::new("id", ArrowDataType::Int32, false),
ArrowField::new("name", ArrowDataType::Utf8, true),
- ArrowField::new("picture", ArrowDataType::Binary, true),
+ ArrowField::new("picture", ArrowDataType::LargeBinary, true),
]));
let picture_refs = pictures
.iter()
@@ -143,7 +143,7 @@ fn blob_batch(ids: Vec<i32>, names: Vec<&str>, pictures:
Vec<Vec<u8>>) -> Record
vec![
Arc::new(Int32Array::from(ids)),
Arc::new(StringArray::from(names)),
- Arc::new(BinaryArray::from(picture_refs)),
+ Arc::new(LargeBinaryArray::from(picture_refs)),
],
)
.unwrap()
@@ -180,7 +180,7 @@ fn collect_blob_rows(batches: &[RecordBatch]) -> Vec<(i32,
String, Option<Vec<u8
let pictures = batch
.column(2)
.as_any()
- .downcast_ref::<BinaryArray>()
+ .downcast_ref::<LargeBinaryArray>()
.unwrap();
for row in 0..batch.num_rows() {
rows.push((
