diff --git a/docs/docs/concepts/spec/fileformat.md b/docs/docs/concepts/spec/fileformat.md index 65a4d5af0c44..ae6c951e95f0 100644 --- a/docs/docs/concepts/spec/fileformat.md +++ b/docs/docs/concepts/spec/fileformat.md @@ -930,4 +930,39 @@ Limitations: 2. BLOB format does not support predicate pushdown. 3. Statistics collection is not supported for BLOB columns. +### Shared BLOB + +Shared BLOB is an independent, versioned format with the `.shared-blob` extension. It reuses the +ordinary scalar BLOB entry encoding for its physical data region, but replaces the positional +footer with two indexes: + +``` ++-----------------------+ +| Physical Blob Entry 1 | Ordinary scalar BLOB entry encoding ++-----------------------+ +| Physical Blob Entry 2 | ++-----------------------+ +| ... | ++-----------------------+ +| Physical Length Index | Delta-Varint compressed record lengths ++-----------------------+ +| Row Reference Index | Delta-Varint compressed physical ordinals ++-----------------------+ +| Physical Index Length | 4 bytes (Little Endian) +| Row Index Length | 4 bytes (Little Endian) +| Magic Number | 4 bytes (0x4C424853, Little Endian) +| Version | 1 byte ++-----------------------+ +``` + +Each non-negative row reference is an ordinal in the physical length index. Multiple logical rows +may contain the same ordinal and therefore resolve to the same physical payload. `-1` represents a +null field and `-2` represents a data-evolution placeholder. Readers validate footer bounds, +physical record lengths, the complete data-region length, and every row-reference ordinal before +returning records. + +Shared BLOB currently supports one scalar BLOB field per file. Physical sharing uses exact input +`BlobDescriptor` identity and is file-local; the format does not contain cross-file references. +Ordinary `.blob` files keep their existing version and layout. + For usage details, configuration options, and examples, see [Blob Type](../../multimodal-table/blob). diff --git a/docs/docs/multimodal-table/blob.mdx b/docs/docs/multimodal-table/blob.mdx index ed92e7a832b1..e85ffb3eb845 100644 --- a/docs/docs/multimodal-table/blob.mdx +++ b/docs/docs/multimodal-table/blob.mdx @@ -49,6 +49,12 @@ When you define a table with a Blob column, Paimon automatically separates the s 1. **Normal Data Files** (e.g., `.parquet`, `.orc`): Store regular columns (INT, STRING, etc.) 2. **Blob Data Files** (`.blob`): Store the actual blob data +For append-only workloads in which many rows refer to the same large object, a scalar BLOB field +can instead use **shared BLOB files** (`.shared-blob`). A shared file stores one physical payload +for each exact `BlobDescriptor` and a compact row-reference index. For example, video-frame rows +can all refer to one MP4 payload while `frame_index` or `timestamp` in the normal data file tells +the decoder which frame to retrieve. + For example, given a table with schema `(id INT, name STRING, picture BLOB)`: ``` @@ -104,6 +110,13 @@ Flink, Spark, and Python. + +
blob-shared-field
+ No + - + String + Names one scalar BLOB field whose logical rows may share payloads in .shared-blob files. This first version supports append-only tables and exact descriptor-backed input only. +
blob-as-descriptor
No @@ -267,6 +280,65 @@ schema = Schema.from_pyarrow_schema( +## Sharing One Video Across Frame Rows + +Use `blob-shared-field` when the table has one row per frame, but the source and storage unit is a +whole video. The option also marks a SQL `BYTES` or `BINARY` column as a BLOB column, so a new +comment directive and a change to the ordinary `.blob` format are not required. + +```sql +CREATE TABLE video_frames ( + episode_id BIGINT, + frame_index INT, + frame_timestamp DOUBLE, + video BYTES +) WITH ( + 'row-tracking.enabled' = 'true', + 'data-evolution.enabled' = 'true', + 'blob-shared-field' = 'video' +); + +-- Every row for this video uses the same serialized BlobDescriptor. +INSERT INTO video_frames +SELECT episode_id, frame_index, frame_timestamp, + sys.path_to_descriptor('file:///data/episode-42.mp4') +FROM frame_metadata; +``` + +The normal data file holds `episode_id`, `frame_index`, and `frame_timestamp`. Its aligned +`.shared-blob` file has two indexes: physical record lengths and logical-row-to-physical-record +references. Several logical positions can therefore resolve to the same `(file, offset, length)` +descriptor. Paimon does not parse MP4 containers or store an MP4 frame offset itself; the frame +columns are interpreted by the application or video decoder. + +Deduplication uses exact descriptor identity (descriptor version, URI, offset, and length), not a +content hash. It is local to each `.shared-blob` file: files are self-contained and never point at +payloads owned by another data file. After a size or row-count target is reached, the rolling +boundary is delayed until a contiguous descriptor group ends, so one video's consecutive frame +rows stay together. This makes the target size soft for large videos. If the same descriptor is +written again after a file has already rolled or a commit/checkpoint has closed the writer, another +physical copy may be stored. + +Compaction reads descriptors, writes a new self-contained `.shared-blob` file, and rebuilds both +indexes, so surviving rows again share one physical payload per exact descriptor in the output. +The ordinary `.blob` format and its behavior are unchanged, and ordinary BLOB fields can coexist +with the shared field in the same table. + +Current restrictions: + +- Append-only tables only; primary-key tables continue to use managed BLOB storage. +- At most one `blob-shared-field` per table. +- The field must be a scalar `BLOB`; `ARRAY` and `MAP` continue to use `.blob`. +- Non-null writes must be exact descriptor-backed `BlobRef` values. Inline `BlobData` is rejected, + because it has no stable identity for sharing. +- A `BlobConsumer` callback is not supported for the shared field. + +Set `blob-as-descriptor=true` when the consumer can decode the referenced MP4 itself. This avoids +materializing the same large payload separately for every selected frame row. + +For Python ingestion and PyTorch `DataLoader` usage, including decoder-session reuse in workers, +see [PyPaimon Multimodal API: Video Frames Sharing One MP4](../pypaimon/multimodal-api#video-frames-sharing-one-mp4). + The comment directive format is `__DIRECTIVE; optional user comment`. Paimon converts `BYTES`/`BINARY` to `BLOB`, `ARRAY`/`ARRAY` to `ARRAY`, and `MAP`/`MAP` to `MAP`. It registers the field in the corresponding option and stores the text after `;` as the column's real comment. Supported directives: diff --git a/docs/docs/primary-key-table/blob-storage.md b/docs/docs/primary-key-table/blob-storage.md index 44ccd638d3a9..20ec7a2029c4 100644 --- a/docs/docs/primary-key-table/blob-storage.md +++ b/docs/docs/primary-key-table/blob-storage.md @@ -37,6 +37,11 @@ This mode stores: For general BLOB concepts and read options, see [BLOB Storage](../multimodal-table/blob). +The append-only `blob-shared-field` mode is deliberately separate from primary-key managed BLOB +storage. It writes self-contained `.shared-blob` files whose logical rows can share a physical +payload, and is not supported on primary-key tables. Enabling it does not change `.managed.blob` +packs, `.blobref` ownership, or their garbage-collection behavior. + ## Create a Table Use `blob-field` to mark scalar, array, or map fields whose payloads should be stored in managed BLOB files. diff --git a/docs/docs/pypaimon/blob.md b/docs/docs/pypaimon/blob.md index dddc32bf2f3f..5d8bc022b9bc 100644 --- a/docs/docs/pypaimon/blob.md +++ b/docs/docs/pypaimon/blob.md @@ -76,6 +76,12 @@ write_builder.new_commit().commit(writer.prepare_commit()) writer.close() ``` +For frame tables in which many logical rows reference one MP4, configure +`blob-shared-field` and write descriptor-backed `Blob` or `BlobDescriptor` +values instead of raw bytes. The high-level multimodal API also provides +single-commit `add_batches` ingestion. See +[Video Frames Sharing One MP4](./multimodal-api#video-frames-sharing-one-mp4). + ## Reading Blob Data ### Batch reading (recommended) @@ -169,3 +175,5 @@ header). This mirrors Java's `Blob.fromBytes(...)`. SQL/Java API - [Data Evolution](./data-evolution) — required for blob tables +- [Multimodal video frames](./multimodal-api#video-frames-sharing-one-mp4) — + shared MP4 writing and PyTorch DataLoader decoding diff --git a/docs/docs/pypaimon/multimodal-api.mdx b/docs/docs/pypaimon/multimodal-api.mdx index 6ca8d85634b3..53aa9b926b05 100644 --- a/docs/docs/pypaimon/multimodal-api.mdx +++ b/docs/docs/pypaimon/multimodal-api.mdx @@ -179,7 +179,10 @@ vector or blob types. `add` accepts `pyarrow.Table`, `pyarrow.RecordBatch`, a list of dictionaries, a dictionary of arrays, or a pandas DataFrame. Input columns are aligned and cast -to the Paimon table schema before writing. +to the Paimon table schema before writing. For a BLOB column, list, dictionary, +and pandas inputs may also contain `Blob` or `BlobDescriptor` objects. A +descriptor-backed `Blob` is copied through its stream; it is not first loaded +into Python memory. ```python docs.add([ @@ -192,6 +195,189 @@ docs.add([ ]) ``` +## Video Frames Sharing One MP4 + +For a frame table, keep the frame mapping in ordinary columns and store one +shared BLOB descriptor in every row for the video. Set `blob-shared-field` to +the video column when creating the table: + +```python +import pyarrow as pa +import pypaimon.multimodal as pm + +frames = conn.create_table( + "video_frames", + schema=pa.schema([ + pa.field("episode_id", pa.int64()), + pa.field("frame_index", pa.int32()), + pa.field("frame_timestamp", pa.float64()), + pa.field("video", pa.large_binary()), + ]), + options={ + "blob-shared-field": "video", + }, +) +``` + +Pass the same descriptor-backed `Blob` for consecutive rows belonging to one +video. `Blob.from_local` creates a reference; `add` serializes that reference +and the shared writer copies the MP4 once for the output file: + +```python +video = pm.Blob.from_local("/data/episode-42.mp4") +fps = 30.0 + +frames.add([ + { + "episode_id": 42, + "frame_index": frame_index, + "frame_timestamp": frame_index / fps, + "video": video, + } + for frame_index in range(1800) +]) +``` + +For larger ingestion jobs, `add_batches` accepts an iterable of any supported +input form and keeps one writer plus one commit for the whole iterable: + +```python +frames.add_batches(frame_batches) +``` + +This is the preferred path when one video's contiguous rows cross Arrow batch +boundaries. An empty iterable is a no-op and does not create a snapshot. If an +error occurs before commit begins, the open writer is aborted; a commit error +has the usual unknown-result semantics and is not retried. + +Remote sources can use an explicit descriptor. Its URI must be readable with +the table's configured `FileIO` credentials: + +```python +video = pm.BlobDescriptor( + "oss://source-bucket/episode-43.mp4", + offset=0, + length=video_size, +) +frames.add([ + { + "episode_id": 43, + "frame_index": frame_index, + "frame_timestamp": frame_index / fps, + "video": video, + } + for frame_index in range(frame_count) +]) +``` + +Keep each video's rows contiguous in the input. Deduplication is by exact +descriptor identity within one `.shared-blob` file, and the writer delays a +pending roll until the current contiguous descriptor group ends. A later call +to `add`, a checkpoint, or a prior file roll can therefore create another +physical copy of the same source video. + +### Read with PyTorch DataLoader + +Install the PyTorch extra, then convert the multimodal scan directly. Unlike a +regular BLOB materialization, `ScanQuery.to_torch` always returns serialized +descriptors for BLOB columns. DataLoader workers receive Paimon splits and open +the selected video ranges themselves: + +```shell +pip install 'pypaimon[torch]' +``` + +```python +from torch.utils.data import DataLoader + +dataset = ( + frames.scan() + .select(["episode_id", "frame_index", "frame_timestamp", "video"]) + .to_torch(streaming=True) +) + +loader = DataLoader( + dataset, + batch_size=32, + num_workers=4, + shuffle=False, +) +``` + +The `video` value above is descriptor bytes, not repeated MP4 bytes. If the +training step needs decoded frames, use `VideoFrameCollator`. It owns a bounded, +process-local LRU cache keyed by exact descriptor, so consecutive frame rows +reuse one decoder session in each DataLoader worker. The decoder is an adapter: +`decoder_factory(stream)` opens any codec library, and +`decode_fn(decoder, row)` locates the frame from `frame_index` or +`frame_timestamp`. + +```python +import av +import torch +import pypaimon.multimodal as pm +from torch.utils.data import DataLoader + + +class SequentialPyAvDecoder: + def __init__(self, stream): + self.container = av.open(stream) + self._reset() + + def _reset(self): + self.frames = iter(self.container.decode(video=0)) + self.next_index = 0 + + def frame(self, index): + if index < self.next_index: + self.container.seek(0) + self._reset() + while self.next_index <= index: + frame = next(self.frames) + self.next_index += 1 + array = frame.to_ndarray(format="rgb24") + return torch.from_numpy(array).permute(2, 0, 1) + + def close(self): + self.container.close() + + +def decode_frame(decoder, row): + return decoder.frame(row["frame_index"]) + + +collator = pm.VideoFrameCollator( + frames, + video_column="video", + decoder_factory=SequentialPyAvDecoder, + decode_fn=decode_frame, + output_column="frame", + max_open_videos=4, +) + +loader = DataLoader( + dataset, + batch_size=32, + num_workers=4, + shuffle=False, + collate_fn=collator, + persistent_workers=True, +) + +for batch in loader: + # batch["frame"] is [N, C, H, W]. + train(batch["frame"], batch["episode_id"]) +``` + +The example decoder additionally requires `pip install av`; PyAV is not a +PyPaimon dependency. + +Define the decoder class and function at module scope when DataLoader uses the +`spawn` multiprocessing method. For sequential video decoding, keep +`shuffle=False`; row-buffer shuffle can turn monotonic frame access into seeks +and require more open decoder sessions. Paimon's worker sharding is split-level, +so it does not duplicate a split across workers. + ## Load HDF5 `MultimodalConnection.load_from_hdf5` streams one or more local or remote HDF5 diff --git a/docs/docs/pypaimon/pytorch.md b/docs/docs/pypaimon/pytorch.md index 9488b20fda1e..c75e8e3f9f6f 100644 --- a/docs/docs/pypaimon/pytorch.md +++ b/docs/docs/pypaimon/pytorch.md @@ -88,6 +88,25 @@ disable a second batching step. Batch streaming does not support `shuffle=True`. Numeric tensors may share read-only Arrow buffers; clone them before in-place mutation. Batch formats currently require `prefetch_concurrency=1`. +## Descriptor-backed video frames + +For a multimodal frame table, use the higher-level scan API: + +```python +dataset = ( + frames.scan() + .select(["episode_id", "frame_index", "video"]) + .to_torch(streaming=True) +) +``` + +This path forces BLOB columns to remain serialized descriptors instead of +materializing the referenced MP4 once per frame row. Use +`pypaimon.multimodal.VideoFrameCollator` as the DataLoader `collate_fn` to open +descriptor ranges and cache decoder sessions per worker. See +[Multimodal API: Video Frames Sharing One MP4](multimodal-api#video-frames-sharing-one-mp4) +for the write path and a complete decoder example. + ## File Format Metadata Cache Reusable PyArrow Dataset metadata is cached across reads. Configure its estimated diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 0494d9d4cdb8..b171781af8c2 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -50,6 +50,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Locale; import java.util.Map; @@ -2685,8 +2686,19 @@ public String toString() { .withDescription( "Specifies column names that should be stored as blob type. " + "This is used when you want to treat a BYTES column as a BLOB. " - + "Fields listed in blob-descriptor-field or blob-view-field " - + "are also treated as BLOB fields."); + + "Fields listed in blob-descriptor-field, blob-view-field, " + + "or blob-shared-field are also treated as BLOB fields."); + + @Immutable + public static final ConfigOption BLOB_SHARED_FIELD = + key("blob-shared-field") + .stringType() + .noDefaultValue() + .withDescription( + "Specifies a scalar BLOB field whose rows may share one physical " + + "payload in '.shared-blob' files. The first version supports " + + "one field on append-only data-evolution tables. Input values " + + "must be exact BlobRef descriptors."); @Immutable public static final ConfigOption BLOB_DESCRIPTOR_FIELD = @@ -3585,6 +3597,11 @@ public Set blobDescriptorField() { return parseCommaSeparatedSet(BLOB_DESCRIPTOR_FIELD); } + /** Resolve the scalar BLOB field stored in shared blob files. */ + public Set blobSharedField() { + return parseCommaSeparatedSet(BLOB_SHARED_FIELD); + } + /** * Resolve blob fields that should be stored as serialized view metadata in data files. * @@ -3953,12 +3970,20 @@ public List sequenceField() { } public static List blobField(Map options) { - String string = options.get(BLOB_FIELD.key()); - if (string == null) { - return Collections.emptyList(); - } + Set fields = new LinkedHashSet<>(); + addCommaSeparatedFields(fields, options.get(BLOB_FIELD.key())); + addCommaSeparatedFields(fields, options.get(BLOB_SHARED_FIELD.key())); + return new ArrayList<>(fields); + } - return Arrays.stream(string.split(",")).map(String::trim).collect(Collectors.toList()); + private static void addCommaSeparatedFields(Set fields, @Nullable String configured) { + if (configured == null) { + return; + } + Arrays.stream(configured.split(",")) + .map(String::trim) + .filter(field -> !field.isEmpty()) + .forEach(fields::add); } public static List blobViewField(Map options) { diff --git a/paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java index 19c5875fd491..14d2ff0e54d6 100644 --- a/paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java @@ -18,6 +18,9 @@ package org.apache.paimon.append; +import org.apache.paimon.data.Blob; +import org.apache.paimon.data.BlobDescriptor; +import org.apache.paimon.data.BlobRef; import org.apache.paimon.data.InternalRow; import org.apache.paimon.fileindex.FileIndexOptions; import org.apache.paimon.format.FileFormat; @@ -50,6 +53,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.function.Supplier; import java.util.stream.Collectors; @@ -110,6 +114,7 @@ public class DedicatedFormatRollingFileWriter vectorStoreWriterFactory; private final long targetFileSize; private final long targetFileRowNum; + private final int sharedBlobFieldIndex; // State management private final List closedWriters; @@ -123,6 +128,8 @@ public class DedicatedFormatRollingFileWriter vectorStoreWriter; private long recordCount = 0; private long currentFileRecordCount = 0; + private @Nullable BlobDescriptor currentSharedBlobGroup; + private boolean pendingGroupAwareRoll; private boolean closed = false; public DedicatedFormatRollingFileWriter( @@ -150,6 +157,7 @@ public DedicatedFormatRollingFileWriter( targetFileRowNum); this.targetFileSize = targetFileSize; this.targetFileRowNum = targetFileRowNum; + this.sharedBlobFieldIndex = sharedBlobFieldIndex(writeSchema, context); this.results = new ArrayList<>(); this.closedWriters = new ArrayList<>(); @@ -207,6 +215,7 @@ public DedicatedFormatRollingFileWriter( blobTargetFileSize, context.blobConsumer(), context.blobInlineFields(), + context.sharedBlobFields(), context.writeNullOnMissingFile(), context.writeNullOnFetchFailure(), context.blobFetchMetricReporter(), @@ -346,6 +355,11 @@ public DedicatedFormatRollingFileWriter( @Override public void write(InternalRow row) throws IOException { try { + BlobDescriptor nextSharedBlobGroup = sharedBlobDescriptor(row); + if (pendingGroupAwareRoll + && !Objects.equals(currentSharedBlobGroup, nextSharedBlobGroup)) { + closeCurrentWriter(); + } if (writerFactory != null && currentWriter == null) { currentWriter = writerFactory.get(); } @@ -366,9 +380,14 @@ public void write(InternalRow row) throws IOException { } recordCount++; currentFileRecordCount++; + currentSharedBlobGroup = nextSharedBlobGroup; if (rollingFile()) { - closeCurrentWriter(); + if (nextSharedBlobGroup == null) { + closeCurrentWriter(); + } else { + pendingGroupAwareRoll = true; + } } } catch (Throwable e) { handleWriteException(e); @@ -478,6 +497,25 @@ private void closeCurrentWriter() throws IOException { // Reset current writer currentWriter = null; currentFileRecordCount = 0; + currentSharedBlobGroup = null; + pendingGroupAwareRoll = false; + } + + private static int sharedBlobFieldIndex( + RowType writeSchema, @Nullable BlobFileContext context) { + if (context == null || context.sharedBlobFields().isEmpty()) { + return -1; + } + String field = context.sharedBlobFields().iterator().next(); + return writeSchema.containsField(field) ? writeSchema.getFieldIndex(field) : -1; + } + + private @Nullable BlobDescriptor sharedBlobDescriptor(InternalRow row) { + if (sharedBlobFieldIndex < 0 || row.isNullAt(sharedBlobFieldIndex)) { + return null; + } + Blob blob = row.getBlob(sharedBlobFieldIndex); + return blob != null && blob.getClass() == BlobRef.class ? blob.toDescriptor() : null; } /** Closes the main writer and returns its metadata. */ diff --git a/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java index 27f40f2ecd1b..6038400328b0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java @@ -22,7 +22,9 @@ import org.apache.paimon.data.BlobFetchMetricReporter; import org.apache.paimon.data.InternalRow; import org.apache.paimon.fileindex.FileIndexOptions; +import org.apache.paimon.format.FileFormat; import org.apache.paimon.format.blob.BlobFileFormat; +import org.apache.paimon.format.blob.SharedBlobFileFormat; import org.apache.paimon.fs.FileIO; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.io.DataFilePathFactory; @@ -30,7 +32,6 @@ import org.apache.paimon.io.RollingFileWriter; import org.apache.paimon.io.RollingFileWriterImpl; import org.apache.paimon.io.RowDataFileWriter; -import org.apache.paimon.io.SingleFileWriter; import org.apache.paimon.manifest.FileSource; import org.apache.paimon.statistics.NoneSimpleColStatsCollector; import org.apache.paimon.statistics.SimpleColStatsCollector; @@ -66,6 +67,7 @@ public MultipleBlobFileWriter( long targetFileSize, @Nullable BlobConsumer blobConsumer, Set blobInlineFields, + Set sharedBlobFields, boolean writeNullOnMissingFile, boolean writeNullOnFetchFailure, BlobFetchMetricReporter blobFetchMetricReporter, @@ -73,36 +75,62 @@ public MultipleBlobFileWriter( RowType blobRowType = new RowType(fieldsInBlobFile(writeSchema, blobInlineFields)); this.blobWriters = new ArrayList<>(); for (String blobFieldName : blobRowType.getFieldNames()) { - BlobFileFormat blobFileFormat = new BlobFileFormat(false, copyBufferSize); - blobFileFormat.setWriteConsumer(blobConsumer); - blobFileFormat.setWriteNullOnMissingFile(writeNullOnMissingFile); - blobFileFormat.setWriteNullOnFetchFailure(writeNullOnFetchFailure); - blobFileFormat.setBlobFetchMetricReporter(blobFetchMetricReporter); + boolean shared = sharedBlobFields.contains(blobFieldName); + FileFormat blobFileFormat; + if (shared) { + if (blobConsumer != null) { + throw new IllegalArgumentException( + "BlobConsumer is not supported for shared blob field '" + + blobFieldName + + "'."); + } + SharedBlobFileFormat format = new SharedBlobFileFormat(false, copyBufferSize); + format.setWriteNullOnMissingFile(writeNullOnMissingFile); + format.setWriteNullOnFetchFailure(writeNullOnFetchFailure); + format.setBlobFetchMetricReporter(blobFetchMetricReporter); + blobFileFormat = format; + } else { + BlobFileFormat format = new BlobFileFormat(false, copyBufferSize); + format.setWriteConsumer(blobConsumer); + format.setWriteNullOnMissingFile(writeNullOnMissingFile); + format.setWriteNullOnFetchFailure(writeNullOnFetchFailure); + format.setBlobFetchMetricReporter(blobFetchMetricReporter); + blobFileFormat = format; + } + RowType fieldType = writeSchema.project(blobFieldName); + Supplier writerFactory = + () -> + new RowDataFileWriter( + fileIO, + RollingFileWriter.createFileWriterContext( + blobFileFormat, + fieldType, + new SimpleColStatsCollector.Factory[] { + NoneSimpleColStatsCollector::new + }, + "none"), + shared + ? pathFactory.newSharedBlobPath() + : pathFactory.newBlobPath(), + fieldType, + schemaId, + seqNumCounterSupplier, + new FileIndexOptions(), + fileSource, + asyncFileWrite, + statsDenseStore, + pathFactory.isExternalPath(), + singletonList(blobFieldName), + null, + null); + RollingFileWriter rollingWriter = + shared + ? new SharedBlobRollingFileWriter<>(writerFactory, targetFileSize) + : new RollingFileWriterImpl<>( + writerFactory, targetFileSize, Long.MAX_VALUE); blobWriters.add( new BlobProjectedFileWriter( - () -> - new RowDataFileWriter( - fileIO, - RollingFileWriter.createFileWriterContext( - blobFileFormat, - writeSchema.project(blobFieldName), - new SimpleColStatsCollector.Factory[] { - NoneSimpleColStatsCollector::new - }, - "none"), - pathFactory.newBlobPath(), - writeSchema.project(blobFieldName), - schemaId, - seqNumCounterSupplier, - new FileIndexOptions(), - fileSource, - asyncFileWrite, - statsDenseStore, - pathFactory.isExternalPath(), - singletonList(blobFieldName), - null, - null), - targetFileSize, + rollingWriter, writeSchema.projectIndexes(singletonList(blobFieldName)))); } } @@ -137,21 +165,27 @@ public List result() throws IOException { List drainAbortExecutors() { List abortExecutors = new ArrayList<>(); for (BlobProjectedFileWriter blobWriter : blobWriters) { - abortExecutors.addAll(blobWriter.writer().drainAbortExecutors()); + abortExecutors.addAll(blobWriter.drainAbortExecutors()); } return abortExecutors; } private static class BlobProjectedFileWriter extends ProjectedFileWriter< - RollingFileWriterImpl, List> { + RollingFileWriter, List> { public BlobProjectedFileWriter( - Supplier> writerFactory, - long targetFileSize, - int[] projection) { - super( - new RollingFileWriterImpl<>(writerFactory, targetFileSize, Long.MAX_VALUE), - projection); + RollingFileWriter writer, int[] projection) { + super(writer, projection); + } + + @SuppressWarnings("unchecked") + private List drainAbortExecutors() { + RollingFileWriter writer = writer(); + if (writer instanceof RollingFileWriterImpl) { + return ((RollingFileWriterImpl) writer) + .drainAbortExecutors(); + } + return ((SharedBlobRollingFileWriter) writer).drainAbortExecutors(); } } } diff --git a/paimon-core/src/main/java/org/apache/paimon/append/SharedBlobRollingFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/append/SharedBlobRollingFileWriter.java new file mode 100644 index 000000000000..751221f57a56 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/append/SharedBlobRollingFileWriter.java @@ -0,0 +1,169 @@ +/* + * 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. + */ + +package org.apache.paimon.append; + +import org.apache.paimon.data.Blob; +import org.apache.paimon.data.BlobDescriptor; +import org.apache.paimon.data.BlobRef; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.io.BundleRecords; +import org.apache.paimon.io.FileWriterAbortExecutor; +import org.apache.paimon.io.RollingFileWriter; +import org.apache.paimon.io.SingleFileWriter; +import org.apache.paimon.utils.Preconditions; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.function.Supplier; + +/** + * Rolls shared blob files only between descriptor groups. + * + *

The target size is soft: after it is reached, all immediately following rows with the same + * exact descriptor remain in the current file. The next different descriptor, NULL, or placeholder + * starts a new file. + */ +class SharedBlobRollingFileWriter implements RollingFileWriter { + + private static final Logger LOG = LoggerFactory.getLogger(SharedBlobRollingFileWriter.class); + + private final Supplier> writerFactory; + private final long targetFileSize; + private final List closedWriters = new ArrayList<>(); + private final List results = new ArrayList<>(); + + private @Nullable SingleFileWriter currentWriter; + private @Nullable BlobDescriptor currentGroup; + private long recordCount; + private boolean pendingRoll; + private boolean closed; + + SharedBlobRollingFileWriter( + Supplier> writerFactory, + long targetFileSize) { + this.writerFactory = writerFactory; + this.targetFileSize = targetFileSize; + } + + @Override + public void write(InternalRow row) throws IOException { + try { + BlobDescriptor nextGroup = descriptor(row); + if (currentWriter != null && pendingRoll && !Objects.equals(currentGroup, nextGroup)) { + closeCurrentWriter(); + } + if (currentWriter == null) { + currentWriter = writerFactory.get(); + } + + currentWriter.write(row); + recordCount++; + currentGroup = nextGroup; + if (currentWriter.reachTargetSize( + recordCount % CHECK_ROLLING_RECORD_CNT == 0, targetFileSize)) { + pendingRoll = true; + } + } catch (Throwable e) { + LOG.warn( + "Exception occurs when writing shared blob file {}. Cleaning up.", + currentWriter == null ? null : currentWriter.path(), + e); + abort(); + throw e; + } + } + + @Override + public void writeBundle(BundleRecords records) throws IOException { + for (InternalRow row : records) { + write(row); + } + } + + @Override + public long recordCount() { + return recordCount; + } + + @Override + public void abort() { + if (currentWriter != null) { + currentWriter.abort(); + currentWriter = null; + } + for (FileWriterAbortExecutor abortExecutor : closedWriters) { + abortExecutor.abort(); + } + } + + @Override + public List result() { + Preconditions.checkState(closed, "Cannot access the results unless close all writers."); + return results; + } + + List drainAbortExecutors() { + Preconditions.checkState(closed, "Cannot drain abort executors unless close all writers."); + List result = new ArrayList<>(closedWriters); + closedWriters.clear(); + return result; + } + + @Override + public void close() throws IOException { + if (closed) { + return; + } + try { + closeCurrentWriter(); + } catch (IOException e) { + abort(); + throw e; + } finally { + closed = true; + } + } + + private void closeCurrentWriter() throws IOException { + if (currentWriter == null) { + return; + } + currentWriter.close(); + currentWriter.abortExecutor().ifPresent(closedWriters::add); + results.add(currentWriter.result()); + currentWriter = null; + currentGroup = null; + pendingRoll = false; + } + + private static @Nullable BlobDescriptor descriptor(InternalRow row) { + if (row.isNullAt(0)) { + return null; + } + Blob blob = row.getBlob(0); + return blob != null && blob.getClass() == BlobRef.class ? blob.toDescriptor() : null; + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionBlobCompactTask.java b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionBlobCompactTask.java index 9056c7b83088..9c337522f8a5 100644 --- a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionBlobCompactTask.java +++ b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionBlobCompactTask.java @@ -23,7 +23,9 @@ import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.fileindex.FileIndexOptions; +import org.apache.paimon.format.FileFormat; import org.apache.paimon.format.blob.BlobFileFormat; +import org.apache.paimon.format.blob.SharedBlobFileFormat; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.io.DataFilePathFactory; import org.apache.paimon.io.FileWriter; @@ -128,7 +130,11 @@ private FileWriter createBlobFileWriter( RowType blobWriteType, String blobFieldName, DataFilePathFactory pathFactory) { - BlobFileFormat blobFileFormat = new BlobFileFormat(false, options.blobCopyBufferSize()); + boolean shared = options.blobSharedField().contains(blobFieldName); + FileFormat blobFileFormat = + shared + ? new SharedBlobFileFormat(false, options.blobCopyBufferSize()) + : new BlobFileFormat(false, options.blobCopyBufferSize()); return new RowDataFileWriter( table.fileIO(), RollingFileWriter.createFileWriterContext( @@ -136,7 +142,7 @@ private FileWriter createBlobFileWriter( blobWriteType, new SimpleColStatsCollector.Factory[] {NoneSimpleColStatsCollector::new}, "none"), - pathFactory.newBlobPath(), + shared ? pathFactory.newSharedBlobPath() : pathFactory.newBlobPath(), blobWriteType, table.schema().id(), () -> new LongCounter(0), diff --git a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignmentPlanner.java b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignmentPlanner.java index 939daa6ff652..45b7c881bce6 100644 --- a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignmentPlanner.java +++ b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignmentPlanner.java @@ -77,6 +77,8 @@ final class DataEvolutionRowIdAssignmentPlanner { private static final BinaryString ROW_ID_FIELD = BinaryString.fromString(SpecialFields.ROW_ID.name()); private static final BinaryString BLOB_FILE_SUFFIX = BinaryString.fromString(".blob"); + private static final BinaryString SHARED_BLOB_FILE_SUFFIX = + BinaryString.fromString(".shared-blob"); private static final BinaryString VECTOR_FILE_MARKER = BinaryString.fromString(".vector."); private static final Projection ADD_IDENTIFIER_PROJECTION = manifestProjection( @@ -455,7 +457,7 @@ private static void readRowRange( } private static int fileOrder(BinaryString fileName) { - if (fileName.endsWith(BLOB_FILE_SUFFIX)) { + if (fileName.endsWith(BLOB_FILE_SUFFIX) || fileName.endsWith(SHARED_BLOB_FILE_SUFFIX)) { return 1; } if (fileName.contains(VECTOR_FILE_MARKER)) { diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DataFilePathFactory.java b/paimon-core/src/main/java/org/apache/paimon/io/DataFilePathFactory.java index 4b8fc56e65a6..98d411dbc796 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/DataFilePathFactory.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/DataFilePathFactory.java @@ -86,6 +86,10 @@ public Path newBlobPath() { return newPathFromName(newFileName(dataFilePrefix, ".blob")); } + public Path newSharedBlobPath() { + return newPathFromName(newFileName(dataFilePrefix, ".shared-blob")); + } + public Path newChangelogPath() { return newPath(changelogFilePrefix); } diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/BlobFileContext.java b/paimon-core/src/main/java/org/apache/paimon/operation/BlobFileContext.java index b1b50e157b82..764a63754ce8 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/BlobFileContext.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/BlobFileContext.java @@ -34,6 +34,7 @@ public class BlobFileContext { private final Set blobDescriptorFields; private final Set blobInlineFields; + private final Set sharedBlobFields; private final boolean writeNullOnMissingFile; private final boolean writeNullOnFetchFailure; private final int copyBufferSize; @@ -44,11 +45,13 @@ public class BlobFileContext { private BlobFileContext( Set blobDescriptorFields, Set blobInlineFields, + Set sharedBlobFields, boolean writeNullOnMissingFile, boolean writeNullOnFetchFailure, int copyBufferSize) { this.blobDescriptorFields = blobDescriptorFields; this.blobInlineFields = blobInlineFields; + this.sharedBlobFields = sharedBlobFields; this.writeNullOnMissingFile = writeNullOnMissingFile; this.writeNullOnFetchFailure = writeNullOnFetchFailure; this.copyBufferSize = copyBufferSize; @@ -74,6 +77,7 @@ public static BlobFileContext create(RowType rowType, CoreOptions options) { return new BlobFileContext( descriptorFields, inlineFields, + options.blobSharedField(), options.blobWriteNullOnMissingFile(), options.blobWriteNullOnFetchFailure(), options.blobCopyBufferSize()); @@ -105,6 +109,10 @@ public Set blobInlineFields() { return blobInlineFields; } + public Set sharedBlobFields() { + return sharedBlobFields; + } + @Nullable public BlobConsumer blobConsumer() { return blobConsumer; diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java index cc0ee88ad702..58ef8412c63c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java @@ -238,6 +238,8 @@ public static void validateTableSchema(TableSchema schema, Set dynamicOp Set blobDescriptorFields = validateBlobDescriptorFields(tableRowType, options); Set blobViewFields = validateBlobViewFields(tableRowType, options, blobDescriptorFields); + validateSharedBlobFields( + schema, tableRowType, options, blobDescriptorFields, blobViewFields); validatePrimaryKeyBlobKeyConfiguration(schema, options); validatePrimaryKeyBlobConfiguration(schema, options); Set blobInlineFields = new HashSet<>(blobDescriptorFields); @@ -1599,6 +1601,45 @@ private static Set validateBlobViewFields( return configured; } + private static void validateSharedBlobFields( + TableSchema schema, + RowType rowType, + CoreOptions options, + Set blobDescriptorFields, + Set blobViewFields) { + Set configured = options.blobSharedField(); + checkArgument( + configured.size() <= 1, + "'%s' currently supports exactly one field, but found %s.", + CoreOptions.BLOB_SHARED_FIELD.key(), + configured); + for (String field : configured) { + checkArgument( + rowType.containsField(field) + && rowType.getTypeAt(rowType.getFieldIndex(field)).getTypeRoot() + == DataTypeRoot.BLOB, + "Field '%s' in '%s' must be a scalar BLOB field in table schema.", + field, + CoreOptions.BLOB_SHARED_FIELD.key()); + checkArgument( + !blobDescriptorFields.contains(field), + "Field '%s' in '%s' can not also be in '%s'.", + field, + CoreOptions.BLOB_SHARED_FIELD.key(), + CoreOptions.BLOB_DESCRIPTOR_FIELD.key()); + checkArgument( + !blobViewFields.contains(field), + "Field '%s' in '%s' can not also be in '%s'.", + field, + CoreOptions.BLOB_SHARED_FIELD.key(), + CoreOptions.BLOB_VIEW_FIELD.key()); + } + checkArgument( + configured.isEmpty() || schema.primaryKeys().isEmpty(), + "'%s' only supports append-only tables.", + CoreOptions.BLOB_SHARED_FIELD.key()); + } + private static void validatePrimaryKeyBlobConfiguration( TableSchema schema, CoreOptions options) { if (schema.primaryKeys().isEmpty()) { diff --git a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java index 93bde01a2f1b..3719a392c3e0 100644 --- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java @@ -270,6 +270,16 @@ public void testBlobCopyBufferSize() { .hasMessageContaining("blob.copy-buffer-size"); } + @Test + public void testSharedBlobFieldIsRecognizedAsBlobField() { + Options options = new Options(); + options.set(CoreOptions.BLOB_FIELD, "image, video"); + options.set(CoreOptions.BLOB_SHARED_FIELD, "video"); + + assertThat(CoreOptions.blobField(options.toMap())).containsExactly("image", "video"); + assertThat(new CoreOptions(options).blobSharedField()).containsExactly("video"); + } + @Test public void testLocalKvDbBlockSize() { Options conf = new Options(); diff --git a/paimon-core/src/test/java/org/apache/paimon/append/BlobTableTest.java b/paimon-core/src/test/java/org/apache/paimon/append/BlobTableTest.java index caa268e0ca82..f4d8edd359b3 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/BlobTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/BlobTableTest.java @@ -22,6 +22,7 @@ import org.apache.paimon.append.dataevolution.DataEvolutionCompactCoordinator; import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask; import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.Blob; import org.apache.paimon.data.BlobArrayPlaceholder; @@ -38,9 +39,11 @@ import org.apache.paimon.data.InternalMap; import org.apache.paimon.data.InternalRow; import org.apache.paimon.data.serializer.InternalRowSerializer; +import org.apache.paimon.format.blob.SharedBlobFileMeta; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.operation.DataEvolutionSplitRead; @@ -1437,6 +1440,136 @@ public void testBlobCompactionSingleField() throws Exception { assertThat(tasks2.stream().anyMatch(task -> task.type() == BLOB)).isFalse(); } + @Test + public void testSharedBlobRollingAndCompaction() throws Exception { + Schema.Builder schemaBuilder = Schema.newBuilder(); + schemaBuilder.column("id", DataTypes.INT()); + schemaBuilder.column("video", DataTypes.BLOB()); + schemaBuilder.option(CoreOptions.TARGET_FILE_SIZE.key(), "1 GB"); + schemaBuilder.option(CoreOptions.BLOB_TARGET_FILE_SIZE.key(), "1 GB"); + schemaBuilder.option(CoreOptions.TARGET_FILE_ROW_NUM.key(), "1"); + schemaBuilder.option(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2"); + schemaBuilder.option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true"); + schemaBuilder.option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true"); + schemaBuilder.option(CoreOptions.BLOB_SHARED_FIELD.key(), "video"); + catalog.createTable(identifier(), schemaBuilder.build(), true); + + byte[] firstBytes = "first-video".getBytes(); + byte[] secondBytes = "second-video".getBytes(); + java.nio.file.Path firstSource = tempPath.resolve("first-source.mp4"); + java.nio.file.Path secondSource = tempPath.resolve("second-source.mp4"); + java.nio.file.Files.write(firstSource, firstBytes); + java.nio.file.Files.write(secondSource, secondBytes); + Blob first = + Blob.fromFile( + LocalFileIO.create(), + new Path(firstSource.toUri()).toString(), + 0, + firstBytes.length); + Blob second = + Blob.fromFile( + LocalFileIO.create(), + new Path(secondSource.toUri()).toString(), + 0, + secondBytes.length); + + writeRows( + getTableDefault(), + Arrays.asList( + GenericRow.of(0, first), + GenericRow.of(1, first), + GenericRow.of(2, first), + GenericRow.of(3, second), + GenericRow.of(4, second))); + + FileStoreTable table = getTableDefault(); + List sharedFiles = liveSharedBlobFiles(table); + assertThat(sharedFiles.size()).isEqualTo(2); + assertThat( + sharedFiles.stream() + .map(DataFileMeta::rowCount) + .sorted() + .collect(Collectors.toList())) + .isEqualTo(Arrays.asList(2L, 3L)); + for (DataFileMeta sharedFile : sharedFiles) { + Path path = + table.store() + .pathFactory() + .createDataFilePathFactory(BinaryRow.EMPTY_ROW, 0) + .toPath(sharedFile); + try (SeekableInputStream in = table.fileIO().newInputStream(path)) { + SharedBlobFileMeta meta = + new SharedBlobFileMeta(in, table.fileIO().getFileSize(path), null); + assertThat(meta.physicalBlobNumber()).isOne(); + assertThat(meta.recordNumber()).isEqualTo(sharedFile.rowCount()); + } + } + assertSharedBlobRows(table, firstBytes, secondBytes); + + DataEvolutionCompactCoordinator coordinator = + new DataEvolutionCompactCoordinator( + table, true, false, table.latestSnapshot().get()); + List tasks = coordinator.plan(); + List compactMessages = new ArrayList<>(); + int blobTaskCount = 0; + for (DataEvolutionCompactTask task : tasks) { + if (task.type() == BLOB) { + blobTaskCount++; + } + compactMessages.add(task.doCompact(table, commitUser)); + } + assertThat(blobTaskCount).isEqualTo(1); + commitDefault(compactMessages); + + table = getTableDefault(); + sharedFiles = liveSharedBlobFiles(table); + assertThat(sharedFiles.size()).isEqualTo(1); + DataFileMeta compacted = sharedFiles.get(0); + Path compactedPath = + table.store() + .pathFactory() + .createDataFilePathFactory(BinaryRow.EMPTY_ROW, 0) + .toPath(compacted); + try (SeekableInputStream in = table.fileIO().newInputStream(compactedPath)) { + SharedBlobFileMeta meta = + new SharedBlobFileMeta(in, table.fileIO().getFileSize(compactedPath), null); + assertThat(meta.recordNumber()).isEqualTo(5); + assertThat(meta.physicalBlobNumber()).isEqualTo(2); + } + assertSharedBlobRows(table, firstBytes, secondBytes); + } + + private List liveSharedBlobFiles(FileStoreTable table) { + return table.store().newScan().plan().files().stream() + .map(ManifestEntry::file) + .filter(file -> file.fileName().endsWith(".shared-blob")) + .collect(Collectors.toList()); + } + + private void assertSharedBlobRows(FileStoreTable table, byte[] firstBytes, byte[] secondBytes) + throws Exception { + Map readOptions = new HashMap<>(); + readOptions.put(CoreOptions.BLOB_AS_DESCRIPTOR.key(), "true"); + Table descriptorTable = table.copy(readOptions); + ReadBuilder readBuilder = descriptorTable.newReadBuilder(); + List rows = new ArrayList<>(); + InternalRowSerializer serializer = new InternalRowSerializer(descriptorTable.rowType()); + try (RecordReader reader = + readBuilder.newRead().createReader(readBuilder.newScan().plan())) { + reader.forEachRemaining(row -> rows.add(serializer.copy(row))); + } + rows.sort((left, right) -> Integer.compare(left.getInt(0), right.getInt(0))); + assertThat(rows.size()).isEqualTo(5); + assertThat(rows.get(0).getBlob(1).toData()).isEqualTo(firstBytes); + assertThat(rows.get(1).getBlob(1).toDescriptor()) + .isEqualTo(rows.get(0).getBlob(1).toDescriptor()); + assertThat(rows.get(2).getBlob(1).toDescriptor()) + .isEqualTo(rows.get(0).getBlob(1).toDescriptor()); + assertThat(rows.get(3).getBlob(1).toData()).isEqualTo(secondBytes); + assertThat(rows.get(4).getBlob(1).toDescriptor()) + .isEqualTo(rows.get(3).getBlob(1).toDescriptor()); + } + @Test public void testPartitionedTableWithBlob() throws Exception { Schema.Builder schemaBuilder = Schema.newBuilder(); diff --git a/paimon-core/src/test/java/org/apache/paimon/io/DataFilePathFactoryTest.java b/paimon-core/src/test/java/org/apache/paimon/io/DataFilePathFactoryTest.java index 266e6b4dc900..7cb927d03787 100644 --- a/paimon-core/src/test/java/org/apache/paimon/io/DataFilePathFactoryTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/io/DataFilePathFactoryTest.java @@ -96,6 +96,24 @@ public void testWithPartition() { .toString()); } + @Test + public void testSharedBlobPathAndFormatIdentifier() { + DataFilePathFactory pathFactory = + new DataFilePathFactory( + new Path(tempDir + "/bucket-123"), + CoreOptions.FILE_FORMAT.defaultValue(), + CoreOptions.DATA_FILE_PREFIX.defaultValue(), + CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(), + CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(), + CoreOptions.FILE_COMPRESSION.defaultValue(), + null); + + Path sharedBlob = pathFactory.newSharedBlobPath(); + assertThat(sharedBlob.getName()).endsWith(".shared-blob"); + assertThat(DataFilePathFactory.formatIdentifier(sharedBlob.getName())) + .isEqualTo("shared-blob"); + } + @Test public void testEntropyInjectWithNoPartition() { EntropyInjectExternalPathProvider externalPathProvider = diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java index 5f5aa2fce0eb..69c2bd27e672 100644 --- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java @@ -459,6 +459,60 @@ public void testPrimaryKeyBlobFileField() { assertThatCode(() -> validateTableSchema(schema)).doesNotThrowAnyException(); } + @Test + public void testSharedBlobFieldValidation() { + List fields = + Arrays.asList( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "video", DataTypes.BLOB()), + new DataField(2, "other_video", DataTypes.BLOB())); + Map options = new HashMap<>(); + options.put(BUCKET.key(), "-1"); + options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true"); + options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true"); + options.put(CoreOptions.BLOB_SHARED_FIELD.key(), "video"); + + TableSchema schema = new TableSchema(1, fields, 10, emptyList(), emptyList(), options, ""); + assertThatCode(() -> validateTableSchema(schema)).doesNotThrowAnyException(); + + options.put(CoreOptions.BLOB_SHARED_FIELD.key(), "video,other_video"); + assertThatThrownBy(() -> validateTableSchema(schema)) + .hasMessageContaining("currently supports exactly one field"); + + options.put(CoreOptions.BLOB_SHARED_FIELD.key(), "video"); + options.put(CoreOptions.BLOB_DESCRIPTOR_FIELD.key(), "video"); + assertThatThrownBy(() -> validateTableSchema(schema)) + .hasMessageContaining("blob-shared-field") + .hasMessageContaining("blob-descriptor-field"); + } + + @Test + public void testSharedBlobRejectsNestedAndPrimaryKeyFields() { + Map options = new HashMap<>(); + options.put(BUCKET.key(), "-1"); + options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true"); + options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true"); + options.put(CoreOptions.BLOB_SHARED_FIELD.key(), "video"); + List nestedFields = + Arrays.asList( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "video", DataTypes.ARRAY(DataTypes.BLOB()))); + TableSchema nested = + new TableSchema(1, nestedFields, 10, emptyList(), emptyList(), options, ""); + assertThatThrownBy(() -> validateTableSchema(nested)) + .hasMessageContaining("must be a scalar BLOB field"); + + options.put(BUCKET.key(), "1"); + List scalarFields = + Arrays.asList( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "video", DataTypes.BLOB())); + TableSchema primaryKey = + new TableSchema(1, scalarFields, 10, emptyList(), singletonList("id"), options, ""); + assertThatThrownBy(() -> validateTableSchema(primaryKey)) + .hasMessageContaining("only supports append-only tables"); + } + @Test public void testPrimaryKeyInlineBlobDoesNotTriggerManagedRestrictions() { // Partial-update supports scalar blob-descriptor-field, managed blob-field, and diff --git a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java index f00877d3494f..15eed056a13d 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java @@ -65,7 +65,8 @@ public BlobFileFormat(boolean blobAsDescriptor, int copyBufferSize) { } public static boolean isBlobFile(String fileName) { - return fileName.endsWith("." + BlobFileFormatFactory.IDENTIFIER); + return fileName.endsWith("." + BlobFileFormatFactory.IDENTIFIER) + || fileName.endsWith("." + SharedBlobFileFormatFactory.IDENTIFIER); } public void setWriteNullOnMissingFile(boolean writeNullOnMissingFile) { diff --git a/paimon-format/src/main/java/org/apache/paimon/format/blob/SharedBlobFileFormat.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/SharedBlobFileFormat.java new file mode 100644 index 000000000000..9c4f5add2452 --- /dev/null +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/SharedBlobFileFormat.java @@ -0,0 +1,164 @@ +/* + * 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. + */ + +package org.apache.paimon.format.blob; + +import org.apache.paimon.data.BlobFetchMetricReporter; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.format.EmptyStatsExtractor; +import org.apache.paimon.format.FileFormat; +import org.apache.paimon.format.FormatReaderFactory; +import org.apache.paimon.format.FormatWriter; +import org.apache.paimon.format.FormatWriterFactory; +import org.apache.paimon.format.SimpleStatsExtractor; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.reader.FileRecordReader; +import org.apache.paimon.statistics.SimpleColStatsCollector; +import org.apache.paimon.types.DataTypeRoot; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.IOUtils; +import org.apache.paimon.utils.Preconditions; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.List; +import java.util.Optional; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** File format in which logical rows may share one physical BLOB payload. */ +public class SharedBlobFileFormat extends FileFormat { + + private final boolean blobAsDescriptor; + private final int copyBufferSize; + private boolean writeNullOnMissingFile; + private boolean writeNullOnFetchFailure; + private BlobFetchMetricReporter blobFetchMetricReporter = BlobFetchMetricReporter.NOOP; + + public SharedBlobFileFormat(boolean blobAsDescriptor, int copyBufferSize) { + super(SharedBlobFileFormatFactory.IDENTIFIER); + this.blobAsDescriptor = blobAsDescriptor; + this.copyBufferSize = copyBufferSize; + } + + public void setWriteNullOnMissingFile(boolean writeNullOnMissingFile) { + this.writeNullOnMissingFile = writeNullOnMissingFile; + } + + public void setWriteNullOnFetchFailure(boolean writeNullOnFetchFailure) { + this.writeNullOnFetchFailure = writeNullOnFetchFailure; + } + + public void setBlobFetchMetricReporter(BlobFetchMetricReporter blobFetchMetricReporter) { + this.blobFetchMetricReporter = blobFetchMetricReporter; + } + + @Override + public FormatReaderFactory createReaderFactory( + RowType dataSchemaRowType, + RowType projectedRowType, + @Nullable List filters) { + return new SharedBlobFormatReaderFactory(blobAsDescriptor, projectedRowType); + } + + @Override + public FormatWriterFactory createWriterFactory(RowType type) { + validateDataFields(type); + return new SharedBlobFormatWriterFactory(type); + } + + @Override + public void validateDataFields(RowType rowType) { + checkArgument( + rowType.getFieldCount() == 1 + && rowType.getTypeAt(0).getTypeRoot() == DataTypeRoot.BLOB, + "SharedBlobFileFormat only supports one scalar BLOB field."); + } + + @Override + public Optional createStatsExtractor( + RowType type, SimpleColStatsCollector.Factory[] statsCollectors) { + return Optional.of(new EmptyStatsExtractor()); + } + + private class SharedBlobFormatWriterFactory implements FormatWriterFactory { + + private final RowType type; + + private SharedBlobFormatWriterFactory(RowType type) { + this.type = type; + } + + @Override + public FormatWriter create(PositionOutputStream out, String compression) { + return new SharedBlobFormatWriter( + out, + type, + writeNullOnMissingFile, + writeNullOnFetchFailure, + blobFetchMetricReporter, + copyBufferSize); + } + } + + private static class SharedBlobFormatReaderFactory implements FormatReaderFactory { + + private final boolean blobAsDescriptor; + private final int fieldCount; + private final int blobIndex; + + private SharedBlobFormatReaderFactory(boolean blobAsDescriptor, RowType projectedRowType) { + this.blobAsDescriptor = blobAsDescriptor; + this.fieldCount = projectedRowType.getFieldCount(); + this.blobIndex = findBlobFieldIndex(projectedRowType); + Preconditions.checkState( + blobIndex >= 0, + "Read type of a shared blob format does not contain a scalar BLOB field."); + } + + @Override + public FileRecordReader createReader(Context context) throws IOException { + FileIO fileIO = context.fileIO(); + Path filePath = context.filePath(); + SeekableInputStream in = fileIO.newInputStream(filePath); + SharedBlobFileMeta fileMeta; + try { + fileMeta = new SharedBlobFileMeta(in, context.fileSize(), context.selection()); + } catch (Exception e) { + IOUtils.closeQuietly(in); + throw e; + } + return new SharedBlobFormatReader( + fileIO, filePath, fileMeta, in, fieldCount, blobIndex, blobAsDescriptor); + } + + private static int findBlobFieldIndex(RowType rowType) { + for (int i = 0; i < rowType.getFieldCount(); i++) { + if (rowType.getTypeAt(i).getTypeRoot() == DataTypeRoot.BLOB) { + return i; + } + } + return -1; + } + } +} diff --git a/paimon-format/src/main/java/org/apache/paimon/format/blob/SharedBlobFileFormatFactory.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/SharedBlobFileFormatFactory.java new file mode 100644 index 000000000000..48d6545b6452 --- /dev/null +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/SharedBlobFileFormatFactory.java @@ -0,0 +1,43 @@ +/* + * 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. + */ + +package org.apache.paimon.format.blob; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.format.FileFormat; +import org.apache.paimon.format.FileFormatFactory; + +/** Factory for {@link SharedBlobFileFormat}. */ +public class SharedBlobFileFormatFactory implements FileFormatFactory { + + public static final String IDENTIFIER = "shared-blob"; + + @Override + public String identifier() { + return IDENTIFIER; + } + + @Override + public FileFormat create(FormatContext formatContext) { + boolean blobAsDescriptor = formatContext.options().get(CoreOptions.BLOB_AS_DESCRIPTOR); + int copyBufferSize = + CoreOptions.checkedBlobCopyBufferSize( + formatContext.options().get(CoreOptions.BLOB_COPY_BUFFER_SIZE).getBytes()); + return new SharedBlobFileFormat(blobAsDescriptor, copyBufferSize); + } +} diff --git a/paimon-format/src/main/java/org/apache/paimon/format/blob/SharedBlobFileMeta.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/SharedBlobFileMeta.java new file mode 100644 index 000000000000..988b3c3d1ca0 --- /dev/null +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/SharedBlobFileMeta.java @@ -0,0 +1,205 @@ +/* + * 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. + */ + +package org.apache.paimon.format.blob; + +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.memory.BytesUtils; +import org.apache.paimon.utils.DeltaVarintCompressor; +import org.apache.paimon.utils.IOUtils; +import org.apache.paimon.utils.RoaringBitmap32; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.Iterator; + +/** Metadata and logical-row mapping of a shared blob file. */ +public class SharedBlobFileMeta { + + private static final int FILE_FOOTER_LENGTH = Integer.BYTES * 3 + Byte.BYTES; + private static final int MIN_RECORD_LENGTH = Integer.BYTES + Long.BYTES + Integer.BYTES; + + private final long[] physicalBlobLengths; + private final long[] physicalBlobOffsets; + private final long[] rowReferences; + private final @Nullable int[] returnedPositions; + + public SharedBlobFileMeta( + SeekableInputStream in, long fileSize, @Nullable RoaringBitmap32 selection) + throws IOException { + if (fileSize < FILE_FOOTER_LENGTH) { + throw corrupt( + "file size %s is smaller than footer size %s.", fileSize, FILE_FOOTER_LENGTH); + } + + in.seek(fileSize - FILE_FOOTER_LENGTH); + byte[] footer = new byte[FILE_FOOTER_LENGTH]; + IOUtils.readFully(in, footer); + int physicalIndexLength = BytesUtils.getInt(footer, 0); + int rowIndexLength = BytesUtils.getInt(footer, Integer.BYTES); + int magic = BytesUtils.getInt(footer, Integer.BYTES * 2); + byte version = footer[Integer.BYTES * 3]; + if (magic != SharedBlobFormatWriter.MAGIC_NUMBER) { + throw corrupt("invalid footer magic %s.", magic); + } + if (version != SharedBlobFormatWriter.VERSION) { + throw new IOException("Unsupported shared blob version: " + version); + } + + long maximumIndexLength = fileSize - FILE_FOOTER_LENGTH; + long totalIndexLength = (long) physicalIndexLength + rowIndexLength; + if (physicalIndexLength < 0 + || rowIndexLength < 0 + || totalIndexLength > maximumIndexLength) { + throw corrupt( + "invalid index lengths %s and %s for file size %s.", + physicalIndexLength, rowIndexLength, fileSize); + } + + long physicalIndexStart = maximumIndexLength - totalIndexLength; + long rowIndexStart = physicalIndexStart + physicalIndexLength; + long[] physicalBlobLengths = + readIndex(in, physicalIndexStart, physicalIndexLength, "physical blob"); + long[] rowReferences = readIndex(in, rowIndexStart, rowIndexLength, "row reference"); + + long[] physicalBlobOffsets = new long[physicalBlobLengths.length]; + long offset = 0; + for (int i = 0; i < physicalBlobLengths.length; i++) { + long blobLength = physicalBlobLengths[i]; + if (blobLength < MIN_RECORD_LENGTH) { + throw corrupt("invalid physical blob length %s at ordinal %s.", blobLength, i); + } + if (blobLength > physicalIndexStart - offset) { + throw corrupt( + "physical blob length %s at ordinal %s exceeds the data region.", + blobLength, i); + } + physicalBlobOffsets[i] = offset; + offset += blobLength; + } + if (offset != physicalIndexStart) { + throw corrupt( + "indexed blobs use %s bytes, but data region contains %s bytes.", + offset, physicalIndexStart); + } + validateRowReferences(rowReferences, physicalBlobLengths.length); + + int[] returnedPositions = null; + if (selection != null) { + long selectionCardinality = selection.getCardinality(); + if (selectionCardinality > rowReferences.length) { + throw new IOException( + String.format( + "Invalid shared blob selection: cardinality %s exceeds row count %s.", + selectionCardinality, rowReferences.length)); + } + int cardinality = (int) selectionCardinality; + returnedPositions = new int[cardinality]; + long[] selectedReferences = new long[cardinality]; + Iterator iterator = selection.iterator(); + for (int i = 0; i < cardinality; i++) { + int position = iterator.next(); + if (position < 0 || position >= rowReferences.length) { + throw new IOException( + String.format( + "Invalid shared blob selection: position %s is outside row count %s.", + position, rowReferences.length)); + } + selectedReferences[i] = rowReferences[position]; + returnedPositions[i] = position; + } + rowReferences = selectedReferences; + } + + this.physicalBlobLengths = physicalBlobLengths; + this.physicalBlobOffsets = physicalBlobOffsets; + this.rowReferences = rowReferences; + this.returnedPositions = returnedPositions; + } + + public boolean isNull(int row) { + return rowReferences[row] == SharedBlobFormatWriter.NULL_REFERENCE; + } + + public boolean isPlaceHolder(int row) { + return rowReferences[row] == SharedBlobFormatWriter.PLACEHOLDER_REFERENCE; + } + + public long blobLength(int row) { + return physicalBlobLengths[physicalOrdinal(row)]; + } + + public long blobOffset(int row) { + return physicalBlobOffsets[physicalOrdinal(row)]; + } + + public int returnedPosition(int currentPosition) { + return returnedPositions == null + ? currentPosition - 1 + : returnedPositions[currentPosition - 1]; + } + + public int recordNumber() { + return rowReferences.length; + } + + public int physicalBlobNumber() { + return physicalBlobLengths.length; + } + + private int physicalOrdinal(int row) { + long reference = rowReferences[row]; + if (reference < 0 || reference > Integer.MAX_VALUE) { + throw new IllegalStateException("Row " + row + " does not reference a physical blob."); + } + return (int) reference; + } + + private static long[] readIndex(SeekableInputStream in, long start, int length, String name) + throws IOException { + in.seek(start); + byte[] bytes = new byte[length]; + IOUtils.readFully(in, bytes); + try { + return DeltaVarintCompressor.decompress(bytes); + } catch (RuntimeException e) { + throw new IOException("Corrupt shared blob file: invalid " + name + " index.", e); + } + } + + private static void validateRowReferences(long[] references, int physicalBlobCount) + throws IOException { + for (int i = 0; i < references.length; i++) { + long reference = references[i]; + if (reference == SharedBlobFormatWriter.NULL_REFERENCE + || reference == SharedBlobFormatWriter.PLACEHOLDER_REFERENCE) { + continue; + } + if (reference < 0 || reference >= physicalBlobCount) { + throw corrupt( + "row %s references physical blob %s, but physical blob count is %s.", + i, reference, physicalBlobCount); + } + } + } + + private static IOException corrupt(String message, Object... args) { + return new IOException("Corrupt shared blob file: " + String.format(message, args)); + } +} diff --git a/paimon-format/src/main/java/org/apache/paimon/format/blob/SharedBlobFormatReader.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/SharedBlobFormatReader.java new file mode 100644 index 000000000000..4a42984ee482 --- /dev/null +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/SharedBlobFormatReader.java @@ -0,0 +1,123 @@ +/* + * 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. + */ + +package org.apache.paimon.format.blob; + +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.reader.FileRecordIterator; +import org.apache.paimon.reader.FileRecordReader; + +import javax.annotation.Nullable; + +import java.io.IOException; + +/** {@link FileRecordReader} for a shared blob file. */ +public class SharedBlobFormatReader implements FileRecordReader { + + private final Path filePath; + private final SharedBlobFileMeta fileMeta; + private final int fieldCount; + private final int blobIndex; + private final BlobElementSerializer.Reader elementReader; + + private boolean returned; + + public SharedBlobFormatReader( + FileIO fileIO, + Path filePath, + SharedBlobFileMeta fileMeta, + @Nullable SeekableInputStream in, + int fieldCount, + int blobIndex, + boolean blobAsDescriptor) { + this.filePath = filePath; + this.fileMeta = fileMeta; + this.fieldCount = fieldCount; + this.blobIndex = blobIndex; + this.elementReader = + BlobElementSerializer.createReader( + new RawBlobElementSerializer(), fileIO, filePath, in, blobAsDescriptor); + } + + @Nullable + @Override + public FileRecordIterator readBatch() throws IOException { + if (returned) { + return null; + } + returned = true; + return new FileRecordIterator() { + + private int currentPosition; + + @Override + public long returnedPosition() { + return fileMeta.returnedPosition(currentPosition); + } + + @Override + public Path filePath() { + return filePath; + } + + @Nullable + @Override + public InternalRow next() { + if (currentPosition >= fileMeta.recordNumber()) { + return null; + } + + Object field; + if (fileMeta.isNull(currentPosition)) { + field = null; + } else if (fileMeta.isPlaceHolder(currentPosition)) { + field = elementReader.placeholder(); + } else { + long payloadPosition = fileMeta.blobOffset(currentPosition) + Integer.BYTES; + long payloadLength = fileMeta.blobLength(currentPosition) - 16; + field = elementReader.read(payloadPosition, payloadLength); + } + currentPosition++; + GenericRow row = new GenericRow(fieldCount); + row.setField(blobIndex, field); + return row; + } + + @Override + public boolean skip() { + if (currentPosition >= fileMeta.recordNumber()) { + return false; + } + currentPosition++; + return true; + } + + @Override + public void releaseBatch() {} + }; + } + + @Override + public void close() throws IOException { + elementReader.close(); + } +} diff --git a/paimon-format/src/main/java/org/apache/paimon/format/blob/SharedBlobFormatWriter.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/SharedBlobFormatWriter.java new file mode 100644 index 000000000000..79dc6c3c9476 --- /dev/null +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/SharedBlobFormatWriter.java @@ -0,0 +1,166 @@ +/* + * 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. + */ + +package org.apache.paimon.format.blob; + +import org.apache.paimon.data.Blob; +import org.apache.paimon.data.BlobDescriptor; +import org.apache.paimon.data.BlobFetchMetricReporter; +import org.apache.paimon.data.BlobPlaceholder; +import org.apache.paimon.data.BlobRef; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.format.FileAwareFormatWriter; +import org.apache.paimon.format.FormatWriter; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.DeltaVarintCompressor; +import org.apache.paimon.utils.LongArrayList; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +import static org.apache.paimon.utils.Preconditions.checkArgument; +import static org.apache.paimon.utils.StreamUtils.intToLittleEndian; + +/** + * {@link FormatWriter} for a shared blob file. + * + *

The data region stores every exact {@link BlobDescriptor} once. A second index maps each + * logical row to a physical blob ordinal, allowing many rows to reference the same payload without + * copying it again. + */ +public class SharedBlobFormatWriter implements FileAwareFormatWriter { + + static final byte VERSION = 1; + static final int MAGIC_NUMBER = 0x4C424853; // "SHBL" in little endian + static final long NULL_REFERENCE = -1L; + static final long PLACEHOLDER_REFERENCE = -2L; + + private final PositionOutputStream out; + private final BlobElementSerializer.Writer elementWriter; + private final LongArrayList physicalBlobLengths; + private final LongArrayList rowReferences; + private final Map physicalBlobs; + + public SharedBlobFormatWriter( + PositionOutputStream out, + RowType type, + boolean writeNullOnMissingFile, + boolean writeNullOnFetchFailure, + BlobFetchMetricReporter blobFetchMetricReporter, + int copyBufferSize) { + checkArgument(type.getFieldCount() == 1, "SharedBlobFormatWriter only supports one field."); + this.out = out; + this.elementWriter = + new RawBlobElementSerializer() + .createWriter( + out, + type.getFieldNames().get(0), + null, + writeNullOnMissingFile, + writeNullOnFetchFailure, + blobFetchMetricReporter, + copyBufferSize); + this.physicalBlobLengths = new LongArrayList(16); + this.rowReferences = new LongArrayList(16); + this.physicalBlobs = new HashMap<>(); + } + + @Override + public void setFile(Path file) { + elementWriter.setFile(file); + } + + @Override + public boolean deleteFileUponAbort() { + return true; + } + + @Override + public void addElement(InternalRow element) throws IOException { + checkArgument( + element.getFieldCount() == 1, "SharedBlobFormatWriter only supports one field."); + if (element.isNullAt(0)) { + rowReferences.add(NULL_REFERENCE); + return; + } + + Blob blob = element.getBlob(0); + if (blob == BlobPlaceholder.INSTANCE) { + rowReferences.add(PLACEHOLDER_REFERENCE); + return; + } + checkArgument( + blob != null && blob.getClass() == BlobRef.class, + "Shared blob fields require an exact BlobRef with a stable descriptor; " + + "inline BlobData and custom Blob implementations are not supported."); + + BlobDescriptor descriptor = blob.toDescriptor(); + Integer physicalBlob = physicalBlobs.get(descriptor); + if (physicalBlob != null) { + rowReferences.add(physicalBlob); + return; + } + + long length = elementWriter.write(element); + if (length == BlobFormatWriter.NULL_LENGTH) { + rowReferences.add(NULL_REFERENCE); + return; + } + + int ordinal = physicalBlobLengths.size(); + physicalBlobLengths.add(length); + physicalBlobs.put(descriptor, ordinal); + rowReferences.add(ordinal); + } + + @Override + public boolean reachTargetSize(boolean suggestedCheck, long targetSize) throws IOException { + return out.getPos() >= targetSize; + } + + @Override + public void close() throws IOException { + Throwable primary = null; + try { + byte[] physicalIndex = DeltaVarintCompressor.compressLongArrayList(physicalBlobLengths); + byte[] rowIndex = DeltaVarintCompressor.compressLongArrayList(rowReferences); + out.write(physicalIndex); + out.write(rowIndex); + out.write(intToLittleEndian(physicalIndex.length)); + out.write(intToLittleEndian(rowIndex.length)); + out.write(intToLittleEndian(MAGIC_NUMBER)); + out.write(VERSION); + } catch (RuntimeException | Error | IOException e) { + primary = e; + throw e; + } finally { + if (primary == null) { + elementWriter.close(); + } else { + try { + elementWriter.close(); + } catch (RuntimeException | Error | IOException suppressed) { + primary.addSuppressed(suppressed); + } + } + } + } +} diff --git a/paimon-format/src/main/resources/META-INF/services/org.apache.paimon.format.FileFormatFactory b/paimon-format/src/main/resources/META-INF/services/org.apache.paimon.format.FileFormatFactory index f34a5af57e48..cfe227cb3f44 100644 --- a/paimon-format/src/main/resources/META-INF/services/org.apache.paimon.format.FileFormatFactory +++ b/paimon-format/src/main/resources/META-INF/services/org.apache.paimon.format.FileFormatFactory @@ -20,4 +20,5 @@ org.apache.paimon.format.csv.CsvFileFormatFactory org.apache.paimon.format.text.TextFileFormatFactory org.apache.paimon.format.json.JsonFileFormatFactory org.apache.paimon.format.blob.BlobFileFormatFactory +org.apache.paimon.format.blob.SharedBlobFileFormatFactory org.apache.paimon.format.row.RowFileFormatFactory diff --git a/paimon-format/src/test/java/org/apache/paimon/format/blob/SharedBlobFileFormatTest.java b/paimon-format/src/test/java/org/apache/paimon/format/blob/SharedBlobFileFormatTest.java new file mode 100644 index 000000000000..5b95a27d3e95 --- /dev/null +++ b/paimon-format/src/test/java/org/apache/paimon/format/blob/SharedBlobFileFormatTest.java @@ -0,0 +1,231 @@ +/* + * 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. + */ + +package org.apache.paimon.format.blob; + +import org.apache.paimon.data.Blob; +import org.apache.paimon.data.BlobData; +import org.apache.paimon.data.BlobPlaceholder; +import org.apache.paimon.data.BlobRef; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.format.FileAwareFormatWriter; +import org.apache.paimon.format.FileFormat; +import org.apache.paimon.format.FormatReaderContext; +import org.apache.paimon.format.FormatReaderFactory; +import org.apache.paimon.format.FormatWriter; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.options.Options; +import org.apache.paimon.reader.FileRecordReader; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.DeltaVarintCompressor; +import org.apache.paimon.utils.RoaringBitmap32; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.List; + +import static org.apache.paimon.utils.StreamUtils.intToLittleEndian; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link SharedBlobFileFormat}. */ +public class SharedBlobFileFormatTest { + + @TempDir java.nio.file.Path tempPath; + + private FileIO fileIO; + private Path file; + private RowType rowType; + + @BeforeEach + public void beforeEach() { + fileIO = LocalFileIO.create(); + file = new Path(tempPath.resolve("data.shared-blob").toUri()); + rowType = RowType.of(DataTypes.BLOB()); + } + + @Test + public void testRowsSharePhysicalBlobByExactDescriptor() throws IOException { + Blob first = sourceBlob("first.mp4", "first-video"); + Blob second = sourceBlob("second.mp4", "second-video"); + + write(first, first, second, first, null, BlobPlaceholder.INSTANCE); + + try (SeekableInputStream in = fileIO.newInputStream(file)) { + SharedBlobFileMeta meta = new SharedBlobFileMeta(in, fileIO.getFileSize(file), null); + assertThat(meta.recordNumber()).isEqualTo(6); + assertThat(meta.physicalBlobNumber()).isEqualTo(2); + assertThat(meta.blobOffset(0)).isEqualTo(meta.blobOffset(1)); + assertThat(meta.blobOffset(0)).isEqualTo(meta.blobOffset(3)); + assertThat(meta.blobLength(0)).isEqualTo(meta.blobLength(1)); + assertThat(meta.blobOffset(2)).isNotEqualTo(meta.blobOffset(0)); + assertThat(meta.isNull(4)).isTrue(); + assertThat(meta.isPlaceHolder(5)).isTrue(); + } + + List rows = read(true, null); + assertThat(rows).hasSize(6); + BlobRef firstRow = (BlobRef) rows.get(0).getBlob(0); + BlobRef secondRow = (BlobRef) rows.get(1).getBlob(0); + BlobRef fourthRow = (BlobRef) rows.get(3).getBlob(0); + assertThat(secondRow.toDescriptor()).isEqualTo(firstRow.toDescriptor()); + assertThat(fourthRow.toDescriptor()).isEqualTo(firstRow.toDescriptor()); + assertThat(firstRow.toData()).isEqualTo("first-video".getBytes()); + assertThat(rows.get(2).getBlob(0).toData()).isEqualTo("second-video".getBytes()); + assertThat(rows.get(4).isNullAt(0)).isTrue(); + assertThat(rows.get(5).getBlob(0)).isSameAs(BlobPlaceholder.INSTANCE); + } + + @Test + public void testSelectionKeepsLogicalRowPositions() throws IOException { + Blob first = sourceBlob("first.mp4", "first-video"); + Blob second = sourceBlob("second.mp4", "second-video"); + write(first, first, second, first); + + RoaringBitmap32 selection = new RoaringBitmap32(); + selection.add(1); + selection.add(3); + + SharedBlobFileFormat format = + new SharedBlobFileFormat(true, BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); + FormatReaderFactory readerFactory = format.createReaderFactory(null, rowType, null); + FormatReaderContext context = + new FormatReaderContext(fileIO, file, fileIO.getFileSize(file), selection, null); + try (FileRecordReader reader = readerFactory.createReader(context)) { + org.apache.paimon.reader.FileRecordIterator iterator = reader.readBatch(); + Blob firstSelected = iterator.next().getBlob(0); + assertThat(iterator.returnedPosition()).isOne(); + Blob secondSelected = iterator.next().getBlob(0); + assertThat(iterator.returnedPosition()).isEqualTo(3L); + assertThat(secondSelected.toDescriptor()).isEqualTo(firstSelected.toDescriptor()); + assertThat(iterator.next()).isNull(); + } + } + + @Test + public void testRejectInlineBlobAndNestedBlobTypes() throws IOException { + SharedBlobFileFormat format = + new SharedBlobFileFormat(true, BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); + assertThatThrownBy( + () -> + format.validateDataFields( + RowType.of(DataTypes.ARRAY(DataTypes.BLOB())))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("scalar BLOB"); + + try (PositionOutputStream out = fileIO.newOutputStream(file, false)) { + FormatWriter writer = format.createWriterFactory(rowType).create(out, null); + assertThatThrownBy( + () -> + writer.addElement( + GenericRow.of(new BlobData("inline".getBytes())))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("exact BlobRef"); + writer.close(); + } + } + + @Test + public void testIndependentFormatRegistrationAndClassification() { + assertThat(FileFormat.fromIdentifier("shared-blob", new Options())) + .isInstanceOf(SharedBlobFileFormat.class); + assertThat(BlobFileFormat.isBlobFile("a.blob")).isTrue(); + assertThat(BlobFileFormat.isBlobFile("a.shared-blob")).isTrue(); + assertThat(BlobFileFormat.isBlobFile("a.parquet")).isFalse(); + } + + @Test + public void testRejectCorruptRowReference() throws IOException { + byte[] physicalIndex = DeltaVarintCompressor.compress(new long[0]); + byte[] rowIndex = DeltaVarintCompressor.compress(new long[] {0}); + byte[] bytes = + new byte[physicalIndex.length + rowIndex.length + Integer.BYTES * 3 + Byte.BYTES]; + int position = 0; + System.arraycopy(physicalIndex, 0, bytes, position, physicalIndex.length); + position += physicalIndex.length; + System.arraycopy(rowIndex, 0, bytes, position, rowIndex.length); + position += rowIndex.length; + position = putInt(bytes, position, physicalIndex.length); + position = putInt(bytes, position, rowIndex.length); + position = putInt(bytes, position, SharedBlobFormatWriter.MAGIC_NUMBER); + bytes[position] = SharedBlobFormatWriter.VERSION; + Files.write(java.nio.file.Paths.get(file.toUri()), bytes); + + assertThatThrownBy( + () -> { + try (SeekableInputStream in = fileIO.newInputStream(file)) { + new SharedBlobFileMeta(in, fileIO.getFileSize(file), null); + } + }) + .isInstanceOf(IOException.class) + .hasMessageContaining( + "row 0 references physical blob 0, but physical blob count is 0"); + } + + private Blob sourceBlob(String name, String value) throws IOException { + byte[] bytes = value.getBytes(); + java.nio.file.Path source = tempPath.resolve(name); + Files.write(source, bytes); + return Blob.fromFile(fileIO, new Path(source.toUri()).toString(), 0, bytes.length); + } + + private void write(Object... blobs) throws IOException { + SharedBlobFileFormat format = + new SharedBlobFileFormat(true, BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); + try (PositionOutputStream out = fileIO.newOutputStream(file, false)) { + FormatWriter writer = format.createWriterFactory(rowType).create(out, null); + ((FileAwareFormatWriter) writer).setFile(file); + for (Object blob : blobs) { + writer.addElement(GenericRow.of(blob)); + } + writer.close(); + } + } + + private List read(boolean blobAsDescriptor, RoaringBitmap32 selection) + throws IOException { + SharedBlobFileFormat format = + new SharedBlobFileFormat( + blobAsDescriptor, BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); + FormatReaderFactory readerFactory = format.createReaderFactory(null, rowType, null); + FormatReaderContext context = + new FormatReaderContext(fileIO, file, fileIO.getFileSize(file), selection, null); + List rows = new ArrayList<>(); + try (FileRecordReader reader = readerFactory.createReader(context)) { + reader.forEachRemaining(rows::add); + } + return rows; + } + + private static int putInt(byte[] target, int position, int value) { + byte[] bytes = intToLittleEndian(value); + System.arraycopy(bytes, 0, target, position, bytes.length); + return position + bytes.length; + } +} diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index e76db7887000..2591d0128960 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -132,6 +132,7 @@ class CoreOptions: "data-evolution.enabled", "index-file-in-data-file-dir", "blob-field", + "blob-shared-field", "blob-descriptor-field", "blob-view-field", "pk-clustering-override", @@ -142,6 +143,7 @@ class CoreOptions: FILE_FORMAT_AVRO: str = "avro" FILE_FORMAT_PARQUET: str = "parquet" FILE_FORMAT_BLOB: str = "blob" + FILE_FORMAT_SHARED_BLOB: str = "shared-blob" FILE_FORMAT_LANCE: str = "lance" FILE_FORMAT_VORTEX: str = "vortex" FILE_FORMAT_ROW: str = "row" @@ -399,6 +401,16 @@ class CoreOptions: .with_description("Comma-separated column names that should be stored as blob type.") ) + BLOB_SHARED_FIELD: ConfigOption[str] = ( + ConfigOptions.key("blob-shared-field") + .string_type() + .no_default_value() + .with_description( + "A scalar BLOB field whose rows may share one physical payload in " + "'.shared-blob' files." + ) + ) + BLOB_DESCRIPTOR_FIELD: ConfigOption[str] = ( ConfigOptions.key("blob-descriptor-field") .string_type() @@ -1235,6 +1247,10 @@ def blob_field(self, default=None): value = self.options.get(CoreOptions.BLOB_FIELD, default) return CoreOptions._parse_field_set(value) + def blob_shared_fields(self, default=None): + value = self.options.get(CoreOptions.BLOB_SHARED_FIELD, default) + return CoreOptions._parse_field_set(value) + def blob_view_resolve_enabled(self, default=True): return self.options.get(CoreOptions.BLOB_VIEW_RESOLVE_ENABLED, default) diff --git a/paimon-python/pypaimon/daft/daft_datasource.py b/paimon-python/pypaimon/daft/daft_datasource.py index cc1dde88d397..ad2f3a8c5d51 100644 --- a/paimon-python/pypaimon/daft/daft_datasource.py +++ b/paimon-python/pypaimon/daft/daft_datasource.py @@ -493,7 +493,8 @@ def _blob_native_covering_files( native reader, or ``None`` if the split must use the pypaimon fallback. A blob table stores each column bunch in its own file: scalar columns in - parquet, BLOB / ARRAY / MAP columns in ``.blob`` files, vector columns in + parquet, BLOB / ARRAY / MAP columns in ``.blob`` or + ``.shared-blob`` files, vector columns in ``.vector`` files, aligned by row id. Reading the base parquet files natively is only correct when every projected data column lives in parquet files that each fully cover the projection over disjoint row-id ranges -- @@ -512,7 +513,7 @@ def _blob_native_covering_files( name = f.file_name write_cols = set(f.write_cols or []) carried = write_cols & projected - if name.endswith(".blob") or ".vector." in name: + if name.endswith((".blob", ".shared-blob")) or ".vector." in name: if carried: return None # a projected column lives in a blob/vector bunch continue diff --git a/paimon-python/pypaimon/manifest/schema/data_file_meta.py b/paimon-python/pypaimon/manifest/schema/data_file_meta.py index 4cc9a3a93519..8a02128c297d 100644 --- a/paimon-python/pypaimon/manifest/schema/data_file_meta.py +++ b/paimon-python/pypaimon/manifest/schema/data_file_meta.py @@ -175,7 +175,7 @@ def copy_without_stats(self) -> 'DataFileMeta': @staticmethod def is_blob_file(file_name: str) -> bool: - return file_name.endswith(".blob") + return file_name.endswith(".blob") or file_name.endswith(".shared-blob") @staticmethod def is_vector_file(file_name: str) -> bool: diff --git a/paimon-python/pypaimon/multimodal/__init__.py b/paimon-python/pypaimon/multimodal/__init__.py index 96afcbbea392..3aa7af018bf5 100644 --- a/paimon-python/pypaimon/multimodal/__init__.py +++ b/paimon-python/pypaimon/multimodal/__init__.py @@ -36,6 +36,8 @@ text_route, vector_route, ) +from pypaimon.multimodal.video import VideoFrameCollator +from pypaimon.table.row.blob import Blob, BlobDescriptor from pypaimon.table.data_evolution_merge_into import ( lit, source_col, @@ -43,6 +45,8 @@ ) __all__ = [ + "Blob", + "BlobDescriptor", "BlobObject", "BlobStore", "Hdf5File", @@ -54,6 +58,7 @@ "PutObjectResult", "TextRoute", "VectorRoute", + "VideoFrameCollator", "connect", "lit", "source_col", diff --git a/paimon-python/pypaimon/multimodal/query.py b/paimon-python/pypaimon/multimodal/query.py index 39d24fd88d48..9046498f7cf8 100644 --- a/paimon-python/pypaimon/multimodal/query.py +++ b/paimon-python/pypaimon/multimodal/query.py @@ -72,8 +72,10 @@ def to_arrow(self): plan = scan.plan() return read_builder.new_read().to_arrow(plan.splits()) - def _configured_read_builder(self): - read_builder = self._table.new_read_builder() + def _configured_read_builder(self, table=None): + read_builder = ( + self._table if table is None else table + ).new_read_builder() if self._predicate is not None: read_builder = read_builder.with_filter(self._predicate) projection = self._effective_projection() @@ -106,6 +108,49 @@ def to_pandas(self): def to_list(self) -> List[dict]: return self.to_arrow().to_pylist() + def to_torch( + self, + streaming: bool = True, + prefetch_concurrency: int = 1, + *, + batch_format: str = "row", + batch_size: Optional[int] = None, + to_tensor_fn: Optional[Callable] = None, + shuffle: bool = False, + seed: int = 0, + buffer_size: int = 1000, + max_buffer_input_splits: int = 10): + """Read this scan as a PyTorch Dataset. + + BLOB columns stay as serialized descriptors so DataLoader workers can + open and decode the referenced payload without materialising it in the + planning process. Use :class:`VideoFrameCollator` as ``collate_fn`` to + reuse one decoder session across rows that reference the same video. + """ + if self._result_factory is not None: + raise TypeError( + "to_torch is only supported on scan(), not search queries." + ) + + from pypaimon.common.options.core_options import CoreOptions + read_table = self._table.copy({ + CoreOptions.BLOB_AS_DESCRIPTOR.key(): "true" + }) + read_builder = self._configured_read_builder(read_table) + splits = read_builder.new_scan().plan().splits() + return read_builder.new_read().to_torch( + splits, + streaming=streaming, + prefetch_concurrency=prefetch_concurrency, + batch_format=batch_format, + batch_size=batch_size, + to_tensor_fn=to_tensor_fn, + shuffle=shuffle, + seed=seed, + buffer_size=buffer_size, + max_buffer_input_splits=max_buffer_input_splits, + ) + def to_ray( self, *, diff --git a/paimon-python/pypaimon/multimodal/table.py b/paimon-python/pypaimon/multimodal/table.py index d6ef7faa4183..53867ed610f3 100644 --- a/paimon-python/pypaimon/multimodal/table.py +++ b/paimon-python/pypaimon/multimodal/table.py @@ -113,6 +113,58 @@ def add(self, data): table_commit.close() return self + def add_batches(self, batches): + """Append an iterable of batches with one writer and one commit. + + Keeping the writer open across batches also keeps a contiguous shared + BLOB descriptor group intact when it crosses an input batch boundary. + """ + try: + iterator = iter(batches) + except TypeError as error: + raise ValueError("batches must be an iterable of input batches.") from error + + target_schema = _target_schema(self.raw_table) + table_write = None + table_commit = None + commit_started = False + try: + for data in iterator: + arrow_table = _to_arrow_table(data, target_schema) + if arrow_table.num_rows == 0: + continue + if table_write is None: + write_builder = self.raw_table.new_batch_write_builder() + table_write = write_builder.new_write() + table_commit = write_builder.new_commit() + table_write.write_arrow(arrow_table) + + close_iterator = getattr(iterator, "close", None) + iterator = None + if close_iterator is not None: + close_iterator() + if table_write is None: + return self + commit_messages = table_write.prepare_commit() + commit_started = True + table_commit.commit(commit_messages) + return self + except BaseException: + if table_write is not None and not commit_started: + table_write.abort() + raise + finally: + if iterator is not None: + close_iterator = getattr(iterator, "close", None) + if close_iterator is not None: + close_iterator() + try: + if table_write is not None: + table_write.close() + finally: + if table_commit is not None: + table_commit.close() + def overwrite(self, data, partition: Optional[Mapping[str, object]] = None): arrow_table = _to_arrow_table(data, _target_schema(self.raw_table)) overwrite_partition = dict(partition) if partition is not None else None @@ -380,6 +432,8 @@ def _blob_columns(table): def _to_arrow_table(data, target_schema=None): + if target_schema is not None: + data = _serialize_blob_values(data, target_schema) if isinstance(data, pa.Table): table = data elif isinstance(data, pa.RecordBatch): @@ -398,6 +452,91 @@ def _to_arrow_table(data, target_schema=None): return _align_to_schema(table, target_schema) +def _serialize_blob_values(data, target_schema): + if isinstance(data, (pa.Table, pa.RecordBatch)): + return data + binary_fields = { + field.name: field.type + for field in target_schema + if _contains_binary(field.type) + } + if not binary_fields: + return data + + if isinstance(data, list): + return [ + { + name: _serialize_blob_value(value, binary_fields.get(name)) + for name, value in row.items() + } + if isinstance(row, Mapping) + else row + for row in data + ] + if isinstance(data, dict): + converted = dict(data) + for name, arrow_type in binary_fields.items(): + if name not in converted: + continue + column = converted[name] + if isinstance(column, (pa.Array, pa.ChunkedArray)): + column = column.to_pylist() + converted[name] = [ + _serialize_blob_value(value, arrow_type) + for value in column + ] + return converted + if ( + hasattr(data, "__dataframe__") + or data.__class__.__module__.startswith("pandas") + ): + converted = data.copy() + for name, arrow_type in binary_fields.items(): + if name in converted.columns: + converted[name] = converted[name].map( + lambda value: _serialize_blob_value(value, arrow_type) + ) + return converted + return data + + +def _contains_binary(arrow_type): + if pa.types.is_binary(arrow_type) or pa.types.is_large_binary(arrow_type): + return True + if pa.types.is_list(arrow_type) or pa.types.is_large_list(arrow_type): + return _contains_binary(arrow_type.value_type) + if pa.types.is_map(arrow_type): + return _contains_binary(arrow_type.item_type) + return False + + +def _serialize_blob_value(value, arrow_type): + if value is None or arrow_type is None: + return value + if pa.types.is_binary(arrow_type) or pa.types.is_large_binary(arrow_type): + from pypaimon.table.row.blob import Blob, BlobDescriptor + if isinstance(value, BlobDescriptor): + return value.serialize() + if isinstance(value, Blob): + try: + return value.to_descriptor().serialize() + except RuntimeError: + return value.to_data() + return value + if pa.types.is_list(arrow_type) or pa.types.is_large_list(arrow_type): + return [ + _serialize_blob_value(element, arrow_type.value_type) + for element in value + ] + if pa.types.is_map(arrow_type): + entries = value.items() if isinstance(value, Mapping) else value + return [ + (key, _serialize_blob_value(element, arrow_type.item_type)) + for key, element in entries + ] + return value + + def _coerce_row_ids(row_ids): if row_ids is None or isinstance(row_ids, (str, bytes)): raise ValueError("row_ids must be an iterable of row id integers.") diff --git a/paimon-python/pypaimon/multimodal/video.py b/paimon-python/pypaimon/multimodal/video.py new file mode 100644 index 000000000000..635310cfd35e --- /dev/null +++ b/paimon-python/pypaimon/multimodal/video.py @@ -0,0 +1,191 @@ +# 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. + +"""PyTorch DataLoader helpers for descriptor-backed video frame rows.""" + +import os +from collections import OrderedDict +from collections.abc import Mapping + +from pypaimon.table.row.blob import Blob, BlobDescriptor + + +class VideoFrameCollator: + """Decode frame rows in a DataLoader worker while reusing video sessions. + + ``decoder_factory`` receives a seekable stream containing exactly one + descriptor-backed video. ``decode_fn`` receives the cached decoder and one + row dictionary. This keeps Paimon independent of a particular video codec + library while allowing PyAV, TorchCodec, or an application decoder to be + plugged in. + + The cache is process-local and keyed by exact ``BlobDescriptor`` identity. + ``collate_fn`` defaults to PyTorch's ``default_collate`` and may be replaced + for decoders that already return batched objects. + """ + + def __init__( + self, + table, + *, + video_column, + decoder_factory, + decode_fn, + output_column="frame", + max_open_videos=8, + collate_fn=None): + if not video_column: + raise ValueError("video_column is required.") + if not callable(decoder_factory): + raise ValueError("decoder_factory must be callable.") + if not callable(decode_fn): + raise ValueError("decode_fn must be callable.") + if ( + isinstance(max_open_videos, bool) + or not isinstance(max_open_videos, int) + or max_open_videos <= 0 + ): + raise ValueError("max_open_videos must be a positive int.") + if collate_fn is not None and not callable(collate_fn): + raise ValueError("collate_fn must be callable or None.") + + raw_table = getattr(table, "raw_table", table) + file_io = getattr(raw_table, "file_io", None) + if file_io is None: + raise ValueError("table must provide raw_table.file_io or file_io.") + + self.file_io = file_io + self.video_column = video_column + self.decoder_factory = decoder_factory + self.decode_fn = decode_fn + self.output_column = output_column + self.max_open_videos = max_open_videos + self.collate_fn = collate_fn + self._decoders = OrderedDict() + self._owner_pid = os.getpid() + + def __call__(self, rows): + self._ensure_process_local_cache() + single_row = isinstance(rows, Mapping) + input_rows = [rows] if single_row else list(rows) + decoded_rows = [self._decode_row(row) for row in input_rows] + if single_row: + return decoded_rows[0] + return self._collate(decoded_rows) + + def close(self): + while self._decoders: + _, resource = self._decoders.popitem(last=False) + self._close_resource(resource) + + def __getstate__(self): + # DataLoader spawn workers must never inherit non-picklable decoder or + # stream state opened in the parent process. + state = self.__dict__.copy() + state["_decoders"] = OrderedDict() + state["_owner_pid"] = None + return state + + def _decode_row(self, row): + if not isinstance(row, Mapping): + raise ValueError("VideoFrameCollator expects row dictionaries.") + if self.video_column not in row: + raise ValueError( + "Video column %r is missing from the row." % self.video_column + ) + + raw = row[self.video_column] + output = dict(row) + if raw is None: + output[self.output_column] = None + return output + if hasattr(raw, "as_py"): + raw = raw.as_py() + if not BlobDescriptor.is_blob_descriptor(raw): + raise ValueError( + "Video column %r must contain serialized BlobDescriptor bytes. " + "Read the table with blob-as-descriptor=true." + % self.video_column + ) + + serialized = bytes(raw) + descriptor = BlobDescriptor.deserialize(serialized) + if descriptor.serialize() != serialized: + raise ValueError( + "Video column %r must contain one exact serialized " + "BlobDescriptor without trailing bytes." % self.video_column + ) + decoder = self._decoder(descriptor) + output[self.output_column] = self.decode_fn(decoder, output) + return output + + def _decoder(self, descriptor): + resource = self._decoders.pop(descriptor, None) + if resource is not None: + self._decoders[descriptor] = resource + return resource[0] + + # Reuse the table's resolved FileIO. Rebuilding a reader from raw URI + # options can drop merged REST/DLF storage credentials in workers. + stream = Blob.from_file( + self.file_io, + descriptor.uri, + descriptor.offset, + descriptor.length, + ).new_input_stream() + try: + decoder = self.decoder_factory(stream) + except Exception: + stream.close() + raise + resource = (decoder, stream) + self._decoders[descriptor] = resource + if len(self._decoders) > self.max_open_videos: + _, evicted = self._decoders.popitem(last=False) + self._close_resource(evicted) + return decoder + + def _collate(self, rows): + if self.collate_fn is not None: + return self.collate_fn(rows) + try: + from torch.utils.data import default_collate + except ImportError as error: + raise ImportError( + "VideoFrameCollator requires PyTorch for its default collate " + "function; install pypaimon[torch] or pass collate_fn=." + ) from error + return default_collate(rows) + + def _ensure_process_local_cache(self): + pid = os.getpid() + if self._owner_pid == pid: + return + # A forked worker owns duplicate descriptors for any inherited file + # handles. Closing them here affects only the worker's copies. + self.close() + self._owner_pid = pid + + @staticmethod + def _close_resource(resource): + decoder, stream = resource + close = getattr(decoder, "close", None) + try: + if close is not None: + close() + finally: + stream.close() diff --git a/paimon-python/pypaimon/read/reader/format_blob_reader.py b/paimon-python/pypaimon/read/reader/format_blob_reader.py index 1e403f53a0ce..c5456d6997d4 100644 --- a/paimon-python/pypaimon/read/reader/format_blob_reader.py +++ b/paimon-python/pypaimon/read/reader/format_blob_reader.py @@ -42,6 +42,10 @@ class FormatBlobReader(RecordBatchReader): NULL_LENGTH = -1 PLACE_HOLDER_LENGTH = -2 + SHARED_BLOB_MAGIC_NUMBER = 0x4C424853 + SHARED_BLOB_VERSION = 1 + SHARED_BLOB_FOOTER_SIZE = 13 + MIN_RECORD_LENGTH = 16 def __init__(self, file_io: FileIO, file_path: str, read_fields: List[str], full_fields: List[DataField], push_down_predicate: Any, blob_as_descriptor: bool, @@ -290,6 +294,10 @@ def close(self): self._input_stream = None def _read_index(self) -> None: + if self.file_path.endswith('.shared-blob'): + self._read_shared_index() + return + f = self._input_stream # Seek to header: last 5 bytes @@ -326,6 +334,85 @@ def _read_index(self) -> None: self.blob_lengths = blob_lengths self.blob_offsets = blob_offsets + def _read_shared_index(self) -> None: + if self._file_size < self.SHARED_BLOB_FOOTER_SIZE: + raise IOError( + "Corrupt shared blob file: file is smaller than its footer." + ) + + f = self._input_stream + f.seek(self._file_size - self.SHARED_BLOB_FOOTER_SIZE) + footer = f.read(self.SHARED_BLOB_FOOTER_SIZE) + if len(footer) != self.SHARED_BLOB_FOOTER_SIZE: + raise IOError("Corrupt shared blob file: cannot read footer.") + physical_index_length, row_index_length, magic, version = struct.unpack( + ' data_and_index_length: + raise IOError( + "Corrupt shared blob file: indexes exceed the file size." + ) + physical_index_start = data_and_index_length - total_index_length + row_index_start = physical_index_start + physical_index_length + + f.seek(physical_index_start) + physical_index = f.read(physical_index_length) + f.seek(row_index_start) + row_index = f.read(row_index_length) + if len(physical_index) != physical_index_length: + raise IOError("Corrupt shared blob file: cannot read physical index.") + if len(row_index) != row_index_length: + raise IOError("Corrupt shared blob file: cannot read row index.") + try: + physical_lengths = DeltaVarintCompressor.decompress(physical_index) + row_references = DeltaVarintCompressor.decompress(row_index) + except RuntimeError as error: + raise IOError("Corrupt shared blob file: invalid index.") from error + + physical_offsets = [] + offset = 0 + for ordinal, length in enumerate(physical_lengths): + if length < self.MIN_RECORD_LENGTH: + raise IOError( + "Corrupt shared blob file: invalid physical blob length " + f"{length} at ordinal {ordinal}." + ) + if length > physical_index_start - offset: + raise IOError( + "Corrupt shared blob file: physical blob exceeds the data region." + ) + physical_offsets.append(offset) + offset += length + if offset != physical_index_start: + raise IOError( + "Corrupt shared blob file: indexed blobs do not cover the data region." + ) + + logical_lengths = [] + logical_offsets = [] + for row, reference in enumerate(row_references): + if reference == self.NULL_LENGTH or reference == self.PLACE_HOLDER_LENGTH: + logical_lengths.append(reference) + logical_offsets.append(-1) + elif reference < 0 or reference >= len(physical_lengths): + raise IOError( + f"Corrupt shared blob file: row {row} references physical blob " + f"{reference}, but physical blob count is {len(physical_lengths)}." + ) + else: + logical_lengths.append(physical_lengths[reference]) + logical_offsets.append(physical_offsets[reference]) + + self.blob_lengths = logical_lengths + self.blob_offsets = logical_offsets + def _apply_row_indices(self, row_indices: Optional[Any]) -> None: if row_indices is None: return diff --git a/paimon-python/pypaimon/read/split_read.py b/paimon-python/pypaimon/read/split_read.py index cc0cca45c955..fea81fe8c781 100644 --- a/paimon-python/pypaimon/read/split_read.py +++ b/paimon-python/pypaimon/read/split_read.py @@ -264,6 +264,7 @@ def file_reader_supplier(self, file: DataFileMeta, for_merge_read: bool, parquet_row_ranges = None if effective_row_ranges is not None: row_index_formats = (CoreOptions.FILE_FORMAT_BLOB, + CoreOptions.FILE_FORMAT_SHARED_BLOB, CoreOptions.FILE_FORMAT_VORTEX, CoreOptions.FILE_FORMAT_LANCE, CoreOptions.FILE_FORMAT_ROW) @@ -329,7 +330,9 @@ def file_reader_supplier(self, file: DataFileMeta, for_merge_read: bool, list(name_to_field.values()), read_arrow_predicate, batch_size=batch_size, nested_name_paths=avro_nested_paths) - elif file_format == CoreOptions.FILE_FORMAT_BLOB: + elif file_format in ( + CoreOptions.FILE_FORMAT_BLOB, + CoreOptions.FILE_FORMAT_SHARED_BLOB): if has_nested: raise NotImplementedError( "Nested-field projection is not supported on BLOB files") diff --git a/paimon-python/pypaimon/schema/schema_manager.py b/paimon-python/pypaimon/schema/schema_manager.py index 928e177a8b8b..f57d408afb92 100644 --- a/paimon-python/pypaimon/schema/schema_manager.py +++ b/paimon-python/pypaimon/schema/schema_manager.py @@ -393,6 +393,19 @@ def _validate_blob_fields( descriptor_fields = core_options.blob_descriptor_fields() view_fields = core_options.blob_view_fields() + shared_fields = core_options.blob_shared_fields() + + if len(shared_fields) > 1: + raise ValueError( + "'blob-shared-field' currently supports exactly one field, but found " + f"{sorted(shared_fields)}." + ) + non_scalar_shared_fields = shared_fields.difference(scalar_blob_field_names) + if non_scalar_shared_fields: + raise ValueError( + "Fields in 'blob-shared-field' must be scalar BLOB fields in schema. " + f"Invalid fields: {sorted(non_scalar_shared_fields)}" + ) all_inline_fields = descriptor_fields.union(view_fields) non_blob_inline_fields = all_inline_fields.difference(scalar_blob_field_names) @@ -421,6 +434,15 @@ def _validate_blob_fields( "Overlapping fields: {}".format(sorted(overlapping_inline_fields)) ) + overlapping_shared_fields = shared_fields.intersection(all_inline_fields) + if overlapping_shared_fields: + raise ValueError( + "Fields in 'blob-shared-field' must not also use descriptor-only or " + "blob-view storage. Overlapping fields: {}".format( + sorted(overlapping_shared_fields) + ) + ) + if blob_field_names: required_options = { CoreOptions.ROW_TRACKING_ENABLED.key(): 'true', diff --git a/paimon-python/pypaimon/tests/data_evolution_row_rolling_test.py b/paimon-python/pypaimon/tests/data_evolution_row_rolling_test.py index 63bebc717310..569feeb68f04 100644 --- a/paimon-python/pypaimon/tests/data_evolution_row_rolling_test.py +++ b/paimon-python/pypaimon/tests/data_evolution_row_rolling_test.py @@ -25,7 +25,7 @@ from pypaimon import CatalogFactory, Schema from pypaimon.common.uri_reader import FileUriReader -from pypaimon.table.row.blob import Blob +from pypaimon.table.row.blob import Blob, BlobDescriptor class DataEvolutionRowRollingTest(unittest.TestCase): @@ -196,6 +196,50 @@ def test_blob_writer_supports_target_file_row_num(self): self.assertEqual([1, 3, 3], blob_rows) self.assertEqual(list(range(7)), self._read_ids(table)) + def test_shared_blob_writer_rolls_between_descriptor_groups(self): + first = os.path.join(self.tempdir, 'first.mp4') + second = os.path.join(self.tempdir, 'second.mp4') + with open(first, 'wb') as output: + output.write(b'first-video') + with open(second, 'wb') as output: + output.write(b'second-video') + first_descriptor = BlobDescriptor(first, 0, len(b'first-video')).serialize() + second_descriptor = BlobDescriptor(second, 0, len(b'second-video')).serialize() + + table = self._create_with_schema( + self.blob_schema, + { + **self.de_options, + 'target-file-row-num': '1', + 'blob-shared-field': 'payload', + }, + ) + data = pa.Table.from_pydict( + { + 'id': list(range(5)), + 'payload': [ + first_descriptor, + first_descriptor, + first_descriptor, + second_descriptor, + second_descriptor, + ], + }, + schema=self.blob_schema, + ) + + files = self._write_files(table, data) + + shared_rows = sorted( + f.row_count for f in files if f.file_name.endswith('.shared-blob') + ) + normal_rows = sorted( + f.row_count for f in files if not f.file_name.endswith('.shared-blob') + ) + self.assertEqual([2, 3], shared_rows) + self.assertEqual([2, 3], normal_rows) + self.assertEqual(list(range(5)), self._read_ids(table)) + def test_blob_consumer_descriptors_survive_abort_after_rolling(self): table = self._create_with_schema( self.blob_schema, diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py b/paimon-python/pypaimon/tests/multimodal_table_test.py index 4a826626c839..8e7df6d5274e 100644 --- a/paimon-python/pypaimon/tests/multimodal_table_test.py +++ b/paimon-python/pypaimon/tests/multimodal_table_test.py @@ -21,6 +21,7 @@ import shutil import tempfile import unittest +from unittest.mock import patch import pyarrow as pa import pypaimon.multimodal as pmm @@ -117,6 +118,180 @@ def test_create_table_defaults_data_evolution_options(self): self.assertEqual(["id", "content", "embedding", "payload"], [field.name for field in table.raw_table.fields]) + def test_add_accepts_descriptor_backed_blob_objects(self): + table = self.conn.create_table( + "video_frames", + schema=_schema({ + "episode_id": pa.int64(), + "frame_index": pa.int32(), + "video": pa.large_binary(), + }), + options=dict(_PARQUET_OPTIONS, **{ + "blob-shared-field": "video", + "blob-as-descriptor": "true", + }), + ) + video_path = os.path.join(self.temp_dir, "episode-42.mp4") + video_bytes = b"fake-mp4-payload" + with open(video_path, "wb") as output: + output.write(video_bytes) + video = pmm.Blob.from_local(video_path) + + table.add([ + {"episode_id": 42, "frame_index": 0, "video": video}, + { + "episode_id": 42, + "frame_index": 1, + "video": video.to_descriptor(), + }, + {"episode_id": 42, "frame_index": 2, "video": video}, + ]) + + rows = table.scan().select( + ["episode_id", "frame_index", "video"] + ).to_list() + self.assertEqual([0, 1, 2], [row["frame_index"] for row in rows]) + descriptors = [ + pmm.BlobDescriptor.deserialize(row["video"]) + for row in rows + ] + self.assertEqual(descriptors[0], descriptors[1]) + self.assertEqual(descriptors[0], descriptors[2]) + self.assertTrue(descriptors[0].uri.endswith(".shared-blob")) + self.assertEqual(len(video_bytes), descriptors[0].length) + + def test_add_batches_keeps_shared_video_group_in_one_commit(self): + from pypaimon.table.row.blob import Blob, BlobDescriptor + + table = self.conn.create_table( + "batched_video_frames", + schema=_schema({ + "episode_id": pa.int64(), + "frame_index": pa.int32(), + "video": pa.large_binary(), + }), + options=dict(_PARQUET_OPTIONS, **{ + "blob-shared-field": "video", + "blob-as-descriptor": "true", + }), + ) + video_path = os.path.join(self.temp_dir, "batched-episode.mp4") + video_bytes = b"batched-video" + with open(video_path, "wb") as output: + output.write(video_bytes) + video = Blob.from_local(video_path) + + table.add_batches([ + [ + {"episode_id": 1, "frame_index": 0, "video": video}, + {"episode_id": 1, "frame_index": 1, "video": video}, + ], + { + "episode_id": [1, 1], + "frame_index": [2, 3], + "video": [video, video], + }, + ]) + + snapshot = table.raw_table.snapshot_manager().get_latest_snapshot() + self.assertEqual(1, snapshot.id) + rows = table.scan().select(["frame_index", "video"]).to_list() + rows.sort(key=lambda row: row["frame_index"]) + descriptors = [ + BlobDescriptor.deserialize(row["video"]) + for row in rows + ] + self.assertEqual(list(range(4)), [row["frame_index"] for row in rows]) + self.assertTrue(all(value == descriptors[0] for value in descriptors)) + self.assertEqual(len(video_bytes), descriptors[0].length) + + def test_add_batches_aborts_before_commit_on_invalid_shared_blob(self): + from pypaimon.table.row.blob import Blob + + table = self.conn.create_table( + "failed_batched_video_frames", + schema=_schema({ + "frame_index": pa.int32(), + "video": pa.large_binary(), + }), + options=dict(_PARQUET_OPTIONS, **{ + "blob-shared-field": "video", + }), + ) + video_path = os.path.join(self.temp_dir, "valid-before-failure.mp4") + with open(video_path, "wb") as output: + output.write(b"valid-video") + + with self.assertRaisesRegex(ValueError, "exact BlobRef"): + table.add_batches([ + [{"frame_index": 0, "video": Blob.from_local(video_path)}], + [{"frame_index": 1, "video": b"inline-is-invalid"}], + ]) + + self.assertIsNone( + table.raw_table.snapshot_manager().get_latest_snapshot() + ) + + def test_add_batches_empty_iterable_is_noop(self): + table = self.conn.create_table( + "empty_batched_frames", + schema=_schema({"frame_index": pa.int32()}), + options=_PARQUET_OPTIONS, + ) + + self.assertIs(table, table.add_batches(iter(()))) + self.assertIsNone( + table.raw_table.snapshot_manager().get_latest_snapshot() + ) + + def test_scan_to_torch_keeps_blob_descriptors(self): + table = self.conn.create_table( + "torch_video_frames", + schema=_schema({ + "frame_index": pa.int32(), + "video": pa.large_binary(), + }), + options=dict(_PARQUET_OPTIONS, **{ + "blob-shared-field": "video", + "blob-as-descriptor": "false", + }), + ) + from pypaimon.table.row.blob import Blob + video_path = os.path.join(self.temp_dir, "torch-episode.mp4") + with open(video_path, "wb") as output: + output.write(b"fake-video") + table.add([{ + "frame_index": 0, + "video": Blob.from_local(video_path), + }]) + sentinel = object() + captured = {} + + def fake_to_torch(table_read, splits, **kwargs): + captured["descriptor"] = table_read.table.options.blob_as_descriptor() + captured["splits"] = splits + captured["kwargs"] = kwargs + return sentinel + + with patch( + "pypaimon.read.table_read.TableRead.to_torch", + autospec=True, + side_effect=fake_to_torch, + ): + result = table.scan().select( + ["frame_index", "video"] + ).to_torch( + streaming=True, + prefetch_concurrency=2, + shuffle=False, + ) + + self.assertIs(sentinel, result) + self.assertTrue(captured["descriptor"]) + self.assertTrue(captured["splits"]) + self.assertEqual(True, captured["kwargs"]["streaming"]) + self.assertEqual(2, captured["kwargs"]["prefetch_concurrency"]) + def test_create_table_uses_options_and_partitioned(self): table = self.conn.create_table( "docs", diff --git a/paimon-python/pypaimon/tests/multimodal_video_test.py b/paimon-python/pypaimon/tests/multimodal_video_test.py new file mode 100644 index 000000000000..9f952fa7b719 --- /dev/null +++ b/paimon-python/pypaimon/tests/multimodal_video_test.py @@ -0,0 +1,175 @@ +# 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. + +import io +import os +import tempfile +import unittest +from types import SimpleNamespace + +from pypaimon.common.file_io import FileIO +from pypaimon.multimodal import VideoFrameCollator +from pypaimon.table.row.blob import BlobDescriptor + + +class _Decoder: + + def __init__(self, stream, calls): + self._stream = stream + self._calls = calls + self.closed = False + + def decode(self, row): + self._stream.seek(0) + return self._stream.read(), row["frame_index"] + + def close(self): + self.closed = True + self._calls.append("close") + + +class VideoFrameCollatorTest(unittest.TestCase): + + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + self.file_io = FileIO.get("file://" + self.temp_dir.name, {}) + self.table = SimpleNamespace( + raw_table=SimpleNamespace(file_io=self.file_io) + ) + + def tearDown(self): + self.temp_dir.cleanup() + + def test_reuses_decoder_for_rows_with_same_descriptor(self): + descriptor = self._descriptor("episode-1.mp4", b"video-one") + factory_calls = [] + + def factory(stream): + factory_calls.append("open") + return _Decoder(stream, factory_calls) + + collator = VideoFrameCollator( + self.table, + video_column="video", + decoder_factory=factory, + decode_fn=lambda decoder, row: decoder.decode(row), + collate_fn=lambda rows: rows, + ) + try: + result = collator([ + {"frame_index": 0, "video": descriptor}, + {"frame_index": 1, "video": descriptor}, + ]) + finally: + collator.close() + + self.assertEqual(["open", "close"], factory_calls) + self.assertEqual( + [(b"video-one", 0), (b"video-one", 1)], + [row["frame"] for row in result], + ) + self.assertEqual(descriptor, result[0]["video"]) + + def test_evicts_least_recently_used_decoder(self): + descriptors = [ + self._descriptor("episode-%d.mp4" % index, bytes([index])) + for index in range(3) + ] + factory_calls = [] + + def factory(stream): + factory_calls.append("open") + return _Decoder(stream, factory_calls) + + collator = VideoFrameCollator( + self.table, + video_column="video", + decoder_factory=factory, + decode_fn=lambda decoder, row: decoder.decode(row), + max_open_videos=2, + collate_fn=lambda rows: rows, + ) + try: + collator([ + {"frame_index": index, "video": descriptor} + for index, descriptor in enumerate(descriptors) + ]) + self.assertEqual(3, factory_calls.count("open")) + self.assertEqual(1, factory_calls.count("close")) + + collator([{"frame_index": 3, "video": descriptors[0]}]) + self.assertEqual(4, factory_calls.count("open")) + self.assertEqual(2, factory_calls.count("close")) + finally: + collator.close() + + self.assertEqual(4, factory_calls.count("close")) + + def test_rejects_non_descriptor_video_cell(self): + collator = VideoFrameCollator( + self.table, + video_column="video", + decoder_factory=lambda stream: _Decoder(stream, []), + decode_fn=lambda decoder, row: decoder.decode(row), + collate_fn=lambda rows: rows, + ) + valid = self._descriptor("valid.mp4", b"video") + for value in (b"inline-mp4", valid + b"trailing"): + with self.subTest(value=value): + with self.assertRaisesRegex(ValueError, "serialized BlobDescriptor"): + collator([{"frame_index": 0, "video": value}]) + + def test_reuses_resolved_table_file_io(self): + class ResolvedFileIO: + + @property + def uri_reader_factory(self): + raise AssertionError("must not rebuild a URI reader") + + def new_input_stream(self, path): + self.path = path + return io.BytesIO(b"resolved-video") + + file_io = ResolvedFileIO() + table = SimpleNamespace(raw_table=SimpleNamespace(file_io=file_io)) + descriptor = BlobDescriptor( + "oss://bucket/internal.shared-blob", 0, 14 + ).serialize() + collator = VideoFrameCollator( + table, + video_column="video", + decoder_factory=lambda stream: _Decoder(stream, []), + decode_fn=lambda decoder, row: decoder.decode(row), + collate_fn=lambda rows: rows, + ) + try: + result = collator([{"frame_index": 2, "video": descriptor}]) + finally: + collator.close() + + self.assertEqual("oss://bucket/internal.shared-blob", file_io.path) + self.assertEqual((b"resolved-video", 2), result[0]["frame"]) + + def _descriptor(self, name, data): + path = os.path.join(self.temp_dir.name, name) + with open(path, "wb") as output: + output.write(data) + return BlobDescriptor(path, 0, len(data)).serialize() + + +if __name__ == "__main__": + unittest.main() diff --git a/paimon-python/pypaimon/tests/shared_blob_test.py b/paimon-python/pypaimon/tests/shared_blob_test.py new file mode 100644 index 000000000000..6d8ba1d62d27 --- /dev/null +++ b/paimon-python/pypaimon/tests/shared_blob_test.py @@ -0,0 +1,144 @@ +# 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. + +import struct +import tempfile +import unittest +from pathlib import Path + +from pypaimon.common.delta_varint_compressor import DeltaVarintCompressor +from pypaimon.common.options import Options +from pypaimon.filesystem.local_file_io import LocalFileIO +from pypaimon.read.reader.format_blob_reader import FormatBlobReader +from pypaimon.schema.data_types import AtomicType, DataField +from pypaimon.table.row.blob import Blob, BlobData, BlobDescriptor +from pypaimon.table.row.generic_row import GenericRow +from pypaimon.table.row.row_kind import RowKind +from pypaimon.write.blob_format_writer import BlobFormatWriter +from pypaimon.write.shared_blob_format_writer import SharedBlobFormatWriter + + +class SharedBlobFormatTest(unittest.TestCase): + + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + self.root = Path(self.temp_dir.name) + self.file_io = LocalFileIO(str(self.root), Options({})) + self.field = DataField(0, "video", AtomicType("BLOB")) + + def tearDown(self): + self.temp_dir.cleanup() + + def test_duplicate_descriptors_share_physical_payload(self): + first = self._source_blob("first.mp4", b"first-video") + second = self._source_blob("second.mp4", b"second-video") + target = (self.root / "data.shared-blob").as_uri() + + writer = SharedBlobFormatWriter( + self.file_io.new_output_stream(target), file_path=target + ) + for value in (first, first, second, first, None): + writer.add_element(GenericRow([value], [self.field], RowKind.INSERT)) + self.assertEqual(2, writer.physical_blob_count) + writer.close() + with self.file_io.new_input_stream(target) as stream: + self.assertEqual( + BlobFormatWriter.MAGIC_NUMBER, + struct.unpack(' None: + if not hasattr(row, 'values') or len(row.values) != 1: + raise ValueError("SharedBlobFormatWriter only supports one field") + if not is_blob_type(row.fields[0].type): + raise ValueError("SharedBlobFormatWriter only supports one scalar BLOB field") + + blob_value = row.values[0] + if blob_value is None: + self._row_references.append(self.NULL_REFERENCE) + return + if blob_value is Blob.PLACE_HOLDER: + self._row_references.append(self.PLACE_HOLDER_REFERENCE) + return + if type(blob_value) is not BlobRef: + raise ValueError( + "Shared blob fields require an exact BlobRef with a stable descriptor; " + "inline BlobData and custom Blob implementations are not supported." + ) + + descriptor = blob_value.to_descriptor() + ordinal = self._physical_blobs.get(descriptor) + if ordinal is None: + super().add_blob(row.fields[0].name, blob_value) + ordinal = len(self.lengths) - 1 + self._physical_blobs[descriptor] = ordinal + self._row_references.append(ordinal) + + @property + def physical_blob_count(self) -> int: + return len(self.lengths) + + def close(self) -> None: + physical_index = DeltaVarintCompressor.compress(self.lengths) + row_index = DeltaVarintCompressor.compress(self._row_references) + self.output_stream.write(physical_index) + self.output_stream.write(row_index) + self.output_stream.write( + struct.pack( + ' bool: @@ -137,6 +161,31 @@ def close_current_writer(self): self.current_writer = None self.current_file_path = None + self._current_shared_group = None + self._pending_shared_roll = False + + def _roll_before_shared_group(self, next_group): + if ( + self.shared + and self.current_writer is not None + and self._pending_shared_roll + and self._current_shared_group != next_group + ): + self.close_current_writer() + + @staticmethod + def _shared_descriptor(value): + if hasattr(value, 'as_py'): + value = value.as_py() + if value is None or value is Blob.PLACE_HOLDER: + return None + if type(value) is BlobRef: + return value.to_descriptor() + if isinstance(value, (bytes, bytearray)): + raw = bytes(value) + if BlobDescriptor.is_blob_descriptor(raw): + return BlobDescriptor.deserialize(raw) + return None def _write_data_to_file(self, data): """ @@ -253,6 +302,8 @@ def abort(self): logger.warning(f"Error aborting blob writer: {e}", exc_info=e) self.current_writer = None self.current_file_path = None + self._current_shared_group = None + self._pending_shared_roll = False if not self.delete_file_upon_abort(): self._buffer.reset() self.committed_files.clear() diff --git a/paimon-python/pypaimon/write/writer/dedicated_format_writer.py b/paimon-python/pypaimon/write/writer/dedicated_format_writer.py index 001b0f9f692f..dfbe450cbf44 100644 --- a/paimon-python/pypaimon/write/writer/dedicated_format_writer.py +++ b/paimon-python/pypaimon/write/writer/dedicated_format_writer.py @@ -31,7 +31,7 @@ is_blob_file_field, is_blob_type, ) -from pypaimon.table.row.blob import BlobConsumer +from pypaimon.table.row.blob import Blob, BlobConsumer, BlobDescriptor, BlobRef from pypaimon.table.row.generic_row import GenericRow from pypaimon.write.row_utils import ( require_columns, @@ -49,7 +49,7 @@ class DedicatedFormatWriter(DataWriter): Splits incoming data three ways: - Normal columns → standard data files (.parquet / .orc / .vortex / …) - - Blob columns (large_binary) → .blob files + - Blob columns (large_binary) → .blob or .shared-blob files - Vector columns (when vector.file.format is configured) → .vector. files This mirrors Java's DedicatedFormatRollingFileWriter. @@ -71,8 +71,15 @@ def __init__(self, table, partition: Tuple, bucket: int, max_seq_number: int, op self.blob_column_names = self._get_blob_columns_from_schema() self.blob_descriptor_fields = CoreOptions.blob_descriptor_fields(self.options) self.blob_view_fields = CoreOptions.blob_view_fields(self.options) + self.shared_blob_fields = CoreOptions.blob_shared_fields(self.options) self.blob_inline_fields = self.blob_descriptor_fields.union(self.blob_view_fields) + if len(self.shared_blob_fields) > 1: + raise ValueError("'blob-shared-field' currently supports exactly one field.") + self.shared_blob_column = ( + next(iter(self.shared_blob_fields)) if self.shared_blob_fields else None + ) + unknown_descriptor_fields = self.blob_descriptor_fields.difference( set(self.blob_column_names) ) @@ -92,7 +99,7 @@ def __init__(self, table, partition: Tuple, bucket: int, max_seq_number: int, op f"Invalid inline blob fields: {sorted(inline_nested_blob_fields)}" ) - # Blob fields that should still be written to `.blob` files. + # Blob fields that should still be written to dedicated BLOB files. self.blob_file_column_names = [ col for col in self.blob_column_names if col not in self.blob_inline_fields ] @@ -125,6 +132,8 @@ def __init__(self, table, partition: Tuple, bucket: int, max_seq_number: int, op self.normal_column_names = [ col for col in all_column_names if col not in dedicated_set ] + if self.shared_blob_column not in self.blob_file_column_names: + self.shared_blob_column = None normal_name_set = set(self.normal_column_names) self.normal_columns = [ field for field in self.table.table_schema.fields if field.name in normal_name_set @@ -134,6 +143,8 @@ def __init__(self, table, partition: Tuple, bucket: int, max_seq_number: int, op # State management for blob writer self.record_count = 0 self.closed = False + self._current_shared_blob_group = None + self._pending_shared_group_roll = False # Normal columns are buffered separately from the blob and vector # columns, which their own writers own. @@ -156,6 +167,7 @@ def __init__(self, table, partition: Tuple, bucket: int, max_seq_number: int, op blob_column=blob_column, options=options, blob_consumer=blob_consumer, + shared=blob_column in self.shared_blob_fields, ) # Initialize vector writer when vector.file.format is configured. @@ -210,6 +222,15 @@ def write(self, data: pa.RecordBatch): # writer, or the unfinished flush would lose its chance to be retried. self._require_finished_flush() try: + if self.shared_blob_column is not None: + for index in range(data.num_rows): + row = data.slice(index, 1) + next_group = self._shared_descriptor_from_batch(row) + self._roll_before_shared_group(next_group) + self._current_shared_blob_group = next_group + self._write_batch(row) + return + offset = 0 # _write_batch keeps normal/blob/vector pending rows in lockstep # and closes all writers when the shared row limit is reached. @@ -253,8 +274,7 @@ def _write_batch(self, data: pa.RecordBatch): # Check if normal data rolling is needed if self._should_roll_normal(): - # When normal data rolls, close both writers and fetch blob metadata - self._close_current_writers() + self._roll_or_defer_for_shared_group() def write_row(self, row): self._require_finished_flush() @@ -268,6 +288,11 @@ def write_row(self, row): ) require_columns(values_by_name, required_columns, "write_row") + if self.shared_blob_column is not None: + next_group = self._shared_descriptor(values_by_name[self.shared_blob_column]) + self._roll_before_shared_group(next_group) + self._current_shared_blob_group = next_group + if self.normal_column_names: normal_values = dict(values_by_name) for field_name in self.normal_column_names: @@ -299,7 +324,7 @@ def write_row(self, row): self.record_count += 1 if self._should_roll_normal(): - self._close_current_writers() + self._roll_or_defer_for_shared_group() except Exception as e: logger.error("Exception occurs when writing row. Cleaning up.", exc_info=e) @@ -485,6 +510,44 @@ def _should_roll_normal(self) -> bool: # Check if normal data exceeds target size return self._normal_buffer.nbytes > self.target_file_size + def _roll_or_defer_for_shared_group(self): + if ( + self.shared_blob_column is not None + and self._current_shared_blob_group is not None + ): + self._pending_shared_group_roll = True + else: + self._close_current_writers() + + def _roll_before_shared_group(self, next_group): + if ( + self._pending_shared_group_roll + and self._current_shared_blob_group != next_group + ): + self._close_current_writers() + + def _shared_descriptor_from_batch(self, data: pa.RecordBatch): + column_index = data.schema.get_field_index(self.shared_blob_column) + if column_index < 0: + raise KeyError( + f"Column '{self.shared_blob_column}' was not found in the record batch." + ) + return self._shared_descriptor(data.column(column_index)[0]) + + @staticmethod + def _shared_descriptor(value): + if hasattr(value, 'as_py'): + value = value.as_py() + if value is None or value is Blob.PLACE_HOLDER: + return None + if type(value) is BlobRef: + return value.to_descriptor() + if isinstance(value, (bytes, bytearray)): + raw = bytes(value) + if BlobDescriptor.is_blob_descriptor(raw): + return BlobDescriptor.deserialize(raw) + return None + @property def pending_row_count(self) -> int: # Overrides the base property, which reads a buffer this writer never @@ -555,6 +618,8 @@ def _close_current_writers(self): self._pending_normal_meta = None self.record_count = 0 + self._current_shared_blob_group = None + self._pending_shared_group_roll = False if normal_meta is not None or blob_metas or vector_metas: normal_name = normal_meta.file_name if normal_meta is not None else ''