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 f58be6f2 [core] Write BLOB collections and read packed video files
(#1044)
f58be6f2 is described below
commit f58be6f2187eb20335e8d3ed51a48319ac3455cd
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 6 11:44:16 2026 +0800
[core] Write BLOB collections and read packed video files (#1044)
---
crates/paimon/src/arrow/format/blob.rs | 432 +++------------
crates/paimon/src/arrow/format/blob/writer.rs | 613 +++++++++++++++++++++
crates/paimon/src/arrow/format/delta_varint.rs | 97 ++++
crates/paimon/src/arrow/format/mod.rs | 12 +-
crates/paimon/src/arrow/format/video.rs | 514 +++++++++++++++++
crates/paimon/src/spec/core_options.rs | 7 +
crates/paimon/src/spec/mod.rs | 3 +
crates/paimon/src/spec/schema.rs | 31 +-
crates/paimon/src/spec/video_frame_descriptor.rs | 215 ++++++++
crates/paimon/src/table/data_evolution_reader.rs | 15 +-
.../table/data_evolution_reader/blob_fallback.rs | 62 ++-
crates/paimon/src/table/data_evolution_writer.rs | 4 +-
.../src/table/dedicated_format_file_writer.rs | 7 +-
crates/paimon/src/table/managed_blob_writer.rs | 2 +-
crates/paimon/src/table/source.rs | 2 +-
crates/paimon/src/table/table_commit.rs | 2 +-
crates/paimon/src/table/table_scan.rs | 2 +-
crates/paimon/src/table/table_write.rs | 7 +-
crates/paimon/src/table/write_batch_normalize.rs | 286 +++++++++-
.../testdata/video/video-frame-descriptor-v1.hex | 18 +
crates/paimon/testdata/video/video-v1.hex | 18 +
crates/paimon/tests/blob_collection_write_test.rs | 270 +++++++++
22 files changed, 2214 insertions(+), 405 deletions(-)
diff --git a/crates/paimon/src/arrow/format/blob.rs
b/crates/paimon/src/arrow/format/blob.rs
index c1da6905..36d538bb 100644
--- a/crates/paimon/src/arrow/format/blob.rs
+++ b/crates/paimon/src/arrow/format/blob.rs
@@ -15,17 +15,22 @@
// specific language governing permissions and limitations
// under the License.
+use super::delta_varint::{
+ decode_delta_varints, encode_delta_varints as encode_delta_varints_write,
+};
+#[cfg(test)]
+use super::delta_varint::{decode_varint, encode_varint};
use super::metadata_cache::FileMetadataCache;
-use super::{FilePredicates, FormatFileReader, FormatFileWriter,
FormatWriteResult};
+use super::{FilePredicates, FormatFileReader};
use crate::arrow::build_target_arrow_schema;
-use crate::io::{BlobIndexCacheContext, FileRead, FileWrite};
+use crate::io::{BlobIndexCacheContext, FileRead};
use crate::spec::{BlobDescriptor, DataField, DataType};
use crate::table::{ArrowRecordBatchStream, RowRange};
use crate::Error;
use arrow_array::builder::{LargeBinaryBuilder, ListBuilder};
use arrow_array::{
- Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array,
Int16Array,
- Int32Array, Int64Array, Int8Array, LargeBinaryArray, MapArray,
RecordBatch, RecordBatchOptions,
+ ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array,
Int16Array, Int32Array,
+ Int64Array, Int8Array, LargeBinaryArray, MapArray, RecordBatch,
RecordBatchOptions,
StringArray, StructArray, Time32MillisecondArray,
};
use arrow_buffer::{BooleanBuffer, NullBuffer, OffsetBuffer, ScalarBuffer};
@@ -37,6 +42,9 @@ use futures::{StreamExt, TryStreamExt};
use std::ops::Range;
use std::sync::Arc;
+mod writer;
+pub(crate) use writer::BlobFormatWriter;
+
pub(crate) struct BlobFormatReader {
descriptor_mode: bool,
file_path: String,
@@ -1911,6 +1919,15 @@ impl BlobEntry {
next_offset = entry_end;
}
+ if next_offset != data_region_end {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Corrupt blob file: indexed records use {next_offset}
bytes, but data region contains {data_region_end} bytes."
+ ),
+ source: None,
+ });
+ }
+
Ok(entries)
}
}
@@ -2037,138 +2054,6 @@ impl RowSelectionCursor {
}
}
-fn decode_delta_varints(bytes: &[u8]) -> crate::Result<Vec<i64>> {
- let mut values = Vec::new();
- let mut cursor = 0usize;
- let mut previous = 0_i64;
-
- while cursor < bytes.len() {
- let (delta, consumed) = decode_varint(&bytes[cursor..])?;
- cursor += consumed;
-
- let value = if values.is_empty() {
- delta
- } else {
- previous
- .checked_add(delta)
- .ok_or_else(|| Error::DataInvalid {
- message: format!(
- "Blob delta-varint index overflow after previous value
{previous}"
- ),
- source: None,
- })?
- };
- values.push(value);
- previous = value;
- }
-
- Ok(values)
-}
-
-fn decode_varint(bytes: &[u8]) -> crate::Result<(i64, usize)> {
- let mut value = 0_u64;
- let mut shift = 0_u32;
-
- for (idx, byte) in bytes.iter().copied().enumerate() {
- value |= u64::from(byte & 0x7f) << shift;
- if (byte & 0x80) == 0 {
- let decoded = ((value >> 1) as i64) ^ (-((value & 1) as i64));
- return Ok((decoded, idx + 1));
- }
-
- shift += 7;
- if shift > 63 {
- return Err(Error::DataInvalid {
- message: "Blob delta-varint index overflow".to_string(),
- source: None,
- });
- }
- }
-
- Err(Error::DataInvalid {
- message: "Unexpected end of blob delta-varint index".to_string(),
- source: None,
- })
-}
-
-// --- Blob Format Writer ---
-
-pub(crate) struct BlobFormatWriter {
- writer: Box<dyn FileWrite>,
- file_io: Option<crate::io::FileIO>,
- bytes_written: u64,
- lengths: Vec<i64>,
-}
-
-impl BlobFormatWriter {
- pub(crate) async fn new(
- output: &crate::io::OutputFile,
- file_io: Option<crate::io::FileIO>,
- ) -> crate::Result<Self> {
- let writer = output.writer().await?;
- Ok(Self {
- writer,
- file_io,
- bytes_written: 0,
- lengths: Vec::new(),
- })
- }
-
- /// Append one managed BLOB and return the payload range in this pack.
- /// The range excludes the four-byte entry magic and twelve-byte trailer,
- /// matching Java `BlobFormatWriter`'s descriptor callback.
- pub(crate) async fn write_managed_value(&mut self, value: &[u8]) ->
crate::Result<(i64, i64)> {
- let start = self.bytes_written;
- let schema =
Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
- "blob",
- ArrowDataType::LargeBinary,
- false,
- )]));
- let batch = RecordBatch::try_new(
- schema,
- vec![Arc::new(LargeBinaryArray::from(vec![Some(value)]))],
- )
- .map_err(|error| Error::DataInvalid {
- message: format!("Failed to build managed BLOB input: {error}"),
- source: Some(Box::new(error)),
- })?;
- self.write(&batch).await?;
- let entry_len =
- self.bytes_written
- .checked_sub(start)
- .ok_or_else(|| Error::DataInvalid {
- message: "Managed BLOB writer position moved
backwards".to_string(),
- source: None,
- })?;
- let payload_len =
- entry_len
- .checked_sub(BLOB_ENTRY_OVERHEAD)
- .ok_or_else(|| Error::DataInvalid {
- message: "Managed BLOB entry is shorter than its
framing".to_string(),
- source: None,
- })?;
- let offset =
- start
- .checked_add(BLOB_INLINE_HEADER_SIZE)
- .ok_or_else(|| Error::DataInvalid {
- message: "Managed BLOB payload offset overflows
u64".to_string(),
- source: None,
- })?;
- Ok((
- i64::try_from(offset).map_err(|error| Error::DataInvalid {
- message: "Managed BLOB payload offset exceeds i64".to_string(),
- source: Some(Box::new(error)),
- })?,
- i64::try_from(payload_len).map_err(|error| Error::DataInvalid {
- message: "Managed BLOB payload length exceeds i64".to_string(),
- source: Some(Box::new(error)),
- })?,
- ))
- }
-}
-
-const BLOB_WRITE_BUFFER_SIZE: u64 = 8 * 1024 * 1024; // 8 MB
-
fn checked_blob_entry_length(payload_len: u64) -> crate::Result<i64> {
let entry_length = payload_len
.checked_add(BLOB_ENTRY_OVERHEAD)
@@ -2186,212 +2071,6 @@ fn checked_blob_entry_length(payload_len: u64) ->
crate::Result<i64> {
})
}
-#[async_trait]
-impl FormatFileWriter for BlobFormatWriter {
- async fn write(&mut self, batch: &RecordBatch) -> crate::Result<()> {
- if batch.num_rows() == 0 {
- return Ok(());
- }
-
- let col = batch
- .column(0)
- .as_any()
- .downcast_ref::<arrow_array::LargeBinaryArray>()
- .ok_or_else(|| Error::DataInvalid {
- message: "BlobFormatWriter expects a single LargeBinary
column".to_string(),
- source: None,
- })?;
-
- for row_idx in 0..col.len() {
- if col.is_null(row_idx) {
- self.lengths.push(-1);
- continue;
- }
-
- let value = col.value(row_idx);
-
- if BlobDescriptor::is_blob_descriptor(value) {
- let desc = BlobDescriptor::deserialize(value)?;
- let range = desc.range_spec()?;
-
- let file_io = self.file_io.as_ref().ok_or_else(||
Error::DataInvalid {
- message:
- "BlobFormatWriter received a BlobDescriptor but has no
FileIO to resolve it"
- .to_string(),
- source: None,
- })?;
- let input = crate::io::uri_reader::UriInput::new(file_io,
desc.uri())?;
- let offset = range.offset();
- let payload_len = match range.length() {
- Some(length) => length,
- None => input
- .size()
- .await
- .map_err(|e| Error::UnexpectedError {
- message: format!(
- "Failed to read metadata for BlobDescriptor
'{}': {e}",
-
crate::io::uri_reader::sanitize_blob_uri(desc.uri())
- ),
- source: Some(Box::new(e)),
- })?
- .saturating_sub(offset),
- };
- let end = offset
- .checked_add(payload_len)
- .ok_or_else(|| Error::DataInvalid {
- message: format!(
- "BlobDescriptor range overflows u64:
offset={offset}, length={payload_len}"
- ),
- source: None,
- })?;
- let entry_length = checked_blob_entry_length(payload_len)?;
- let entry_length_u64 = entry_length as u64;
- let bytes_written = self
- .bytes_written
- .checked_add(entry_length_u64)
- .ok_or_else(|| Error::DataInvalid {
- message: format!(
- "Blob file size overflows u64: current_size={},
entry_length={entry_length_u64}",
- self.bytes_written
- ),
- source: None,
- })?;
- let reader = if payload_len == 0 {
- None
- } else {
- Some(input.reader_for_range(offset..end).await?)
- };
-
- let mut hasher = crc32fast::Hasher::new();
-
- hasher.update(&BLOB_MAGIC_NUMBER_BYTES);
- self.writer
- .write(Bytes::copy_from_slice(&BLOB_MAGIC_NUMBER_BYTES))
- .await?;
-
- // Stream payload in chunks to avoid loading entire blob into
memory
- if let Some(reader) = reader.as_ref() {
- let mut pos = offset;
- while pos < end {
- let chunk_end =
pos.saturating_add(BLOB_WRITE_BUFFER_SIZE).min(end);
- let chunk =
reader.read(pos..chunk_end).await.map_err(|e| {
- Error::UnexpectedError {
- message: format!(
- "Failed to read BlobDescriptor '{}' range
{pos}..{chunk_end}: {e}",
-
crate::io::uri_reader::sanitize_blob_uri(desc.uri())
- ),
- source: Some(Box::new(e)),
- }
- })?;
- let actual_len = chunk.len() as u64;
- let expected_len = chunk_end - pos;
- if actual_len != expected_len {
- return Err(Error::DataInvalid {
- message: format!(
- "Failed to read BlobDescriptor '{}': short
read for range {pos}..{chunk_end}, expected={expected_len} bytes,
actual={actual_len} bytes",
-
crate::io::uri_reader::sanitize_blob_uri(desc.uri())
- ),
- source: None,
- });
- }
- hasher.update(&chunk);
- self.writer.write(chunk).await?;
- pos = chunk_end;
- }
- }
-
- let entry_length_bytes = entry_length.to_le_bytes();
- hasher.update(&entry_length_bytes);
- self.writer
- .write(Bytes::copy_from_slice(&entry_length_bytes))
- .await?;
-
- self.writer
-
.write(Bytes::copy_from_slice(&hasher.finalize().to_le_bytes()))
- .await?;
-
- self.lengths.push(entry_length);
- self.bytes_written = bytes_written;
- } else {
- let entry_length = (value.len() + BLOB_ENTRY_OVERHEAD as
usize) as i64;
- self.lengths.push(entry_length);
-
- let mut buf = Vec::with_capacity(entry_length as usize);
- let mut hasher = crc32fast::Hasher::new();
-
- hasher.update(&BLOB_MAGIC_NUMBER_BYTES);
- buf.extend_from_slice(&BLOB_MAGIC_NUMBER_BYTES);
-
- hasher.update(value);
- buf.extend_from_slice(value);
-
- let entry_length_bytes = entry_length.to_le_bytes();
- hasher.update(&entry_length_bytes);
- buf.extend_from_slice(&entry_length_bytes);
-
- buf.extend_from_slice(&hasher.finalize().to_le_bytes());
-
- self.writer.write(Bytes::from(buf)).await?;
- self.bytes_written += entry_length as u64;
- }
- }
-
- Ok(())
- }
-
- fn num_bytes(&self) -> usize {
- self.bytes_written as usize
- }
-
- fn in_progress_size(&self) -> usize {
- 0
- }
-
- async fn flush(&mut self) -> crate::Result<()> {
- Ok(())
- }
-
- async fn close(mut self: Box<Self>) -> crate::Result<FormatWriteResult> {
- let index_bytes = encode_delta_varints_write(&self.lengths);
- let index_length = index_bytes.len() as i32;
-
- self.writer.write(Bytes::from(index_bytes)).await?;
- self.writer
- .write(Bytes::copy_from_slice(&index_length.to_le_bytes()))
- .await?;
- self.writer
- .write(Bytes::from_static(&[BLOB_FORMAT_VERSION]))
- .await?;
-
- let total = self.bytes_written + index_length as u64 +
BLOB_FOOTER_SIZE;
- self.writer.close().await?;
- Ok(FormatWriteResult::new(total))
- }
-}
-
-fn encode_delta_varints_write(values: &[i64]) -> Vec<u8> {
- if values.is_empty() {
- return Vec::new();
- }
- let mut encoded = Vec::new();
- let mut previous = 0_i64;
- for (idx, &value) in values.iter().enumerate() {
- let delta = if idx == 0 { value } else { value - previous };
- previous = value;
- encode_varint(delta, &mut encoded);
- }
- encoded
-}
-
-fn encode_varint(value: i64, out: &mut Vec<u8>) {
- let mut remaining = ((value << 1) ^ (value >> 63)) as u64;
- while (remaining & !0x7f) != 0 {
- out.push(((remaining & 0x7f) as u8) | 0x80);
- remaining >>= 7;
- }
- out.push(remaining as u8);
-}
-
#[cfg(test)]
mod tests {
use super::*;
@@ -2591,11 +2270,11 @@ mod tests {
#[tokio::test]
async fn test_index_tail_prefetch_boundaries() {
for rows in [0, 1, 4090, 4091, 4092, 9000] {
- let mut data = vec![0; 8192];
let index = encode_delta_varints_write(&vec![-1; rows]);
assert_eq!(index.len(), rows);
- data.extend_from_slice(&index);
- data.extend_from_slice(&(index.len() as i32).to_le_bytes());
+ // NULL entries have no physical records or padding before the
index.
+ let mut data = index;
+ data.extend_from_slice(&(rows as i32).to_le_bytes());
data.push(BLOB_FORMAT_VERSION);
let size = data.len() as u64;
let reader = TrackingFileRead::new(Bytes::from(data));
@@ -2606,9 +2285,9 @@ mod tests {
.iter()
.all(|entry| matches!(entry, BlobEntry::Null)));
let mut expected = Vec::with_capacity(2);
- expected.push(size - 4096..size);
+ expected.push(size.saturating_sub(4096)..size);
if rows + 5 > 4096 {
- expected.push(8192..size - BLOB_FOOTER_SIZE);
+ expected.push(0..size - BLOB_FOOTER_SIZE);
}
assert_eq!(reader.ranges(), expected);
}
@@ -2728,16 +2407,19 @@ mod tests {
#[tokio::test]
async fn test_index_tail_prefetch_sixteen_cold_files() {
+ use blob_test_utils::BlobFixtureValue::{Null, Value};
+
let mut requests = 0;
let mut bytes = 0;
let mut legacy_bytes = 0;
let mut legacy_requests = 0;
+ let payload = vec![0; 8192 - BLOB_ENTRY_OVERHEAD as usize];
for rows in (1526..).take(16) {
- let index = encode_delta_varints_write(&vec![-1; rows]);
- let mut data = vec![0; 8192];
- data.extend_from_slice(&index);
- data.extend_from_slice(&(index.len() as i32).to_le_bytes());
- data.push(1);
+ // Retain a data region large enough to fill the prefetched tail,
+ // but account for it with a complete, indexed physical record.
+ let mut values = vec![Null; rows];
+ values[0] = Value(&payload);
+ let data =
blob_test_utils::build_blob_file_bytes_with_values(&values);
let size = data.len() as u64;
let reader = TrackingFileRead::new(Bytes::from(data.clone()));
let legacy_reader = TrackingFileRead::new(Bytes::from(data));
@@ -2769,7 +2451,7 @@ mod tests {
.sum::<u64>();
}
assert_eq!(legacy_requests, 32);
- assert_eq!((requests, bytes, legacy_bytes), (16, 65536, 24616));
+ assert_eq!((requests, bytes, legacy_bytes), (16, 65536, 24680));
}
#[tokio::test]
@@ -4461,6 +4143,50 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn test_blob_reader_rejects_unindexed_bytes_before_footer() {
+ use blob_test_utils::BlobFixtureValue::{Null, Placeholder, Value};
+
+ for rows in [
+ vec![Value(b"hello"), Null, Placeholder, Value(b"")],
+ vec![Null, Placeholder],
+ vec![],
+ ] {
+ let valid =
blob_test_utils::build_blob_file_bytes_with_values(&rows);
+ let footer_start = valid.len() - BLOB_FOOTER_SIZE as usize;
+ let index_length =
+ i32::from_le_bytes(valid[footer_start..footer_start +
4].try_into().unwrap())
+ as usize;
+ let index_start = footer_start - index_length;
+ let mut corrupt = valid.clone();
+ // Preserve the complete records (and their CRCs), footer and
index.
+ // Only add bytes which no indexed record accounts for.
+ corrupt.splice(index_start..index_start,
b"unindexed".iter().copied());
+
+ for descriptor_mode in [false, true] {
+ let reader = IndexedBlobReader::open(
+ Box::new(BytesFileRead(Bytes::from(valid.clone()))),
+ valid.len() as u64,
+ "file:/valid.blob".into(),
+ descriptor_mode,
+ )
+ .await
+ .unwrap();
+ assert_eq!(reader.num_rows(), rows.len());
+
+ let result = IndexedBlobReader::open(
+ Box::new(BytesFileRead(Bytes::from(corrupt.clone()))),
+ corrupt.len() as u64,
+ "file:/corrupt.blob".into(),
+ descriptor_mode,
+ )
+ .await;
+ assert!(matches!(result, Err(Error::DataInvalid { message, ..
})
+ if message == format!("Corrupt blob file: indexed records
use {index_start} bytes, but data region contains {} bytes.", index_start +
b"unindexed".len())));
+ }
+ }
+ }
+
#[tokio::test]
async fn test_blob_reader_rejects_truncated_entry() {
let mut file_bytes =
blob_test_utils::build_blob_file_bytes(&basic_blob_rows());
diff --git a/crates/paimon/src/arrow/format/blob/writer.rs
b/crates/paimon/src/arrow/format/blob/writer.rs
new file mode 100644
index 00000000..7aedc97b
--- /dev/null
+++ b/crates/paimon/src/arrow/format/blob/writer.rs
@@ -0,0 +1,613 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Java BlobElementSerializer record layouts, sharing framing and streamed
payload copying.
+
+use super::{
+ checked_blob_entry_length, encode_delta_varints_write,
validate_read_fields, BlobFieldKind,
+ BLOB_ARRAY_MAGIC_NUMBER, BLOB_ARRAY_VERSION, BLOB_FORMAT_VERSION,
BLOB_MAGIC_NUMBER_BYTES,
+ BLOB_MAP_MAGIC_NUMBER, BLOB_MAP_VERSION,
+};
+use crate::arrow::format::{FormatFileWriter, FormatWriteResult};
+use crate::io::{FileIO, FileWrite, OutputFile};
+use crate::spec::{BlobDescriptor, DataField, DataType};
+use crate::{Error, Result};
+use arrow_array::{
+ Array, BinaryArray, BooleanArray, Date32Array, Decimal128Array,
Int16Array, Int32Array,
+ Int64Array, Int8Array, LargeBinaryArray, ListArray, MapArray, RecordBatch,
StringArray,
+ Time32MillisecondArray,
+};
+use async_trait::async_trait;
+use bytes::Bytes;
+use crc32fast::Hasher;
+use std::collections::HashSet;
+
+const COPY_BUFFER_SIZE: u64 = 8 * 1024 * 1024;
+
+pub(crate) struct BlobFormatWriter {
+ writer: Box<dyn FileWrite>,
+ file_io: Option<FileIO>,
+ kind: BlobFieldKind,
+ bytes_written: u64,
+ lengths: Vec<i64>,
+}
+
+fn invalid(message: impl Into<String>) -> Error {
+ Error::DataInvalid {
+ message: message.into(),
+ source: None,
+ }
+}
+
+impl BlobFormatWriter {
+ pub(crate) async fn new(
+ output: &OutputFile,
+ file_io: Option<FileIO>,
+ field: Option<&DataField>,
+ ) -> Result<Self> {
+ let kind = match field {
+ Some(field) =>
validate_read_fields(std::slice::from_ref(field))?.unwrap(),
+ None => BlobFieldKind::Scalar,
+ };
+ if let BlobFieldKind::Map(key_type) = &kind {
+ validate_key_type(key_type)?;
+ }
+ Ok(Self {
+ writer: output.writer().await?,
+ file_io,
+ kind,
+ bytes_written: 0,
+ lengths: Vec::new(),
+ })
+ }
+
+ /// The callback range excludes the common entry magic and trailer.
+ pub(crate) async fn write_managed_value(&mut self, value: &[u8]) ->
Result<(i64, i64)> {
+ if !matches!(self.kind, BlobFieldKind::Scalar) {
+ return Err(invalid("Managed BLOB packs require scalar elements"));
+ }
+ let start = self.bytes_written;
+ let length = self.write_scalar(value).await?;
+ Ok((
+ i64::try_from(start + 4).map_err(|_| invalid("Managed BLOB offset
exceeds i64"))?,
+ length,
+ ))
+ }
+
+ async fn write_bytes(&mut self, bytes: Bytes, hasher: &mut Hasher) ->
Result<()> {
+ let end = self
+ .bytes_written
+ .checked_add(bytes.len() as u64)
+ .ok_or_else(|| invalid("BLOB file size overflows u64"))?;
+ hasher.update(&bytes);
+ self.writer.write(bytes).await?;
+ self.bytes_written = end;
+ Ok(())
+ }
+
+ async fn start_record(&mut self) -> Result<(u64, Hasher)> {
+ let start = self.bytes_written;
+ let mut hasher = Hasher::new();
+ self.write_bytes(
+ Bytes::copy_from_slice(&BLOB_MAGIC_NUMBER_BYTES),
+ &mut hasher,
+ )
+ .await?;
+ Ok((start, hasher))
+ }
+
+ async fn finish_record(&mut self, start: u64, mut hasher: Hasher) ->
Result<()> {
+ let length = checked_blob_entry_length(self.bytes_written - start -
4)?;
+ self.write_bytes(Bytes::copy_from_slice(&length.to_le_bytes()), &mut
hasher)
+ .await?;
+ let crc = hasher.finalize().to_le_bytes();
+ // CRC itself is outside its checksum domain.
+ self.writer.write(Bytes::copy_from_slice(&crc)).await?;
+ self.bytes_written = self
+ .bytes_written
+ .checked_add(4)
+ .ok_or_else(|| invalid("BLOB file size overflows u64"))?;
+ self.lengths.push(length);
+ Ok(())
+ }
+
+ async fn write_scalar(&mut self, value: &[u8]) -> Result<i64> {
+ let (start, mut hasher) = self.start_record().await?;
+ let length = self.write_payload(value, &mut hasher).await?;
+ self.finish_record(start, hasher).await?;
+ Ok(length)
+ }
+
+ /// Resolve descriptors without materializing a complete referenced object.
+ async fn write_payload(&mut self, value: &[u8], hasher: &mut Hasher) ->
Result<i64> {
+ if !BlobDescriptor::is_blob_descriptor(value) {
+ self.write_bytes(Bytes::copy_from_slice(value), hasher)
+ .await?;
+ return i64::try_from(value.len()).map_err(|_| invalid("BLOB
payload exceeds i64"));
+ }
+ let descriptor = BlobDescriptor::deserialize(value)?;
+ let range = descriptor.range_spec()?;
+ let file_io = self.file_io.as_ref().ok_or_else(|| {
+ invalid("BlobFormatWriter received a BlobDescriptor but has no
FileIO to resolve it")
+ })?;
+ let input = crate::io::uri_reader::UriInput::new(file_io,
descriptor.uri())?;
+ let offset = range.offset();
+ let length = match range.length() {
+ Some(length) => length,
+ None => input
+ .size()
+ .await
+ .map_err(|error| Error::UnexpectedError {
+ message: format!(
+ "Failed to read metadata for BlobDescriptor '{}':
{error}",
+
crate::io::uri_reader::sanitize_blob_uri(descriptor.uri())
+ ),
+ source: Some(Box::new(error)),
+ })?
+ .saturating_sub(offset),
+ };
+ let end = offset
+ .checked_add(length)
+ .ok_or_else(|| invalid("BlobDescriptor range overflows u64"))?;
+ let length_i64 = i64::try_from(length).map_err(|_| invalid("BLOB
payload exceeds i64"))?;
+ if length == 0 {
+ return Ok(0);
+ }
+ let reader = input.reader_for_range(offset..end).await?;
+ let mut position = offset;
+ while position < end {
+ let chunk_end = position.saturating_add(COPY_BUFFER_SIZE).min(end);
+ let chunk =
+ reader
+ .read(position..chunk_end)
+ .await
+ .map_err(|error| Error::UnexpectedError {
+ message: format!(
+ "Failed to read BlobDescriptor '{}' range
{position}..{chunk_end}: {error}",
+
crate::io::uri_reader::sanitize_blob_uri(descriptor.uri())
+ ),
+ source: Some(Box::new(error)),
+ })?;
+ if chunk.len() as u64 != chunk_end - position {
+ return Err(invalid(format!("Failed to read BlobDescriptor
'{}': short read for range {position}..{chunk_end}, expected={} bytes,
actual={} bytes",
+
crate::io::uri_reader::sanitize_blob_uri(descriptor.uri()), chunk_end -
position, chunk.len())));
+ }
+ self.write_bytes(chunk, hasher).await?;
+ position = chunk_end;
+ }
+ Ok(length_i64)
+ }
+
+ async fn write_blob_values(
+ &mut self,
+ values: &LargeBinaryArray,
+ hasher: &mut Hasher,
+ ) -> Result<Vec<i64>> {
+ let mut lengths = Vec::with_capacity(values.len());
+ for value in values {
+ lengths.push(match value {
+ None => -1,
+ Some(value) => self.write_payload(value, hasher).await?,
+ });
+ }
+ Ok(lengths)
+ }
+
+ async fn write_index(&mut self, values: &[i64], hasher: &mut Hasher) ->
Result<i32> {
+ let bytes = encode_delta_varints_write(values);
+ let length =
+ i32::try_from(bytes.len()).map_err(|_| invalid("BLOB element index
exceeds i32"))?;
+ self.write_bytes(Bytes::from(bytes), hasher).await?;
+ Ok(length)
+ }
+
+ async fn write_collection(
+ &mut self,
+ values: &LargeBinaryArray,
+ keys: Option<Vec<Vec<u8>>>,
+ ) -> Result<()> {
+ let (start, mut hasher) = self.start_record().await?;
+ let (magic, version) = if keys.is_some() {
+ (BLOB_MAP_MAGIC_NUMBER, BLOB_MAP_VERSION)
+ } else {
+ (BLOB_ARRAY_MAGIC_NUMBER, BLOB_ARRAY_VERSION)
+ };
+ let count =
+ i32::try_from(values.len()).map_err(|_| invalid("BLOB element
count exceeds i32"))?;
+ let mut header = Vec::with_capacity(9);
+ header.extend_from_slice(&magic.to_le_bytes());
+ header.push(version);
+ header.extend_from_slice(&count.to_le_bytes());
+ self.write_bytes(header.into(), &mut hasher).await?;
+ let key_lengths = if let Some(keys) = keys {
+ let mut lengths = Vec::with_capacity(keys.len());
+ for key in keys {
+ lengths.push(
+ i64::try_from(key.len()).map_err(|_| invalid("BLOB MAP key
exceeds i64"))?,
+ );
+ self.write_bytes(key.into(), &mut hasher).await?;
+ }
+ Some(lengths)
+ } else {
+ None
+ };
+ let value_lengths = self.write_blob_values(values, &mut hasher).await?;
+ let key_index_length = if let Some(lengths) = key_lengths {
+ Some(self.write_index(&lengths, &mut hasher).await?)
+ } else {
+ None
+ };
+ let value_index_length = self.write_index(&value_lengths, &mut
hasher).await?;
+ if let Some(length) = key_index_length {
+ self.write_bytes(Bytes::copy_from_slice(&length.to_le_bytes()),
&mut hasher)
+ .await?;
+ }
+ self.write_bytes(
+ Bytes::copy_from_slice(&value_index_length.to_le_bytes()),
+ &mut hasher,
+ )
+ .await?;
+ self.finish_record(start, hasher).await
+ }
+}
+
+#[async_trait]
+impl FormatFileWriter for BlobFormatWriter {
+ async fn write(&mut self, batch: &RecordBatch) -> Result<()> {
+ if batch.num_columns() != 1 {
+ return Err(invalid("BlobFormatWriter requires exactly one
column"));
+ }
+ let column = batch.column(0);
+ let kind = self.kind.clone();
+ for row in 0..batch.num_rows() {
+ if column.is_null(row) {
+ self.lengths.push(-1);
+ continue;
+ }
+ match &kind {
+ BlobFieldKind::Scalar => {
+ self.write_scalar(blob_values(column.as_ref())?.value(row))
+ .await?;
+ }
+ BlobFieldKind::Array => {
+ let list = column
+ .as_any()
+ .downcast_ref::<ListArray>()
+ .ok_or_else(|| invalid("ARRAY<BLOB> writer requires
ListArray"))?;
+
self.write_collection(blob_values(list.value(row).as_ref())?, None)
+ .await?;
+ }
+ BlobFieldKind::Map(key_type) => {
+ let map = column
+ .as_any()
+ .downcast_ref::<MapArray>()
+ .ok_or_else(|| invalid("MAP<X, BLOB> writer requires
MapArray"))?;
+ let entries = map.value(row);
+ // Java validates all keys before starting the record.
+ let keys = serialize_map_keys(entries.column(0).as_ref(),
key_type)?;
+
self.write_collection(blob_values(entries.column(1).as_ref())?, Some(keys))
+ .await?;
+ }
+ }
+ }
+ Ok(())
+ }
+
+ fn num_bytes(&self) -> usize {
+ self.bytes_written as usize
+ }
+ fn in_progress_size(&self) -> usize {
+ 0
+ }
+ async fn flush(&mut self) -> Result<()> {
+ Ok(())
+ }
+ async fn close(mut self: Box<Self>) -> Result<FormatWriteResult> {
+ let index = encode_delta_varints_write(&self.lengths);
+ let length =
+ i32::try_from(index.len()).map_err(|_| invalid("BLOB file index
exceeds i32"))?;
+ self.writer.write(index.into()).await?;
+ self.writer
+ .write(Bytes::copy_from_slice(&length.to_le_bytes()))
+ .await?;
+ self.writer
+ .write(Bytes::from_static(&[BLOB_FORMAT_VERSION]))
+ .await?;
+ self.writer.close().await?;
+ Ok(FormatWriteResult::new(
+ self.bytes_written + length as u64 + 5,
+ ))
+ }
+}
+
+fn blob_values(array: &dyn Array) -> Result<&LargeBinaryArray> {
+ array
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .ok_or_else(|| invalid("BLOB values require LargeBinaryArray"))
+}
+
+fn validate_key_type(key_type: &DataType) -> Result<()> {
+ match key_type {
+ DataType::Boolean(_)
+ | DataType::TinyInt(_)
+ | DataType::SmallInt(_)
+ | DataType::Int(_)
+ | DataType::BigInt(_)
+ | DataType::Date(_)
+ | DataType::Time(_)
+ | DataType::Decimal(_)
+ | DataType::Char(_)
+ | DataType::VarChar(_)
+ | DataType::Binary(_)
+ | DataType::VarBinary(_) => Ok(()),
+ other => Err(Error::Unsupported {
+ message: format!("Unsupported key type for MAP<X, BLOB>:
{other:?}"),
+ }),
+ }
+}
+
+fn serialize_map_keys(array: &dyn Array, key_type: &DataType) ->
Result<Vec<Vec<u8>>> {
+ let mut seen = HashSet::new();
+ let mut keys = Vec::with_capacity(array.len());
+ for row in 0..array.len() {
+ if array.is_null(row) {
+ return Err(invalid("Arrow MAP keys must not be null"));
+ }
+ macro_rules! fixed {
+ ($array:ty) => {
+ array
+ .as_any()
+ .downcast_ref::<$array>()
+ .ok_or_else(|| {
+ invalid("MAP<X, BLOB> key array does not match its
declared type")
+ })?
+ .value(row)
+ .to_le_bytes()
+ .to_vec()
+ };
+ }
+ let key = match key_type {
+ DataType::TinyInt(_) => fixed!(Int8Array),
+ DataType::SmallInt(_) => fixed!(Int16Array),
+ DataType::Int(_) => fixed!(Int32Array),
+ DataType::BigInt(_) => fixed!(Int64Array),
+ DataType::Date(_) => fixed!(Date32Array),
+ DataType::Time(_) => fixed!(Time32MillisecondArray),
+ DataType::Boolean(_) => vec![u8::from(
+ array
+ .as_any()
+ .downcast_ref::<BooleanArray>()
+ .ok_or_else(|| invalid("MAP BOOLEAN key requires
BooleanArray"))?
+ .value(row),
+ )],
+ DataType::Char(_) | DataType::VarChar(_) => array
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .ok_or_else(|| invalid("MAP string key requires StringArray"))?
+ .value(row)
+ .as_bytes()
+ .to_vec(),
+ DataType::Binary(_) | DataType::VarBinary(_) => array
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .ok_or_else(|| invalid("MAP binary key requires BinaryArray"))?
+ .value(row)
+ .to_vec(),
+ DataType::Decimal(decimal) => {
+ let value = array
+ .as_any()
+ .downcast_ref::<Decimal128Array>()
+ .ok_or_else(|| invalid("MAP DECIMAL key requires
Decimal128Array"))?
+ .value(row);
+ if decimal.precision() <= 18 {
+ i64::try_from(value)
+ .map_err(|_| invalid("Compact MAP DECIMAL key exceeds
i64"))?
+ .to_le_bytes()
+ .to_vec()
+ } else {
+ // BigInteger.toByteArray(): minimal signed
two's-complement, big endian.
+ let bytes = value.to_be_bytes();
+ let mut start = 0;
+ while start < 15
+ && ((bytes[start] == 0 && bytes[start + 1] < 0x80)
+ || (bytes[start] == 0xff && bytes[start + 1] >=
0x80))
+ {
+ start += 1;
+ }
+ bytes[start..].to_vec()
+ }
+ }
+ other => {
+ return Err(Error::Unsupported {
+ message: format!("Unsupported key type for MAP<X, BLOB>:
{other:?}"),
+ })
+ }
+ };
+ if !seen.insert(key.clone()) {
+ return Err(invalid("MAP<X, BLOB> keys must be unique"));
+ }
+ keys.push(key);
+ }
+ Ok(keys)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::{
+ BigIntType, BinaryType, BooleanType, CharType, DateType, DecimalType,
FloatType, IntType,
+ SmallIntType, TimeType, TinyIntType, VarBinaryType, VarCharType,
+ };
+ use arrow_array::ArrayRef;
+ use std::sync::Arc;
+
+ #[tokio::test]
+ async fn production_array_records_match_java_golden_bytes_including_crc() {
+ use crate::io::{FileIOBuilder, FileRead};
+ use crate::spec::{ArrayType, BlobType};
+ use arrow_array::builder::{LargeBinaryBuilder, ListBuilder};
+ let io = FileIOBuilder::new("memory").build().unwrap();
+ let output = io.new_output("memory:/array.blob").unwrap();
+ let field = DataField::new(
+ 0,
+ "payloads".into(),
+ DataType::Array(ArrayType::new(DataType::Blob(BlobType::new()))),
+ );
+ let mut list = ListBuilder::new(LargeBinaryBuilder::new());
+ list.values().append_value(b"hello");
+ list.values().append_null();
+ list.values().append_value(b"world");
+ list.append(true);
+ list.append(false);
+ list.append(false);
+ list.append(true);
+ let batch = RecordBatch::try_from_iter([("payloads",
Arc::new(list.finish()) as ArrayRef)])
+ .unwrap();
+ let mut writer = BlobFormatWriter::new(&output, Some(io.clone()),
Some(&field))
+ .await
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ let result = Box::new(writer).close().await.unwrap();
+ let written = io
+ .new_input("memory:/array.blob")
+ .unwrap()
+ .reader()
+ .await
+ .unwrap()
+ .read(0..result.file_size)
+ .await
+ .unwrap();
+ let golden =
include_bytes!("../../../../testdata/blob/blob-array.blob");
+ fn data_end(bytes: &[u8]) -> usize {
+ bytes.len()
+ - 5
+ - i32::from_le_bytes(bytes[bytes.len() - 5..bytes.len() -
1].try_into().unwrap())
+ as usize
+ }
+ // The fixture's placeholder and our second NULL have no physical
+ // record; only their footer indexes differ. All physical records,
+ // including framing and CRC, must be byte-identical to Java.
+ assert_eq!(&written[..data_end(&written)],
&golden[..data_end(golden)]);
+ }
+
+ #[test]
+ fn map_keys_match_java_serializers_and_reader_for_every_supported_type() {
+ let types: Vec<(DataType, ArrayRef)> = vec![
+ (
+ DataType::Boolean(BooleanType::new()),
+ Arc::new(BooleanArray::from(vec![true, false])),
+ ),
+ (
+ DataType::TinyInt(TinyIntType::new()),
+ Arc::new(Int8Array::from(vec![i8::MIN, i8::MAX])),
+ ),
+ (
+ DataType::SmallInt(SmallIntType::new()),
+ Arc::new(Int16Array::from(vec![i16::MIN, i16::MAX])),
+ ),
+ (
+ DataType::Int(IntType::new()),
+ Arc::new(Int32Array::from(vec![i32::MIN, i32::MAX])),
+ ),
+ (
+ DataType::BigInt(BigIntType::new()),
+ Arc::new(Int64Array::from(vec![i64::MIN, i64::MAX])),
+ ),
+ (
+ DataType::Date(DateType::new()),
+ Arc::new(Date32Array::from(vec![-1, 0])),
+ ),
+ (
+ DataType::Time(TimeType::new(3).unwrap()),
+ Arc::new(Time32MillisecondArray::from(vec![0, 123])),
+ ),
+ (
+ DataType::Char(CharType::new(10).unwrap()),
+ Arc::new(StringArray::from(vec!["", "你好"])),
+ ),
+ (
+ DataType::VarChar(VarCharType::new(10).unwrap()),
+ Arc::new(StringArray::from(vec!["", "你好"])),
+ ),
+ (
+ DataType::Binary(BinaryType::new(2).unwrap()),
+ Arc::new(BinaryArray::from_iter_values([b"ab", b"cd"])),
+ ),
+ (
+ DataType::VarBinary(VarBinaryType::new(10).unwrap()),
+ Arc::new(BinaryArray::from_iter_values([b"".as_slice(),
b"abc"])),
+ ),
+ (
+ DataType::Decimal(DecimalType::new(18, 2).unwrap()),
+ Arc::new(
+ Decimal128Array::from(vec![-999, 128])
+ .with_precision_and_scale(18, 2)
+ .unwrap(),
+ ),
+ ),
+ (
+ DataType::Decimal(DecimalType::new(38, 2).unwrap()),
+ Arc::new(
+ Decimal128Array::from(vec![
+ 0,
+ 127,
+ 128,
+ -128,
+ -129,
+ 99999999999999999999999999999999999999,
+ ])
+ .with_precision_and_scale(38, 2)
+ .unwrap(),
+ ),
+ ),
+ ];
+ for (data_type, array) in types {
+ let keys = serialize_map_keys(array.as_ref(), &data_type).unwrap();
+ let bytes: Vec<_> =
keys.iter().cloned().map(Bytes::from).collect();
+ let restored = super::super::decode_blob_map_keys(&bytes,
&data_type).unwrap();
+ assert_eq!(restored.to_data(), array.to_data(), "{data_type:?}");
+ match data_type {
+ DataType::Boolean(_) => assert_eq!(keys, vec![vec![1],
vec![0]]),
+ DataType::Int(_) => {
+ assert_eq!(keys, vec![vec![0, 0, 0, 128], vec![255, 255,
255, 127]])
+ }
+ DataType::Decimal(decimal) if decimal.precision() > 18 =>
assert_eq!(
+ &keys[..5],
+ &[vec![0], vec![127], vec![0, 128], vec![128], vec![255,
127]]
+ ),
+ _ => {}
+ }
+ }
+ }
+
+ #[test]
+ fn duplicate_map_keys_are_rejected_before_the_record() {
+ for keys in [
+ Arc::new(StringArray::from(vec!["a", "a"])) as ArrayRef,
+ Arc::new(StringArray::from(vec!["", ""])) as ArrayRef,
+ ] {
+ let error = serialize_map_keys(
+ keys.as_ref(),
+ &DataType::VarChar(VarCharType::new(10).unwrap()),
+ )
+ .unwrap_err();
+ assert!(error.to_string().contains("unique"));
+ }
+
assert!(validate_key_type(&DataType::Float(FloatType::new())).is_err());
+ }
+}
diff --git a/crates/paimon/src/arrow/format/delta_varint.rs
b/crates/paimon/src/arrow/format/delta_varint.rs
new file mode 100644
index 00000000..4b266b98
--- /dev/null
+++ b/crates/paimon/src/arrow/format/delta_varint.rs
@@ -0,0 +1,97 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Java DeltaVarintCompressor signed-long indexes.
+
+use crate::Error;
+
+pub(crate) fn decode_delta_varints(bytes: &[u8]) -> crate::Result<Vec<i64>> {
+ let mut values = Vec::new();
+ let mut cursor = 0usize;
+ let mut previous = 0_i64;
+
+ while cursor < bytes.len() {
+ let (delta, consumed) = decode_varint(&bytes[cursor..])?;
+ cursor += consumed;
+
+ let value = if values.is_empty() {
+ delta
+ } else {
+ previous
+ .checked_add(delta)
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!(
+ "Blob delta-varint index overflow after previous value
{previous}"
+ ),
+ source: None,
+ })?
+ };
+ values.push(value);
+ previous = value;
+ }
+
+ Ok(values)
+}
+
+pub(super) fn decode_varint(bytes: &[u8]) -> crate::Result<(i64, usize)> {
+ let mut value = 0_u64;
+ let mut shift = 0_u32;
+
+ for (idx, byte) in bytes.iter().copied().enumerate() {
+ value |= u64::from(byte & 0x7f) << shift;
+ if (byte & 0x80) == 0 {
+ let decoded = ((value >> 1) as i64) ^ (-((value & 1) as i64));
+ return Ok((decoded, idx + 1));
+ }
+
+ shift += 7;
+ if shift > 63 {
+ return Err(Error::DataInvalid {
+ message: "Blob delta-varint index overflow".to_string(),
+ source: None,
+ });
+ }
+ }
+
+ Err(Error::DataInvalid {
+ message: "Unexpected end of blob delta-varint index".to_string(),
+ source: None,
+ })
+}
+
+pub(crate) fn encode_delta_varints(values: &[i64]) -> Vec<u8> {
+ if values.is_empty() {
+ return Vec::new();
+ }
+ let mut encoded = Vec::new();
+ let mut previous = 0_i64;
+ for (idx, &value) in values.iter().enumerate() {
+ let delta = if idx == 0 { value } else { value - previous };
+ previous = value;
+ encode_varint(delta, &mut encoded);
+ }
+ encoded
+}
+
+pub(super) fn encode_varint(value: i64, out: &mut Vec<u8>) {
+ let mut remaining = ((value << 1) ^ (value >> 63)) as u64;
+ while (remaining & !0x7f) != 0 {
+ out.push(((remaining & 0x7f) as u8) | 0x80);
+ remaining >>= 7;
+ }
+ out.push(remaining as u8);
+}
diff --git a/crates/paimon/src/arrow/format/mod.rs
b/crates/paimon/src/arrow/format/mod.rs
index a400b067..8e141314 100644
--- a/crates/paimon/src/arrow/format/mod.rs
+++ b/crates/paimon/src/arrow/format/mod.rs
@@ -18,6 +18,7 @@
mod avro;
mod avro_write;
pub(crate) mod blob;
+mod delta_varint;
mod metadata_cache;
mod mosaic;
mod mosaic_write;
@@ -27,6 +28,7 @@ mod row;
mod shredding;
pub(crate) mod text;
mod variant_projection;
+pub(crate) mod video;
#[cfg(feature = "vortex")]
mod vortex;
@@ -396,6 +398,8 @@ pub(crate) fn create_format_reader_with_budget(
blob::BlobFormatReader::new(path.to_string(), blob_as_descriptor)
.with_blob_parallelism(blob_parallelism),
)
+ } else if lower.ends_with(".video") {
+ Box::new(video::VideoFormatReader::new(path.to_string()))
} else if lower.ends_with(".orc") {
Box::new(orc::OrcFormatReader)
} else if let Some((kind, compression)) = text::TextKind::from_path(path) {
@@ -443,6 +447,7 @@ fn supported_read_formats() -> Vec<&'static str> {
vec![
".parquet",
".blob",
+ ".video",
".orc",
".csv",
".json",
@@ -573,7 +578,12 @@ impl FormatWriterFactory for PlainFormatWriterFactory {
let lower = path.to_ascii_lowercase();
if lower.ends_with(".blob") {
Ok(Box::new(
- blob::BlobFormatWriter::new(output, file_io).await?,
+ blob::BlobFormatWriter::new(
+ output,
+ file_io,
+ write_fields.and_then(|fields| fields.first()),
+ )
+ .await?,
))
} else if lower.ends_with(".orc") {
if !matches!(
diff --git a/crates/paimon/src/arrow/format/video.rs
b/crates/paimon/src/arrow/format/video.rs
new file mode 100644
index 00000000..944ff6a2
--- /dev/null
+++ b/crates/paimon/src/arrow/format/video.rs
@@ -0,0 +1,514 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Packed video v1 metadata, matching Java VideoFileMeta and
VideoFormatReader.
+//! Reading returns frame descriptors; encoded videos and keyframe indexes
stay lazy.
+
+use super::blob::{build_blob_batch, BlobReadValue};
+use super::delta_varint::decode_delta_varints;
+use super::{FilePredicates, FormatFileReader};
+use crate::arrow::build_target_arrow_schema;
+use crate::io::FileRead;
+use crate::spec::{DataField, DataType, VideoFrameDescriptor};
+use crate::table::{ArrowRecordBatchStream, RowRange};
+use crate::{Error, Result};
+use arrow_array::{RecordBatch, RecordBatchOptions};
+use async_stream::try_stream;
+use async_trait::async_trait;
+use bytes::Bytes;
+use futures::StreamExt;
+
+const FOOTER_SIZE: u64 = 25;
+const MAGIC: i32 = 0x4F454449;
+const VERSION: u8 = 1;
+const MAX_KEYFRAME_INDEX_BYTES: i64 = 16 * 1024 * 1024;
+const MAX_TOTAL_KEYFRAME_INDEX_BYTES: i64 = 64 * 1024 * 1024;
+
+fn corrupt(message: impl Into<String>) -> Error {
+ Error::DataInvalid {
+ message: format!("Corrupt video file: {}", message.into()),
+ source: None,
+ }
+}
+
+struct VideoFileMeta {
+ physical_lengths: Vec<i64>,
+ physical_offsets: Vec<i64>,
+ keyframe_lengths: Vec<i64>,
+ keyframe_offsets: Vec<i64>,
+ run_ends: Vec<usize>,
+ run_references: Vec<i64>,
+ run_first_frames: Vec<i64>,
+}
+
+impl VideoFileMeta {
+ async fn load(reader: &dyn FileRead, file_size: u64) -> Result<Self> {
+ if file_size < FOOTER_SIZE {
+ return Err(corrupt("file is smaller than its footer"));
+ }
+ let footer_start = file_size - FOOTER_SIZE;
+ let footer = reader.read(footer_start..file_size).await?;
+ if footer.len() != FOOTER_SIZE as usize {
+ return Err(corrupt("short footer read"));
+ }
+ if i32::from_le_bytes(footer[20..24].try_into().unwrap()) != MAGIC {
+ return Err(corrupt("invalid footer magic"));
+ }
+ if footer[24] != VERSION {
+ return Err(Error::Unsupported {
+ message: format!("Unsupported video format version {}",
footer[24]),
+ });
+ }
+ let lengths: Vec<_> = footer[..20]
+ .as_chunks::<4>()
+ .0
+ .iter()
+ .map(|bytes| i32::from_le_bytes(*bytes))
+ .collect();
+ if lengths.iter().any(|length| *length < 0) {
+ return Err(corrupt("negative index length"));
+ }
+ let total_length: u64 = lengths.iter().map(|length| *length as
u64).sum();
+ let index_start = footer_start
+ .checked_sub(total_length)
+ .ok_or_else(|| corrupt("indexes exceed file size"))?;
+ let index = reader.read(index_start..footer_start).await?;
+ if index.len() as u64 != total_length {
+ return Err(corrupt("short index read"));
+ }
+ let mut cursor = 0;
+ let mut indexes = Vec::with_capacity(5);
+ for length in lengths {
+ let end = cursor + length as usize;
+ indexes.push(decode_delta_varints(&index[cursor..end])?);
+ cursor = end;
+ }
+ let mut indexes = indexes.into_iter();
+ let physical_lengths = indexes.next().unwrap();
+ let keyframe_lengths = indexes.next().unwrap();
+ let run_lengths = indexes.next().unwrap();
+ let run_references = indexes.next().unwrap();
+ let run_first_frames = indexes.next().unwrap();
+ if physical_lengths.len() != keyframe_lengths.len() {
+ return Err(corrupt("physical video and keyframe index counts
differ"));
+ }
+ let mut keyframe_size = 0i64;
+ for length in &keyframe_lengths {
+ if !(0..=MAX_KEYFRAME_INDEX_BYTES).contains(length) {
+ return Err(corrupt(
+ "invalid keyframe index length or index exceeds the 16 MiB
limit",
+ ));
+ }
+ keyframe_size += length;
+ if keyframe_size > MAX_TOTAL_KEYFRAME_INDEX_BYTES {
+ return Err(corrupt("keyframe indexes exceed the 64 MiB
limit"));
+ }
+ }
+ let index_start =
+ i64::try_from(index_start).map_err(|_| corrupt("file offset
exceeds i64"))?;
+ let keyframe_start = index_start
+ .checked_sub(keyframe_size)
+ .filter(|offset| *offset >= 0)
+ .ok_or_else(|| corrupt("keyframe indexes exceed file size"))?;
+ let mut physical_offsets = Vec::with_capacity(physical_lengths.len());
+ let mut offset = 0;
+ for length in &physical_lengths {
+ if *length <= 0 || *length > keyframe_start - offset {
+ return Err(corrupt("invalid physical video length"));
+ }
+ physical_offsets.push(offset);
+ offset += length;
+ }
+ if offset != keyframe_start {
+ return Err(corrupt(
+ "physical video lengths do not match payload region",
+ ));
+ }
+ let mut keyframe_offsets = Vec::with_capacity(keyframe_lengths.len());
+ for length in &keyframe_lengths {
+ keyframe_offsets.push(offset);
+ offset += length;
+ }
+ if run_lengths.len() != run_references.len() || run_lengths.len() !=
run_first_frames.len()
+ {
+ return Err(corrupt("run indexes have different counts"));
+ }
+ let mut rows = 0i64;
+ let mut run_ends = Vec::with_capacity(run_lengths.len());
+ for ((length, reference), frame) in run_lengths
+ .iter()
+ .zip(&run_references)
+ .zip(&run_first_frames)
+ {
+ if *length <= 0 || *length > i64::from(i32::MAX) - rows {
+ return Err(corrupt("invalid run length or row count exceeds
i32"));
+ }
+ if *reference != -1 && *reference != -2 {
+ if *reference < 0 || *reference as u64 >=
physical_lengths.len() as u64 {
+ return Err(corrupt("run references an invalid physical
video"));
+ }
+ if *frame < 0 || frame.checked_add(length - 1).is_none() {
+ return Err(corrupt("invalid first frame or frame index
overflow"));
+ }
+ }
+ rows += length;
+ run_ends.push(rows as usize);
+ }
+ Ok(Self {
+ physical_lengths,
+ physical_offsets,
+ keyframe_lengths,
+ keyframe_offsets,
+ run_ends,
+ run_references,
+ run_first_frames,
+ })
+ }
+
+ fn num_rows(&self) -> usize {
+ self.run_ends.last().copied().unwrap_or(0)
+ }
+
+ fn frame(&self, position: usize, path: &str) -> Result<BlobReadValue> {
+ if position >= self.num_rows() {
+ return Err(corrupt("selected row is outside row count"));
+ }
+ let run = self.run_ends.partition_point(|end| *end <= position);
+ let reference = self.run_references[run];
+ match reference {
+ -1 => Ok(BlobReadValue::Null),
+ -2 => Ok(BlobReadValue::Placeholder),
+ _ => {
+ let ordinal = reference as usize;
+ let start = if run == 0 { 0 } else { self.run_ends[run - 1] };
+ let length = self.keyframe_lengths[ordinal];
+ let frame = VideoFrameDescriptor::new(
+ path.into(),
+ self.physical_offsets[ordinal],
+ self.physical_lengths[ordinal],
+ self.run_first_frames[run] + (position - start) as i64,
+ if length == 0 {
+ -1
+ } else {
+ self.keyframe_offsets[ordinal]
+ },
+ length,
+ )?;
+ Ok(BlobReadValue::Value(Bytes::from(frame.serialize())))
+ }
+ }
+ }
+}
+
+pub(crate) struct IndexedVideoReader {
+ metadata: VideoFileMeta,
+ path: String,
+}
+
+impl IndexedVideoReader {
+ pub(crate) async fn open(reader: &dyn FileRead, file_size: u64, path:
String) -> Result<Self> {
+ Ok(Self {
+ metadata: VideoFileMeta::load(reader, file_size).await?,
+ path,
+ })
+ }
+ pub(crate) fn num_rows(&self) -> usize {
+ self.metadata.num_rows()
+ }
+ pub(crate) fn read_positions(&self, positions: &[usize]) ->
Result<Vec<BlobReadValue>> {
+ positions
+ .iter()
+ .map(|position| self.metadata.frame(*position, &self.path))
+ .collect()
+ }
+}
+
+pub(crate) struct VideoFormatReader {
+ path: String,
+}
+
+impl VideoFormatReader {
+ pub(crate) fn new(path: String) -> Self {
+ Self { path }
+ }
+}
+
+#[async_trait]
+impl FormatFileReader for VideoFormatReader {
+ async fn read_batch_stream(
+ &self,
+ reader: Box<dyn FileRead>,
+ file_size: u64,
+ read_fields: &[DataField],
+ predicates: Option<&FilePredicates>,
+ batch_size: Option<usize>,
+ row_selection: Option<Vec<RowRange>>,
+ ) -> Result<ArrowRecordBatchStream> {
+ if let Some(predicates) = predicates {
+ crate::table::row_id_predicate::reject_row_id_filter(
+ &predicates.predicates,
+ "video files",
+ )?;
+ }
+ if read_fields.len() > 1
+ || read_fields
+ .first()
+ .is_some_and(|field| !matches!(field.data_type(),
DataType::Blob(_)))
+ {
+ return Err(corrupt("video format only supports one scalar BLOB
field"));
+ }
+ let schema = build_target_arrow_schema(read_fields)?;
+ let indexed =
+ IndexedVideoReader::open(reader.as_ref(), file_size,
self.path.clone()).await?;
+ let mut positions: Box<dyn Iterator<Item = usize> + Send> = match
row_selection {
+ None => Box::new(0..indexed.num_rows()),
+ Some(ranges) => {
+ for range in &ranges {
+ if range.from() < 0
+ || range.to() < range.from()
+ || range.to() as u64 >= indexed.num_rows() as u64
+ {
+ return Err(corrupt("selected row is outside row
count"));
+ }
+ }
+ Box::new(
+ ranges
+ .into_iter()
+ .flat_map(|range| range.from() as usize..=range.to()
as usize),
+ )
+ }
+ };
+ let batch_size = batch_size.unwrap_or(1024).max(1);
+ Ok(try_stream! {
+ loop {
+ let selected: Vec<_> =
positions.by_ref().take(batch_size).collect();
+ if selected.is_empty() { break; }
+ if schema.fields().is_empty() {
+ yield RecordBatch::try_new_with_options(schema.clone(),
vec![],
+
&RecordBatchOptions::new().with_row_count(Some(selected.len())))
+ .map_err(|error| corrupt(error.to_string()))?;
+ } else {
+ yield build_blob_batch(&schema,
indexed.read_positions(&selected)?)?;
+ }
+ }
+ }
+ .boxed())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::btree::test_util::BytesFileRead;
+ use crate::spec::{BlobType, IntType};
+ use futures::TryStreamExt;
+
+ fn java_fixture() -> Bytes {
+ let text = include_str!("../../../testdata/video/video-v1.hex");
+ hex::decode(
+ text.lines()
+ .find(|line| !line.is_empty() && !line.starts_with('#'))
+ .unwrap(),
+ )
+ .unwrap()
+ .into()
+ }
+
+ fn file(indexes: [&[i64]; 5], payload: &[u8]) -> Bytes {
+ let mut bytes = payload.to_vec();
+ let mut lengths = Vec::new();
+ for values in indexes {
+ let index =
super::super::delta_varint::encode_delta_varints(values);
+ lengths.push(index.len() as i32);
+ bytes.extend(index);
+ }
+ for length in lengths {
+ bytes.extend(length.to_le_bytes());
+ }
+ bytes.extend(MAGIC.to_le_bytes());
+ bytes.push(VERSION);
+ bytes.into()
+ }
+
+ #[tokio::test]
+ async fn
latest_java_fixture_preserves_frames_indexes_nulls_and_placeholders() {
+ let bytes = java_fixture();
+ let reader = IndexedVideoReader::open(
+ &BytesFileRead(bytes.clone()),
+ bytes.len() as u64,
+ "file:/data.video".into(),
+ )
+ .await
+ .unwrap();
+ assert_eq!(reader.num_rows(), 7);
+ let values = reader.read_positions(&[0, 1, 2, 3, 4, 5, 6]).unwrap();
+ for (index, frame, offset, length, keyframe) in [
+ (0, 2, 0, 3, true),
+ (1, 3, 0, 3, true),
+ (4, 7, 3, 4, false),
+ (5, 8, 3, 4, false),
+ (6, 10, 0, 3, true),
+ ] {
+ let BlobReadValue::Value(bytes) = &values[index] else {
+ panic!("expected frame")
+ };
+ let descriptor = VideoFrameDescriptor::deserialize(bytes).unwrap();
+ assert_eq!(descriptor.frame_index(), frame);
+ assert_eq!(descriptor.payload_descriptor().offset(), offset);
+ assert_eq!(descriptor.payload_descriptor().length(), length);
+ assert_eq!(descriptor.keyframe_index_descriptor().is_some(),
keyframe);
+ if let Some(mapping) = descriptor.keyframe_index_descriptor() {
+ assert_eq!(mapping.offset(), 7);
+ assert_eq!(mapping.length(), 54);
+ }
+ }
+ assert!(matches!(values[2], BlobReadValue::Null));
+ assert!(matches!(values[3], BlobReadValue::Placeholder));
+ assert!(reader.read_positions(&[7]).is_err());
+ }
+
+ #[tokio::test]
+ async fn selection_batching_and_zero_column_reads_keep_logical_rows() {
+ let bytes = java_fixture();
+ let format = VideoFormatReader::new("file:/data.video".into());
+ let field = DataField::new(0, "video".into(),
DataType::Blob(BlobType::new()));
+ for fields in [vec![field], vec![]] {
+ let stream = format
+ .read_batch_stream(
+ Box::new(BytesFileRead(bytes.clone())),
+ bytes.len() as u64,
+ &fields,
+ None,
+ Some(2),
+ Some(vec![RowRange::new(1, 3), RowRange::new(6, 6)]),
+ )
+ .await
+ .unwrap();
+ let batches: Vec<_> = stream.try_collect().await.unwrap();
+ assert_eq!(
+ batches
+ .iter()
+ .map(RecordBatch::num_rows)
+ .collect::<Vec<_>>(),
+ vec![2, 2]
+ );
+ assert!(batches
+ .iter()
+ .all(|batch| batch.num_columns() == fields.len()));
+ }
+ for selection in [vec![RowRange::new(7, 7)], vec![RowRange::new(-1,
0)]] {
+ assert!(format
+ .read_batch_stream(
+ Box::new(BytesFileRead(bytes.clone())),
+ bytes.len() as u64,
+ &[],
+ None,
+ None,
+ Some(selection)
+ )
+ .await
+ .is_err());
+ }
+ let empty = format
+ .read_batch_stream(
+ Box::new(BytesFileRead(bytes.clone())),
+ bytes.len() as u64,
+ &[],
+ None,
+ None,
+ Some(vec![]),
+ )
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert!(empty.is_empty());
+ assert!(format
+ .read_batch_stream(
+ Box::new(BytesFileRead(bytes.clone())),
+ bytes.len() as u64,
+ &[DataField::new(
+ 0,
+ "id".into(),
+ DataType::Int(IntType::new())
+ )],
+ None,
+ None,
+ None
+ )
+ .await
+ .is_err());
+ }
+
+ #[tokio::test]
+ async fn malformed_footer_and_run_indexes_fail_before_returning_rows() {
+ let good = file([&[3], &[0], &[2], &[0], &[0]], b"abc");
+ assert_eq!(
+ VideoFileMeta::load(&BytesFileRead(good.clone()), good.len() as
u64)
+ .await
+ .unwrap()
+ .num_rows(),
+ 2
+ );
+ let mut bad = Vec::new();
+ bad.push(Bytes::from_static(b"short"));
+ let mut bytes = good.to_vec();
+ *bytes.last_mut().unwrap() = 2;
+ bad.push(bytes.into());
+ let mut bytes = good.to_vec();
+ let n = bytes.len();
+ bytes[n - 5] ^= 1;
+ bad.push(bytes.into());
+ let mut bytes = good.to_vec();
+ let n = bytes.len();
+ bytes[n - 25..n - 21].copy_from_slice(&(-1i32).to_le_bytes());
+ bad.push(bytes.into());
+ for indexes in [
+ [&[0][..], &[0], &[1], &[0], &[0]],
+ [&[4][..], &[0], &[1], &[0], &[0]],
+ [&[3][..], &[], &[1], &[0], &[0]],
+ [&[3][..], &[-1], &[1], &[0], &[0]],
+ [&[3][..], &[16 * 1024 * 1024 + 1], &[1], &[0], &[0]],
+ [&[3][..], &[0], &[0], &[0], &[0]],
+ [&[3][..], &[0], &[1], &[1], &[0]],
+ [&[3][..], &[0], &[1], &[-3], &[0]],
+ [&[3][..], &[0], &[1], &[0], &[-1]],
+ [&[3][..], &[0], &[2], &[0], &[i64::MAX]],
+ [&[3][..], &[0], &[i32::MAX as i64, 1], &[0, 0], &[0, 0]],
+ [&[3][..], &[0], &[1], &[0], &[]],
+ ] {
+ bad.push(file(indexes, b"abc"));
+ }
+ for bytes in bad {
+ assert!(
+ VideoFileMeta::load(&BytesFileRead(bytes.clone()), bytes.len()
as u64)
+ .await
+ .is_err()
+ );
+ }
+ let nulls = file([&[], &[], &[3], &[-1], &[0]], b"");
+ let meta = VideoFileMeta::load(&BytesFileRead(nulls.clone()),
nulls.len() as u64)
+ .await
+ .unwrap();
+ assert_eq!(meta.num_rows(), 3);
+ assert!(matches!(
+ meta.frame(2, "unused").unwrap(),
+ BlobReadValue::Null
+ ));
+ }
+}
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index ae04d111..0d33a760 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -182,6 +182,7 @@ pub(crate) const BLOB_FIELD_OPTION: &str = "blob-field";
pub(crate) const BLOB_DESCRIPTOR_FIELD_OPTION: &str = "blob-descriptor-field";
pub(crate) const BLOB_DESCRIPTOR_FIELD_FALLBACK: &str =
"blob.stored-descriptor-fields";
pub(crate) const BLOB_VIEW_FIELD_OPTION: &str = "blob-view-field";
+pub(crate) const VIDEO_FRAME_FIELD_OPTION: &str = "video-frame-field";
pub const BLOB_VIEW_RESOLVE_ENABLED_OPTION: &str = "blob-view.resolve.enabled";
const PK_VECTOR_INDEX_COLUMNS_OPTION: &str = "pk-vector.index.columns";
const PK_FULL_TEXT_INDEX_COLUMNS_OPTION: &str = "pk-full-text.index.columns";
@@ -1783,9 +1784,15 @@ impl<'a> CoreOptions<'a> {
let mut fields = self.parse_csv_set(BLOB_FIELD_OPTION);
fields.extend(self.blob_descriptor_fields());
fields.extend(self.blob_view_fields());
+ fields.extend(self.video_frame_fields());
fields
}
+ /// Scalar BLOB columns stored as logical frame runs in packed `.video`
files.
+ pub fn video_frame_fields(&self) -> HashSet<String> {
+ self.parse_csv_set(VIDEO_FRAME_FIELD_OPTION)
+ }
+
/// Comma-separated BLOB field names stored as serialized BlobDescriptor
/// bytes inline in normal data files (no .blob files for these fields).
pub fn blob_descriptor_fields(&self) -> HashSet<String> {
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index 9c64f6fd..472fa7c2 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -25,6 +25,9 @@ pub use binary_row::*;
mod blob_descriptor;
pub use blob_descriptor::BlobDescriptor;
+mod video_frame_descriptor;
+pub use video_frame_descriptor::VideoFrameDescriptor;
+
mod blob_view_struct;
pub use blob_view_struct::BlobViewStruct;
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index c11ebdfa..b6f73e09 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -1648,6 +1648,36 @@ impl Schema {
primary_keys: &[String],
options: &HashMap<String, String>,
) -> crate::Result<()> {
+ let core_options = CoreOptions::new(options);
+ let video_fields = core_options.video_frame_fields();
+ if !video_fields.is_empty() && !primary_keys.is_empty() {
+ return Err(crate::Error::ConfigInvalid {
+ message: "'video-frame-field' only supports append-only
tables".into(),
+ });
+ }
+ for name in video_fields {
+ if !fields
+ .iter()
+ .any(|field| field.name() == name &&
field.data_type().is_blob_type())
+ {
+ return Err(crate::Error::ConfigInvalid { message:
format!("Field '{name}' in 'video-frame-field' must be a scalar BLOB field in
table schema") });
+ }
+ for (option, names) in [
+ (
+ "blob-descriptor-field",
+ core_options.blob_descriptor_fields(),
+ ),
+ ("blob-view-field", core_options.blob_view_fields()),
+ ] {
+ if names.contains(&name) {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Field '{name}' in 'video-frame-field' can not
also be in '{option}'"
+ ),
+ });
+ }
+ }
+ }
let blob_field_names = Self::top_level_blob_field_names(fields);
for field in fields {
if !Self::is_top_level_blob_file_type(field.data_type())
@@ -1666,7 +1696,6 @@ impl Schema {
return Ok(());
}
- let core_options = CoreOptions::new(options);
let blob_descriptor_fields = core_options.blob_descriptor_fields();
let blob_view_fields = core_options.blob_view_fields();
let mut overlapping_fields = blob_view_fields
diff --git a/crates/paimon/src/spec/video_frame_descriptor.rs
b/crates/paimon/src/spec/video_frame_descriptor.rs
new file mode 100644
index 00000000..b592741c
--- /dev/null
+++ b/crates/paimon/src/spec/video_frame_descriptor.rs
@@ -0,0 +1,215 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Java VideoFrameDescriptor v1, including the persisted keyframe-index range.
+
+use super::BlobDescriptor;
+use crate::{Error, Result};
+
+const VERSION: u8 = 1;
+const MAGIC: i64 = 0x564944454F46524D;
+const FIXED_LENGTH: usize = 1 + 8 + 4 + 5 * 8;
+
+/// One logical frame in an encoded video, with an optional keyframe index.
+/// The payload is the complete encoded video; `frame_index` locates the frame.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct VideoFrameDescriptor {
+ payload: BlobDescriptor,
+ frame_index: i64,
+ keyframe_index_offset: i64,
+ keyframe_index_length: i64,
+}
+
+fn invalid(message: impl Into<String>) -> Error {
+ Error::DataInvalid {
+ message: format!("Invalid VideoFrameDescriptor data: {}",
message.into()),
+ source: None,
+ }
+}
+
+impl VideoFrameDescriptor {
+ /// Construct a descriptor using the same locator validation as Java.
+ pub fn new(
+ uri: String,
+ offset: i64,
+ length: i64,
+ frame_index: i64,
+ keyframe_index_offset: i64,
+ keyframe_index_length: i64,
+ ) -> Result<Self> {
+ if frame_index < 0 {
+ return Err(invalid("negative frame index"));
+ }
+ if !(keyframe_index_length == 0 && keyframe_index_offset == -1
+ || keyframe_index_length > 0 && keyframe_index_offset >= 0)
+ {
+ return Err(invalid("invalid video keyframe index range"));
+ }
+ Ok(Self {
+ payload: BlobDescriptor::new(uri, offset, length),
+ frame_index,
+ keyframe_index_offset,
+ keyframe_index_length,
+ })
+ }
+
+ /// Physical video identity, without the logical frame locator.
+ pub fn payload_descriptor(&self) -> &BlobDescriptor {
+ &self.payload
+ }
+ /// Zero-based frame ordinal within the physical video.
+ pub fn frame_index(&self) -> i64 {
+ self.frame_index
+ }
+ /// The optional index is stored in the same object as the physical video.
+ pub fn keyframe_index_descriptor(&self) -> Option<BlobDescriptor> {
+ (self.keyframe_index_length > 0).then(|| {
+ BlobDescriptor::new(
+ self.payload.uri().into(),
+ self.keyframe_index_offset,
+ self.keyframe_index_length,
+ )
+ })
+ }
+
+ /// Serialize to Java's exact little-endian wire layout.
+ pub fn serialize(&self) -> Vec<u8> {
+ let mut bytes = Vec::with_capacity(FIXED_LENGTH +
self.payload.uri().len());
+ bytes.push(VERSION);
+ bytes.extend_from_slice(&MAGIC.to_le_bytes());
+ bytes.extend_from_slice(&(self.payload.uri().len() as
i32).to_le_bytes());
+ bytes.extend_from_slice(self.payload.uri().as_bytes());
+ for value in [
+ self.payload.offset(),
+ self.payload.length(),
+ self.frame_index,
+ self.keyframe_index_offset,
+ self.keyframe_index_length,
+ ] {
+ bytes.extend_from_slice(&value.to_le_bytes());
+ }
+ bytes
+ }
+
+ /// Recognize the version and magic only, matching Java's detection helper.
+ pub fn is_video_frame_descriptor(bytes: &[u8]) -> bool {
+ bytes.len() >= 9 && bytes[0] == VERSION && bytes[1..9] ==
MAGIC.to_le_bytes()
+ }
+
+ /// Deserialize a complete frame reference; trailing bytes are invalid.
+ pub fn deserialize(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() < FIXED_LENGTH {
+ return Err(invalid("too short"));
+ }
+ if bytes[0] != VERSION {
+ return Err(Error::Unsupported {
+ message: format!("Unsupported VideoFrameDescriptor version
{}", bytes[0]),
+ });
+ }
+ if bytes[1..9] != MAGIC.to_le_bytes() {
+ return Err(invalid("missing magic header"));
+ }
+ let uri_length = i32::from_le_bytes(bytes[9..13].try_into().unwrap());
+ let uri_length = usize::try_from(uri_length).map_err(|_|
invalid("negative URI length"))?;
+ if uri_length > bytes.len() - FIXED_LENGTH {
+ return Err(invalid("URI length exceeds data size"));
+ }
+ if uri_length != bytes.len() - FIXED_LENGTH {
+ return Err(invalid("trailing bytes"));
+ }
+ let end = 13 + uri_length;
+ // Java new String(bytes, UTF_8) replaces invalid UTF-8 sequences.
+ let uri = String::from_utf8_lossy(&bytes[13..end]).into_owned();
+ let values: Vec<_> = bytes[end..]
+ .as_chunks::<8>()
+ .0
+ .iter()
+ .map(|bytes| i64::from_le_bytes(*bytes))
+ .collect();
+ Self::new(uri, values[0], values[1], values[2], values[3], values[4])
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn descriptor_matches_current_java_fixture() {
+ let text =
include_str!("../../testdata/video/video-frame-descriptor-v1.hex");
+ let bytes = hex::decode(
+ text.lines()
+ .find(|line| !line.is_empty() && !line.starts_with('#'))
+ .unwrap(),
+ )
+ .unwrap();
+ let descriptor =
+ VideoFrameDescriptor::new("s3://bucket/视频.mp4".into(), 7, 99, 42,
106, 8).unwrap();
+ assert_eq!(descriptor.serialize(), bytes);
+ assert_eq!(
+ VideoFrameDescriptor::deserialize(&bytes).unwrap(),
+ descriptor
+ );
+ assert_eq!(
+ descriptor.keyframe_index_descriptor().unwrap(),
+ BlobDescriptor::new("s3://bucket/视频.mp4".into(), 106, 8)
+ );
+ assert!(!BlobDescriptor::is_blob_descriptor(&bytes));
+ }
+
+ #[test]
+ fn descriptors_without_indexes_and_unknown_payload_lengths_round_trip() {
+ let frame =
+ VideoFrameDescriptor::new("memory:/video".into(), 4, -1, i64::MAX,
-1, 0).unwrap();
+ assert!(frame.keyframe_index_descriptor().is_none());
+ assert_eq!(
+ frame,
+ VideoFrameDescriptor::deserialize(&frame.serialize()).unwrap()
+ );
+ }
+
+ #[test]
+ fn malformed_descriptors_are_rejected_without_overflow() {
+ let original = VideoFrameDescriptor::new("x".into(), 0, 1, 0, -1, 0)
+ .unwrap()
+ .serialize();
+ for length in 0..original.len() {
+
assert!(VideoFrameDescriptor::deserialize(&original[..length]).is_err());
+ }
+ for length in [-1i32, i32::MAX] {
+ let mut bytes = original.clone();
+ bytes[9..13].copy_from_slice(&length.to_le_bytes());
+ assert!(VideoFrameDescriptor::deserialize(&bytes).is_err());
+ }
+ let mut bytes = original.clone();
+ bytes.push(0);
+ assert!(VideoFrameDescriptor::deserialize(&bytes).is_err());
+ let mut bytes = original.clone();
+ bytes[1] ^= 1;
+ assert!(!VideoFrameDescriptor::is_video_frame_descriptor(&bytes));
+ assert!(VideoFrameDescriptor::deserialize(&bytes).is_err());
+ let mut bytes = original;
+ bytes[0] = 2;
+ assert!(matches!(
+ VideoFrameDescriptor::deserialize(&bytes),
+ Err(Error::Unsupported { .. })
+ ));
+ for (frame, offset, length) in [(-1, -1, 0), (0, 0, 0), (0, -1, 1),
(0, 0, -1)] {
+ assert!(VideoFrameDescriptor::new("x".into(), 0, 1, frame, offset,
length).is_err());
+ }
+ }
+}
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index 88b0bd46..d30d4020 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -34,7 +34,7 @@ use crate::spec::{
BlobDescriptor, BlobViewStruct, CoreOptions, DataField, DataFileMeta,
DataType, Predicate,
ROW_ID_FIELD_NAME,
};
-use crate::table::dedicated_format_file_writer::is_blob_file_name;
+use crate::table::dedicated_format_file_writer::is_blob_or_video_file_name;
use crate::table::schema_manager::SchemaManager;
use crate::table::source::any_range_overlaps_file;
use crate::table::stats_filter::group_by_overlapping_row_id;
@@ -280,7 +280,7 @@ impl DataEvolutionReader {
split
.data_files()
.iter()
- .any(|file| is_blob_file_name(&file.file_name))
+ .any(|file| is_blob_or_video_file_name(&file.file_name))
})
// A legacy raw file without first_row_id cannot be addressed by
// the preselection's global row ranges. Keep its existing read
@@ -2148,7 +2148,8 @@ impl PreparedMergeGroup {
let data_files: Vec<&DataFileMeta> = files
.iter()
.filter(|file| {
- !is_blob_file_name(&file.file_name) &&
!is_vector_store_file_name(&file.file_name)
+ !is_blob_or_video_file_name(&file.file_name)
+ && !is_vector_store_file_name(&file.file_name)
})
.collect();
if data_files.is_empty() {
@@ -2336,7 +2337,7 @@ fn build_source_plan_with_row_id_pushdown(
for (file_idx, file) in prepared_group.files.iter().enumerate() {
let info = &file_infos[file_idx];
- if is_blob_file_name(&file.file_name) {
+ if is_blob_or_video_file_name(&file.file_name) {
let field_id = resolve_blob_field_id(file, info)?;
let expected_row_count = expected_blob_row_count.ok_or_else(||
Error::DataInvalid {
message: format!(
@@ -2688,7 +2689,7 @@ impl BlobBunch {
self.logical_ranges.is_none(),
"Cannot add files to a finalized blob bunch"
);
- if !is_blob_file_name(&file.file_name) {
+ if !is_blob_or_video_file_name(&file.file_name) {
return Err(Error::DataInvalid {
message: "Only blob file can be added to a blob
bunch.".to_string(),
source: None,
@@ -3065,7 +3066,7 @@ fn normalize_merge_group(files: Vec<DataFileMeta>) ->
crate::Result<Vec<DataFile
let mut blob_files = Vec::new();
for file in files {
- if is_blob_file_name(&file.file_name) {
+ if is_blob_or_video_file_name(&file.file_name) {
blob_files.push(file);
} else if is_vector_store_file_name(&file.file_name) {
vector_files.push(file);
@@ -4008,7 +4009,7 @@ mod tests {
.files
.iter()
.map(|file| {
- if is_blob_file_name(&file.file_name) {
+ if is_blob_or_video_file_name(&file.file_name) {
resolved_info(vec![2])
} else {
resolved_info(vec![1])
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 3000aed9..a650ab20 100644
--- a/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
+++ b/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
@@ -24,6 +24,7 @@ use crate::arrow::format::blob::{
build_blob_array_batch, build_blob_batch, build_blob_map_batch,
BlobFieldKind, BlobReadValue,
IndexedBlobReader,
};
+use crate::arrow::format::video::IndexedVideoReader;
use crate::io::FileIO;
use crate::spec::{DataField, DataType};
use crate::table::{ArrowRecordBatchStream, RowRange};
@@ -42,7 +43,21 @@ struct LazyBlobFile {
path: String,
file_size: i64,
row_count: i64,
- reader: Option<IndexedBlobReader>,
+ reader: Option<IndexedDedicatedReader>,
+}
+
+enum IndexedDedicatedReader {
+ Blob(IndexedBlobReader),
+ Video(IndexedVideoReader),
+}
+
+impl IndexedDedicatedReader {
+ fn num_rows(&self) -> usize {
+ match self {
+ Self::Blob(reader) => reader.num_rows(),
+ Self::Video(reader) => reader.num_rows(),
+ }
+ }
}
impl LazyBlobFile {
@@ -64,14 +79,28 @@ impl LazyBlobFile {
})?;
let input = file_io.new_input(&self.path)?;
let reader = input.reader().await?;
- let reader = IndexedBlobReader::open_with_parallelism(
- Box::new(reader),
- file_size,
- self.path.clone(),
- blob_as_descriptor,
- blob_parallelism,
- )
- .await?;
+ let reader = if
self.file_name.to_ascii_lowercase().ends_with(".video") {
+ if !matches!(field_kind, BlobFieldKind::Scalar) {
+ return Err(Error::DataInvalid {
+ message: "Video files require scalar BLOB
fields".into(),
+ source: None,
+ });
+ }
+ IndexedDedicatedReader::Video(
+ IndexedVideoReader::open(&reader, file_size,
self.path.clone()).await?,
+ )
+ } else {
+ IndexedDedicatedReader::Blob(
+ IndexedBlobReader::open_with_parallelism(
+ Box::new(reader),
+ file_size,
+ self.path.clone(),
+ blob_as_descriptor,
+ blob_parallelism,
+ )
+ .await?,
+ )
+ };
let indexed_rows =
i64::try_from(reader.num_rows()).map_err(|e|
Error::DataInvalid {
message: format!(
@@ -97,10 +126,15 @@ impl LazyBlobFile {
.reader
.as_ref()
.expect("blob reader is initialized above");
- match field_kind {
- BlobFieldKind::Scalar => reader.read_positions(positions).await,
- BlobFieldKind::Array =>
reader.read_array_positions(positions).await,
- BlobFieldKind::Map(key_type) =>
reader.read_map_positions(positions, key_type).await,
+ match reader {
+ IndexedDedicatedReader::Video(reader) =>
reader.read_positions(positions),
+ IndexedDedicatedReader::Blob(reader) => match field_kind {
+ BlobFieldKind::Scalar =>
reader.read_positions(positions).await,
+ BlobFieldKind::Array =>
reader.read_array_positions(positions).await,
+ BlobFieldKind::Map(key_type) => {
+ reader.read_map_positions(positions, key_type).await
+ }
+ },
}
}
@@ -396,7 +430,7 @@ mod tests {
path: file_name.to_string(),
file_size: i64::try_from(file_size).unwrap(),
row_count: i64::try_from(values.len()).unwrap(),
- reader: Some(reader),
+ reader: Some(IndexedDedicatedReader::Blob(reader)),
},
ranges,
)
diff --git a/crates/paimon/src/table/data_evolution_writer.rs
b/crates/paimon/src/table/data_evolution_writer.rs
index 2dd6697f..0e82ade7 100644
--- a/crates/paimon/src/table/data_evolution_writer.rs
+++ b/crates/paimon/src/table/data_evolution_writer.rs
@@ -159,7 +159,9 @@ impl DataEvolutionWriter {
});
}
if let Some(field) = schema.fields().iter().find(|f| f.name() ==
top_level) {
- if field.data_type().is_blob_type() &&
!updatable_blob_fields.contains(top_level) {
+ if field.data_type().is_blob_file_field()
+ && !updatable_blob_fields.contains(top_level)
+ {
return Err(crate::Error::Unsupported {
message: format!(
"Cannot update raw-data BLOB column '{col}' in
MERGE INTO. \
diff --git a/crates/paimon/src/table/dedicated_format_file_writer.rs
b/crates/paimon/src/table/dedicated_format_file_writer.rs
index df5cb99c..901b7b0c 100644
--- a/crates/paimon/src/table/dedicated_format_file_writer.rs
+++ b/crates/paimon/src/table/dedicated_format_file_writer.rs
@@ -24,8 +24,9 @@ use arrow_array::RecordBatch;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
-pub(crate) fn is_blob_file_name(file_name: &str) -> bool {
- file_name.to_ascii_lowercase().ends_with(".blob")
+pub(crate) fn is_blob_or_video_file_name(file_name: &str) -> bool {
+ let lower = file_name.to_ascii_lowercase();
+ lower.ends_with(".blob") || lower.ends_with(".video")
}
struct BlobFieldWriter {
@@ -99,7 +100,7 @@ impl AppendDedicatedFormatFileWriter {
let mut vector_field_names = Vec::new();
for (idx, field) in table_fields.iter().enumerate() {
- let is_blob = field.data_type().is_blob_type();
+ let is_blob = field.data_type().is_blob_file_field();
let is_inline = blob_inline_fields.contains(field.name());
let is_dedicated_vector =
vector_file_format.is_some() && matches!(field.data_type(),
DataType::Vector(_));
diff --git a/crates/paimon/src/table/managed_blob_writer.rs
b/crates/paimon/src/table/managed_blob_writer.rs
index 5a1e8d6a..ed951934 100644
--- a/crates/paimon/src/table/managed_blob_writer.rs
+++ b/crates/paimon/src/table/managed_blob_writer.rs
@@ -357,7 +357,7 @@ impl ManagedBlobWriter {
);
let output = self.file_io.new_output(&path)?;
let writer =
- Box::new(BlobFormatWriter::new(&output,
Some(self.file_io.clone())).await?);
+ Box::new(BlobFormatWriter::new(&output,
Some(self.file_io.clone()), None).await?);
self.uncommitted_paths.push(path.clone());
self.fields[field_index].current = Some(ManagedBlobPack { path,
writer });
}
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index ab501eeb..64be1888 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -30,7 +30,7 @@ fn is_vector_store_file_name(file_name: &str) -> bool {
}
pub(crate) fn is_data_evolution_normal_file(file: &DataFileMeta) -> bool {
-
!crate::table::dedicated_format_file_writer::is_blob_file_name(&file.file_name)
+
!crate::table::dedicated_format_file_writer::is_blob_or_video_file_name(&file.file_name)
&& !is_vector_store_file_name(&file.file_name)
}
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 17cc19ba..5ad2d51f 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -3477,7 +3477,7 @@ fn ranges_overlap(left_start: i64, left_end: i64,
right_start: i64, right_end: i
}
fn is_blob_data_file(file: &DataFileMeta) -> bool {
-
crate::table::dedicated_format_file_writer::is_blob_file_name(&file.file_name)
+
crate::table::dedicated_format_file_writer::is_blob_or_video_file_name(&file.file_name)
}
fn is_vector_store_file(file: &DataFileMeta) -> bool {
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 3b93a05e..f31d64fd 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -901,7 +901,7 @@ fn is_vector_store_file_name(file_name: &str) -> bool {
}
fn is_normal_data_file(file: &DataFileMeta) -> bool {
-
!crate::table::dedicated_format_file_writer::is_blob_file_name(&file.file_name)
+
!crate::table::dedicated_format_file_writer::is_blob_or_video_file_name(&file.file_name)
&& !is_vector_store_file_name(&file.file_name)
}
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 30384d9f..b38fed10 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -262,6 +262,11 @@ impl TableWrite {
let write_schema = build_target_arrow_schema(schema.fields())?;
let core_options = CoreOptions::new(schema.options());
let blob_descriptor_fields = core_options.blob_descriptor_fields();
+ if !core_options.video_frame_fields().is_empty() {
+ return Err(crate::Error::Unsupported {
+ message: "Writing packed video files is not supported
yet".into(),
+ });
+ }
let blob_view_fields = core_options.blob_view_fields();
let blob_inline_fields = core_options.blob_inline_fields();
@@ -457,7 +462,7 @@ impl TableWrite {
let has_blob_fields = schema
.fields()
.iter()
- .any(|f| f.data_type().is_blob_type() &&
!blob_inline_fields.contains(f.name()));
+ .any(|f| f.data_type().is_blob_file_field() &&
!blob_inline_fields.contains(f.name()));
let has_dedicated_vector_fields = vector_file_format.is_some()
&& schema
.fields()
diff --git a/crates/paimon/src/table/write_batch_normalize.rs
b/crates/paimon/src/table/write_batch_normalize.rs
index dcd0271a..38d44086 100644
--- a/crates/paimon/src/table/write_batch_normalize.rs
+++ b/crates/paimon/src/table/write_batch_normalize.rs
@@ -20,18 +20,35 @@
//! PyArrow calls list children `item`; Paimon's Arrow schema calls them
//! `element`. Parquet persists child names, so accepting the input schema
//! without rebuilding the arrays would produce a file that later projections
-//! cannot reliably read by the table schema. The buffers and values are
shared;
-//! only nested array wrappers are rebuilt.
+//! cannot reliably read by the table schema. Ordinary inputs share their
+//! payload buffers; hidden children of NULL parents may require compaction.
use crate::{Error, Result};
use arrow_array::{
Array, ArrayRef, BinaryArray, FixedSizeBinaryArray, ListArray, MapArray,
StructArray,
};
+use arrow_buffer::{NullBuffer, OffsetBuffer, ScalarBuffer};
use arrow_schema::DataType;
use std::sync::Arc;
pub(super) fn normalize_write_array(array: &ArrayRef, expected: &DataType) ->
Result<ArrayRef> {
- if array.data_type() == expected {
+ normalize_visible_array(array, expected, None)
+}
+
+fn normalize_visible_array(
+ array: &ArrayRef,
+ expected: &DataType,
+ parent_nulls: Option<&NullBuffer>,
+) -> Result<ArrayRef> {
+ // Foreign Arrow arrays can declare non-nullable nested fields without
+ // satisfying them. Rebuild nested wrappers even when their schema matches
+ // so the Arrow constructors validate the actual children before writing.
+ if array.data_type() == expected
+ && !matches!(
+ expected,
+ DataType::List(_) | DataType::Struct(_) | DataType::Map(_, _)
+ )
+ {
return Ok(array.clone());
}
@@ -47,20 +64,19 @@ pub(super) fn normalize_write_array(array: &ArrayRef,
expected: &DataType) -> Re
}
(DataType::List(_), DataType::List(expected_child)) => {
let list = array.as_any().downcast_ref::<ListArray>().unwrap();
- let values = normalize_write_array(list.values(),
expected_child.data_type())?;
- let normalized = ListArray::try_new(
- expected_child.clone(),
- list.offsets().clone(),
- values,
- list.nulls().cloned(),
- )
- .map_err(normalization_error)?;
+ let nulls = NullBuffer::union(list.nulls(), parent_nulls);
+ let (values, offsets) =
+ visible_children(list.values(), list.offsets(),
nulls.as_ref())?;
+ let values = normalize_write_array(&values,
expected_child.data_type())?;
+ let normalized = ListArray::try_new(expected_child.clone(),
offsets, values, nulls)
+ .map_err(normalization_error)?;
Ok(Arc::new(normalized))
}
(DataType::Struct(actual_fields), DataType::Struct(expected_fields))
if actual_fields.len() == expected_fields.len() =>
{
let row = array.as_any().downcast_ref::<StructArray>().unwrap();
+ let nulls = NullBuffer::union(row.nulls(), parent_nulls);
let columns = actual_fields
.iter()
.zip(expected_fields)
@@ -76,16 +92,24 @@ pub(super) fn normalize_write_array(array: &ArrayRef,
expected: &DataType) -> Re
source: None,
});
}
- normalize_write_array(row.column(index),
expected.data_type())
+ normalize_visible_array(row.column(index),
expected.data_type(), nulls.as_ref())
})
.collect::<Result<Vec<_>>>()?;
- let normalized =
- StructArray::try_new(expected_fields.clone(), columns,
row.nulls().cloned())
- .map_err(normalization_error)?;
+ let normalized = StructArray::try_new_with_length(
+ expected_fields.clone(),
+ columns,
+ nulls,
+ row.len(),
+ )
+ .map_err(normalization_error)?;
Ok(Arc::new(normalized))
}
(DataType::Map(_, _), DataType::Map(expected_entries,
expected_ordered)) => {
let map = array.as_any().downcast_ref::<MapArray>().unwrap();
+ let nulls = NullBuffer::union(map.nulls(), parent_nulls);
+ let entries: ArrayRef = Arc::new(map.entries().clone());
+ let (entries, offsets) = visible_children(&entries, map.offsets(),
nulls.as_ref())?;
+ let entries =
entries.as_any().downcast_ref::<StructArray>().unwrap();
// MAP key/value roles are positional. Their Arrow field names are
// aliases, unlike user-defined ROW field names.
let DataType::Struct(fields) = expected_entries.data_type() else {
@@ -94,8 +118,9 @@ pub(super) fn normalize_write_array(array: &ArrayRef,
expected: &DataType) -> Re
source: None,
});
};
- let columns = [map.keys(), map.values()]
- .into_iter()
+ let columns = entries
+ .columns()
+ .iter()
.zip(fields)
.map(|(column, field)| normalize_write_array(column,
field.data_type()))
.collect::<Result<Vec<_>>>()?;
@@ -103,9 +128,9 @@ pub(super) fn normalize_write_array(array: &ArrayRef,
expected: &DataType) -> Re
StructArray::try_new(fields.clone(), columns,
None).map_err(normalization_error)?;
let normalized = MapArray::try_new(
expected_entries.clone(),
- map.offsets().clone(),
+ offsets,
entries,
- map.nulls().cloned(),
+ nulls,
*expected_ordered,
)
.map_err(normalization_error)?;
@@ -121,6 +146,70 @@ pub(super) fn normalize_write_array(array: &ArrayRef,
expected: &DataType) -> Re
}
}
+/// Keep only children referenced by visible parents. Ordinary inputs and
+/// slices share their payload buffers; only disjoint ranges across NULL
+/// parents need concatenation. Hidden children must not fail NOT NULL checks
+/// or resolve descriptors which Java would never visit.
+fn visible_children(
+ values: &ArrayRef,
+ offsets: &OffsetBuffer<i32>,
+ nulls: Option<&NullBuffer>,
+) -> Result<(ArrayRef, OffsetBuffer<i32>)> {
+ if nulls.is_none_or(|nulls| nulls.null_count() == 0) {
+ let start = offsets[0];
+ let end = offsets[offsets.len() - 1];
+ let normalized_offsets = if start == 0 {
+ offsets.clone()
+ } else {
+ OffsetBuffer::new(ScalarBuffer::from(
+ offsets
+ .iter()
+ .map(|offset| offset - start)
+ .collect::<Vec<_>>(),
+ ))
+ };
+ return Ok((
+ values.slice(start as usize, (end - start) as usize),
+ normalized_offsets,
+ ));
+ }
+ let mut ranges: Vec<std::ops::Range<usize>> = Vec::new();
+ let mut normalized_offsets = Vec::with_capacity(offsets.len());
+ normalized_offsets.push(0);
+ let mut length = 0i32;
+ for (row, pair) in offsets.windows(2).enumerate() {
+ if !nulls.is_some_and(|nulls| nulls.is_null(row)) {
+ let start = pair[0] as usize;
+ let end = pair[1] as usize;
+ if start != end {
+ if let Some(previous) = ranges.last_mut().filter(|range|
range.end == start) {
+ previous.end = end;
+ } else {
+ ranges.push(start..end);
+ }
+ length += pair[1] - pair[0];
+ }
+ }
+ normalized_offsets.push(length);
+ }
+ let children = match ranges.as_slice() {
+ [] => values.slice(0, 0),
+ [range] => values.slice(range.start, range.len()),
+ ranges => {
+ let slices: Vec<_> = ranges
+ .iter()
+ .map(|range| values.slice(range.start, range.len()))
+ .collect();
+ let arrays: Vec<_> = slices.iter().map(|array|
array.as_ref()).collect();
+ arrow_select::concat::concat(&arrays).map_err(normalization_error)?
+ }
+ };
+ Ok((
+ children,
+ OffsetBuffer::new(ScalarBuffer::from(normalized_offsets)),
+ ))
+}
+
fn normalization_error(error: arrow_schema::ArrowError) -> Error {
Error::DataInvalid {
message: format!("Invalid nested Arrow write array: {error}"),
@@ -131,7 +220,7 @@ fn normalization_error(error: arrow_schema::ArrowError) ->
Error {
#[cfg(test)]
mod tests {
use super::*;
- use arrow_array::{Int32Array, StringArray};
+ use arrow_array::{Int32Array, LargeBinaryArray, StringArray};
use arrow_buffer::{OffsetBuffer, ScalarBuffer};
use arrow_schema::Field;
@@ -259,4 +348,161 @@ mod tests {
.unwrap();
assert_eq!(lists.value(1).len(), 2);
}
+
+ #[test]
+ fn matching_schemas_cannot_bypass_nested_map_value_nullability() {
+ let fields = vec![
+ Field::new("key", DataType::Utf8, false),
+ Field::new("value", DataType::LargeBinary, false),
+ ]
+ .into();
+ // PyArrow can pass a MAP through FFI with a NOT NULL value field but
+ // actual null values. Simulate that input without changing its schema.
+ let entries = unsafe {
+ StructArray::new_unchecked(
+ fields,
+ vec![
+ Arc::new(StringArray::from(vec!["key"])),
+ Arc::new(LargeBinaryArray::from(vec![None::<&[u8]>])),
+ ],
+ None,
+ )
+ };
+ let map: ArrayRef = Arc::new(
+ MapArray::try_new(
+ Arc::new(Field::new("entries", entries.data_type().clone(),
false)),
+ OffsetBuffer::new(ScalarBuffer::from(vec![0, 1])),
+ entries,
+ None,
+ false,
+ )
+ .unwrap(),
+ );
+ let row: ArrayRef = Arc::new(StructArray::new(
+ vec![Field::new("payload", map.data_type().clone(), true)].into(),
+ vec![map.clone()],
+ None,
+ ));
+ let list: ArrayRef = Arc::new(ListArray::new(
+ Arc::new(Field::new("element", map.data_type().clone(), true)),
+ OffsetBuffer::new(ScalarBuffer::from(vec![0, 1])),
+ map.clone(),
+ None,
+ ));
+ for input in [map, row, list] {
+ let error = normalize_write_array(&input,
input.data_type()).unwrap_err();
+ assert!(error.to_string().contains("non-nullable"));
+ }
+ }
+
+ #[test]
+ fn matching_empty_row_schema_keeps_its_length_and_nulls() {
+ let input: ArrayRef = Arc::new(StructArray::new_empty_fields(
+ 3,
+ Some(arrow_buffer::NullBuffer::from(vec![true, false, true])),
+ ));
+ let output = normalize_write_array(&input, input.data_type()).unwrap();
+ assert_eq!(output.len(), 3);
+ assert_eq!(output.nulls(), input.nulls());
+ }
+
+ #[test]
+ fn hidden_null_children_do_not_violate_visible_collection_constraints() {
+ let values: ArrayRef = Arc::new(LargeBinaryArray::from(vec![
+ None,
+ Some(b"one".as_slice()),
+ None,
+ Some(b"two".as_slice()),
+ ]));
+ let offsets = OffsetBuffer::new(ScalarBuffer::from(vec![0, 1, 2, 3,
4]));
+ let nulls = Some(NullBuffer::from(vec![false, true, false, true]));
+ let list: ArrayRef = Arc::new(ListArray::new(
+ Arc::new(Field::new("item", DataType::LargeBinary, true)),
+ offsets.clone(),
+ values.clone(),
+ nulls.clone(),
+ ));
+ let expected_list = DataType::List(Arc::new(Field::new(
+ "element",
+ DataType::LargeBinary,
+ false,
+ )));
+ let entries = StructArray::new(
+ vec![
+ Field::new("source_key", DataType::Utf8, false),
+ Field::new("source_value", DataType::LargeBinary, true),
+ ]
+ .into(),
+ vec![
+ Arc::new(StringArray::from(vec!["hidden", "first", "hidden",
"last"])),
+ values,
+ ],
+ None,
+ );
+ let map: ArrayRef = Arc::new(MapArray::new(
+ Arc::new(Field::new(
+ "source_entries",
+ entries.data_type().clone(),
+ false,
+ )),
+ offsets,
+ entries,
+ nulls,
+ false,
+ ));
+ let expected_map = DataType::Map(
+ Arc::new(Field::new(
+ "entries",
+ DataType::Struct(
+ vec![
+ Field::new("key", DataType::Utf8, false),
+ Field::new("value", DataType::LargeBinary, false),
+ ]
+ .into(),
+ ),
+ false,
+ )),
+ false,
+ );
+ for (input, expected) in [(list, expected_list), (map, expected_map)] {
+ for input in [input.clone(), input.slice(1, 1)] {
+ let output = normalize_write_array(&input, &expected).unwrap();
+ assert_eq!(output.data_type(), &expected);
+ assert_eq!(output.len(), input.len());
+ assert_eq!(output.nulls(), input.nulls());
+ let data = output.to_data();
+ data.validate_full().unwrap();
+ }
+ }
+ }
+
+ #[test]
+ fn ancestor_row_nulls_mask_nested_non_nullable_fields() {
+ let inner: ArrayRef = Arc::new(StructArray::new(
+ vec![Field::new("value", DataType::LargeBinary, true)].into(),
+ vec![Arc::new(LargeBinaryArray::from(vec![
+ None,
+ Some(b"ok".as_slice()),
+ ]))],
+ None,
+ ));
+ let input: ArrayRef = Arc::new(StructArray::new(
+ vec![Field::new("inner", inner.data_type().clone(), false)].into(),
+ vec![inner],
+ Some(NullBuffer::from(vec![false, true])),
+ ));
+ let expected = DataType::Struct(
+ vec![Field::new(
+ "inner",
+ DataType::Struct(vec![Field::new("value",
DataType::LargeBinary, false)].into()),
+ false,
+ )]
+ .into(),
+ );
+ let output = normalize_write_array(&input, &expected).unwrap();
+ output.to_data().validate_full().unwrap();
+ assert_eq!(output.len(), 2);
+ assert!(output.is_null(0));
+ assert!(output.is_valid(1));
+ }
}
diff --git a/crates/paimon/testdata/video/video-frame-descriptor-v1.hex
b/crates/paimon/testdata/video/video-frame-descriptor-v1.hex
new file mode 100644
index 00000000..db91bd4e
--- /dev/null
+++ b/crates/paimon/testdata/video/video-frame-descriptor-v1.hex
@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+014d52464f454449561600000073333a2f2f6275636b65742fe8a786e9a2912e6d7034070000000000000063000000000000002a000000000000006a000000000000000800000000000000
diff --git a/crates/paimon/testdata/video/video-v1.hex
b/crates/paimon/testdata/video/video-v1.hex
new file mode 100644
index 00000000..93f04753
--- /dev/null
+++ b/crates/paimon/testdata/video/video-v1.hex
@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+6162635758595a0149464b4f45444956010000000200000000000000000000000100000000000000789c6360c00e78a0f4821e08cd04e503001394013b06026c6b040100020100010106010403000e0602000000020000000500000005000000050000004944454f01
diff --git a/crates/paimon/tests/blob_collection_write_test.rs
b/crates/paimon/tests/blob_collection_write_test.rs
new file mode 100644
index 00000000..0455d9a9
--- /dev/null
+++ b/crates/paimon/tests/blob_collection_write_test.rs
@@ -0,0 +1,270 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod common;
+
+use arrow_array::builder::{LargeBinaryBuilder, ListBuilder, MapBuilder,
StringBuilder};
+use arrow_array::{
+ Array, ArrayRef, Int32Array, LargeBinaryArray, ListArray, MapArray,
RecordBatch,
+};
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs};
+use futures::TryStreamExt;
+use paimon::io::FileRead;
+use paimon::spec::{
+ ArrayType, BlobType, DataType, IntType, MapType, Schema, TableSchema,
VarCharType,
+};
+use paimon::table::Table;
+use std::sync::Arc;
+
+async fn collection_table(options: &[(&str, &str)]) -> Table {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "items",
+ DataType::Array(ArrayType::new(DataType::Blob(BlobType::new()))),
+ )
+ .column(
+ "attrs",
+ DataType::Map(MapType::new(
+ DataType::VarChar(VarCharType::new(100).unwrap()),
+ DataType::Blob(BlobType::new()),
+ )),
+ )
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true");
+ for (key, value) in options {
+ schema = schema.option(*key, *value);
+ }
+ let path = "memory:/collections";
+ let (io, table) = memory_table(path, TableSchema::new(0,
&schema.build().unwrap()));
+ setup_dirs(&io, path).await;
+ persist_table_schema(&io, path, table.schema()).await;
+ table
+}
+
+fn collection_batch(value: &[u8]) -> RecordBatch {
+ let mut list = ListBuilder::new(LargeBinaryBuilder::new());
+ list.values().append_value(b"ignored-before-slice");
+ list.append(true);
+ list.values().append_value(value);
+ list.values().append_null();
+ list.values().append_value(b"");
+ list.append(true);
+ list.append(false);
+ list.append(true);
+ let mut map = MapBuilder::new(None, StringBuilder::new(),
LargeBinaryBuilder::new());
+ map.keys().append_value("ignored");
+ map.values().append_value(b"before-slice");
+ map.append(true).unwrap();
+ for (key, item) in [
+ ("first", Some(value)),
+ ("", Some(b"".as_slice())),
+ ("null", None),
+ ] {
+ map.keys().append_value(key);
+ map.values().append_option(item);
+ }
+ map.append(true).unwrap();
+ map.append(false).unwrap();
+ map.append(true).unwrap();
+ RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![0, 1, 2, 3])) as ArrayRef,
+ ),
+ ("items", Arc::new(list.finish()) as ArrayRef),
+ ("attrs", Arc::new(map.finish()) as ArrayRef),
+ ])
+ .unwrap()
+ .slice(1, 3)
+}
+
+async fn read_batches(table: &Table, descriptor: bool) -> Vec<RecordBatch> {
+ let table = table.copy_with_options(
+ [("blob-as-descriptor".into(), descriptor.to_string())]
+ .into_iter()
+ .collect(),
+ );
+ let builder = table.new_read_builder();
+ let plan = builder.new_scan().plan().await.unwrap();
+ builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap()
+}
+
+#[tokio::test]
+async fn
collections_use_java_blob_files_and_preserve_null_empty_and_sliced_values() {
+ for descriptors in [false, true] {
+ for optimize in [false, true] {
+ let table = collection_table(&[(
+ "data-evolution.write-cols-optimization.enabled",
+ if optimize { "true" } else { "false" },
+ )])
+ .await;
+ let payload = b"payload";
+ let source = "memory:/source";
+ table
+ .file_io()
+ .new_output(source)
+ .unwrap()
+ .write(payload.to_vec().into())
+ .await
+ .unwrap();
+ let descriptor = paimon::spec::BlobDescriptor::new(source.into(),
0, -1).serialize();
+ let input = collection_batch(if descriptors { &descriptor } else {
payload });
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&input).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ let files = &messages[0].new_files;
+ assert_eq!(files.len(), 3);
+ assert!(files[0].file_name.ends_with(".parquet"));
+ assert_eq!(
+ files[0].write_cols,
+ if optimize {
+ None
+ } else {
+ Some(vec!["id".into()])
+ }
+ );
+ for (file, name) in files[1..].iter().zip(["items", "attrs"]) {
+ assert!(file.file_name.ends_with(".blob"), "{}",
file.file_name);
+ assert_eq!(file.write_cols, Some(vec![name.into()]));
+ assert_eq!(file.row_count, 3);
+ }
+ builder.new_commit().commit(messages).await.unwrap();
+ for descriptor_mode in [false, true] {
+ let batches = read_batches(&table, descriptor_mode).await;
+ let batch =
+ arrow_select::concat::concat_batches(&batches[0].schema(),
&batches).unwrap();
+ assert_eq!(batch.num_rows(), 3);
+ let list = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<ListArray>()
+ .unwrap();
+ assert!(list.is_null(1));
+ assert_eq!(list.value_length(2), 0);
+ let items = list.value(0);
+ let values =
items.as_any().downcast_ref::<LargeBinaryArray>().unwrap();
+ assert!(values.is_null(1));
+ let map =
batch.column(2).as_any().downcast_ref::<MapArray>().unwrap();
+ assert!(map.is_null(1));
+ assert_eq!(map.value_length(2), 0);
+ let values_map = map
+ .values()
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap();
+ assert!(values_map.is_null(2));
+ if descriptor_mode {
+ for (bytes, expected) in [
+ (values.value(0), payload.as_slice()),
+ (values.value(2), b"".as_slice()),
+ (values_map.value(0), payload.as_slice()),
+ (values_map.value(1), b"".as_slice()),
+ ] {
+ let desc =
paimon::spec::BlobDescriptor::deserialize(bytes).unwrap();
+ assert!(desc.uri().ends_with(".blob"));
+ assert_eq!(desc.length(), expected.len() as i64);
+ let data = table
+ .file_io()
+ .new_input(desc.uri())
+ .unwrap()
+ .reader()
+ .await
+ .unwrap()
+ .read(desc.offset() as u64..(desc.offset() +
desc.length()) as u64)
+ .await
+ .unwrap();
+ assert_eq!(data.as_ref(), expected);
+ }
+ } else {
+ assert_eq!(values.value(0), payload);
+ assert_eq!(values.value(2), b"");
+ assert_eq!(values_map.value(0), payload);
+ assert_eq!(values_map.value(1), b"");
+ }
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn raw_blob_collection_updates_are_rejected_like_java() {
+ let table = collection_table(&[]).await;
+ for column in ["items", "attrs"] {
+ assert!(
+ table
+ .new_write_builder()
+ .new_data_evolution_writer(vec![column.into()])
+ .is_err(),
+ "raw Blob collection {column} must not be updated"
+ );
+ }
+}
+
+#[test]
+fn video_configuration_matches_java_and_does_not_fall_through_to_blob_writes()
{
+ fn schema(
+ video_type: DataType,
+ options: &[(&str, &str)],
+ primary: bool,
+ ) -> paimon::Result<Schema> {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("video", video_type)
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
+ .option("video-frame-field", "video");
+ if primary {
+ builder = builder.primary_key(vec!["id"]);
+ }
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ builder.build()
+ }
+ let scalar = DataType::Blob(BlobType::new());
+ let valid = schema(scalar.clone(), &[], false).unwrap();
+ let (_, table) = memory_table("memory:/video", TableSchema::new(0,
&valid));
+ assert!(
+ table.new_write_builder().new_write().is_err(),
+ "native video writes must not silently become ordinary .blob files"
+ );
+ assert!(schema(scalar.clone(), &[("blob-descriptor-field", "video")],
false).is_err());
+ assert!(schema(scalar.clone(), &[("blob-view-field", "video")],
false).is_err());
+ assert!(schema(scalar.clone(), &[], true).is_err());
+ assert!(schema(DataType::Array(ArrayType::new(scalar)), &[],
false).is_err());
+ assert!(schema(DataType::Int(IntType::new()), &[], false).is_err());
+ let options = [("video-frame-field".into(), " video, video ,other
".into())]
+ .into_iter()
+ .collect();
+ let options = paimon::spec::CoreOptions::new(&options);
+ assert_eq!(
+ options.video_frame_fields(),
+ ["video".to_string(), "other".to_string()]
+ .into_iter()
+ .collect()
+ );
+ assert_eq!(options.blob_fields(), options.video_frame_fields());
+}
