This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new e2e7b402a1 [format][python] Refactor video storage internals (#9470)
e2e7b402a1 is described below
commit e2e7b402a1f319a20469647bb89848e2eb3584bb
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Aug 30 12:22:14 2026 +0800
[format][python] Refactor video storage internals (#9470)
---
docs/docs/concepts/spec/fileformat.md | 16 ++
.../main/java/org/apache/paimon/CoreOptions.java | 10 +-
.../src/main/java/org/apache/paimon/data/Blob.java | 16 +-
.../org/apache/paimon/data/BlobDescriptor.java | 8 +
.../apache/paimon/data/VideoFrameDescriptor.java | 19 ++
.../paimon/data/VideoFrameDescriptorTest.java | 36 ++++
.../paimon/data/video-frame-descriptor-v1.hex | 18 ++
.../append/DedicatedFormatRollingFileWriter.java | 23 +--
.../paimon/append/MultipleBlobFileWriter.java | 53 ++----
.../paimon/append/VideoRollingFileWriter.java | 128 ++-----------
.../DataEvolutionBlobCompactTask.java | 2 +-
.../apache/paimon/io/RollingFileWriterImpl.java | 24 ++-
.../apache/paimon/operation/BlobFileContext.java | 12 +-
.../org/apache/paimon/schema/SchemaValidation.java | 12 +-
.../java/org/apache/paimon/CoreOptionsTest.java | 2 +-
.../org/apache/paimon/append/BlobTableTest.java | 54 ++++++
.../paimon/format/blob/VideoFormatWriter.java | 7 +-
.../paimon/format/blob/VideoFileFormatTest.java | 49 +++++
.../org/apache/paimon/format/blob/video-v1.hex | 18 ++
.../pypaimon/common/options/core_options.py | 15 +-
paimon-python/pypaimon/multimodal/blob_read.py | 10 +-
paimon-python/pypaimon/multimodal/table.py | 15 +-
paimon-python/pypaimon/multimodal/video.py | 20 +-
paimon-python/pypaimon/ray/ray_paimon.py | 4 +-
.../pypaimon/read/reader/concat_batch_reader.py | 56 ++----
.../pypaimon/read/reader/format_blob_reader.py | 210 ++++-----------------
.../pypaimon/read/reader/video_format_reader.py | 197 +++++++++++++++++++
paimon-python/pypaimon/read/split_read.py | 4 +-
paimon-python/pypaimon/schema/schema_manager.py | 9 +-
paimon-python/pypaimon/table/row/blob.py | 59 ++++--
paimon-python/pypaimon/tests/blob_test.py | 33 ++++
.../tests/data_evolution_row_rolling_test.py | 39 ++++
.../pypaimon/tests/multimodal_video_test.py | 43 +++++
paimon-python/pypaimon/tests/video_format_test.py | 71 ++++++-
.../pypaimon/tests/write/write_buffer_test.py | 1 +
paimon-python/pypaimon/write/blob_format_writer.py | 12 +-
.../pypaimon/write/table_update_by_row_id.py | 5 +-
.../pypaimon/write/writer/blob_file_writer.py | 18 +-
paimon-python/pypaimon/write/writer/blob_writer.py | 62 +++---
.../write/writer/dedicated_format_writer.py | 117 ++++++------
paimon-python/pypaimon/write/writer/video_group.py | 39 ++++
41 files changed, 954 insertions(+), 592 deletions(-)
diff --git a/docs/docs/concepts/spec/fileformat.md
b/docs/docs/concepts/spec/fileformat.md
index f5a4c03550..f912e3620d 100644
--- a/docs/docs/concepts/spec/fileformat.md
+++ b/docs/docs/concepts/spec/fileformat.md
@@ -967,6 +967,22 @@ physical length index. For logical row `r` in a run
beginning at logical row `s`
`run_first_frame + (r - s)`. `-1` is a NULL run and `-2` is a data-evolution
placeholder run.
Non-negative runs have fixed frame stride one in version 1; a discontinuity
starts another run.
+The serialized `VideoFrameDescriptor` stored in an Arrow/data-file cell has
its own versioned
+wire layout. All numeric values are little endian:
+
+| Field | Size | Description |
+| --- | ---: | --- |
+| Version | 1 byte | Descriptor version, currently `1` |
+| Magic | 8 bytes | `0x564944454F46524D` (`VIDEOFRM`) |
+| URI length | 4 bytes | UTF-8 URI byte length |
+| URI | variable | URI of the containing `.video` file |
+| Offset | 8 bytes | Start of the complete encoded-video payload |
+| Length | 8 bytes | Encoded-video payload length |
+| Frame index | 8 bytes | Zero-based presentation-order frame ordinal |
+
+Descriptor bytes are independently versioned from the `.video` container. Java
and Python share
+canonical descriptor and container fixtures to keep both implementations
byte-compatible.
+
Readers validate footer and index bounds, positive physical lengths, full
coverage of the payload
region, equal run-index counts, positive run lengths, physical ordinals, and
non-negative first
frames. The format currently supports one scalar BLOB field per file. Physical
video reuse uses
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 fc8375d5e3..0e102e56f1 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -3599,8 +3599,14 @@ public class CoreOptions implements Serializable {
}
/** Resolve the scalar BLOB field stored as frame runs in video pack
files. */
- public Set<String> videoFrameField() {
- return parseCommaSeparatedSet(VIDEO_FRAME_FIELD);
+ public Optional<String> videoFrameField() {
+ Set<String> fields = parseCommaSeparatedSet(VIDEO_FRAME_FIELD);
+ checkArgument(
+ fields.size() <= 1,
+ "'%s' currently supports exactly one field, but found %s.",
+ VIDEO_FRAME_FIELD.key(),
+ fields);
+ return fields.stream().findFirst();
}
/**
diff --git a/paimon-common/src/main/java/org/apache/paimon/data/Blob.java
b/paimon-common/src/main/java/org/apache/paimon/data/Blob.java
index 049854962e..8f320e2369 100644
--- a/paimon-common/src/main/java/org/apache/paimon/data/Blob.java
+++ b/paimon-common/src/main/java/org/apache/paimon/data/Blob.java
@@ -97,12 +97,8 @@ public interface Blob {
return fromView(BlobViewStruct.deserialize(bytes));
}
- boolean videoFrameDescriptor =
VideoFrameDescriptor.isVideoFrameDescriptor(bytes);
- if (videoFrameDescriptor || BlobDescriptor.isBlobDescriptor(bytes) ||
!allowBlobData) {
- BlobDescriptor descriptor =
- videoFrameDescriptor
- ? VideoFrameDescriptor.deserialize(bytes)
- : BlobDescriptor.deserialize(bytes);
+ if (BlobDescriptor.isSerializedDescriptor(bytes) || !allowBlobData) {
+ BlobDescriptor descriptor = BlobDescriptor.deserialize(bytes);
UriReader reader =
uriReaderFactory != null
? uriReaderFactory.create(descriptor.uri())
@@ -131,12 +127,8 @@ public interface Blob {
return fromView(BlobViewStruct.deserialize(bytes));
}
- boolean videoFrameDescriptor =
VideoFrameDescriptor.isVideoFrameDescriptor(bytes);
- if (videoFrameDescriptor || BlobDescriptor.isBlobDescriptor(bytes) ||
!allowBlobData) {
- BlobDescriptor descriptor =
- videoFrameDescriptor
- ? VideoFrameDescriptor.deserialize(bytes)
- : BlobDescriptor.deserialize(bytes);
+ if (BlobDescriptor.isSerializedDescriptor(bytes) || !allowBlobData) {
+ BlobDescriptor descriptor = BlobDescriptor.deserialize(bytes);
UriReader reader = uriReader == null ? UriReader.fromFile(fileIO)
: uriReader;
return fromDescriptor(reader, descriptor);
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/BlobDescriptor.java
b/paimon-common/src/main/java/org/apache/paimon/data/BlobDescriptor.java
index f802a88470..30986aca54 100644
--- a/paimon-common/src/main/java/org/apache/paimon/data/BlobDescriptor.java
+++ b/paimon-common/src/main/java/org/apache/paimon/data/BlobDescriptor.java
@@ -193,6 +193,9 @@ public class BlobDescriptor implements Serializable {
}
public static boolean isBlobDescriptor(byte[] bytes) {
+ if (bytes == null) {
+ return false;
+ }
if (bytes.length < 9) {
return false;
}
@@ -205,4 +208,9 @@ public class BlobDescriptor implements Serializable {
}
return MAGIC == buffer.getLong();
}
+
+ /** Returns whether the bytes encode any descriptor type understood by
this version. */
+ public static boolean isSerializedDescriptor(byte[] bytes) {
+ return isBlobDescriptor(bytes) ||
VideoFrameDescriptor.isVideoFrameDescriptor(bytes);
+ }
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/VideoFrameDescriptor.java
b/paimon-common/src/main/java/org/apache/paimon/data/VideoFrameDescriptor.java
index 198e196298..bd4fedd732 100644
---
a/paimon-common/src/main/java/org/apache/paimon/data/VideoFrameDescriptor.java
+++
b/paimon-common/src/main/java/org/apache/paimon/data/VideoFrameDescriptor.java
@@ -20,6 +20,8 @@ package org.apache.paimon.data;
import org.apache.paimon.annotation.Public;
+import javax.annotation.Nullable;
+
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
@@ -55,6 +57,23 @@ public class VideoFrameDescriptor extends BlobDescriptor {
return new BlobDescriptor(uri(), offset(), length());
}
+ /** Returns the video frame carried by an exact lazy blob reference, or
{@code null}. */
+ public static @Nullable VideoFrameDescriptor fromBlob(@Nullable Blob blob)
{
+ if (blob == null || blob.getClass() != BlobRef.class) {
+ return null;
+ }
+ BlobDescriptor descriptor = blob.toDescriptor();
+ return descriptor instanceof VideoFrameDescriptor
+ ? (VideoFrameDescriptor) descriptor
+ : null;
+ }
+
+ /** Returns the physical video identity carried by a frame blob, or {@code
null}. */
+ public static @Nullable BlobDescriptor payloadDescriptor(@Nullable Blob
blob) {
+ VideoFrameDescriptor frame = fromBlob(blob);
+ return frame == null ? null : frame.payloadDescriptor();
+ }
+
@Override
public byte[] serialize() {
byte[] uriBytes = uri().getBytes(StandardCharsets.UTF_8);
diff --git
a/paimon-common/src/test/java/org/apache/paimon/data/VideoFrameDescriptorTest.java
b/paimon-common/src/test/java/org/apache/paimon/data/VideoFrameDescriptorTest.java
index 938987fad0..8ab737510b 100644
---
a/paimon-common/src/test/java/org/apache/paimon/data/VideoFrameDescriptorTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/data/VideoFrameDescriptorTest.java
@@ -18,8 +18,11 @@
package org.apache.paimon.data;
+import org.apache.paimon.utils.IOUtils;
+
import org.junit.jupiter.api.Test;
+import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import static org.assertj.core.api.Assertions.assertThat;
@@ -46,6 +49,26 @@ public class VideoFrameDescriptorTest {
assertThat(next.payloadDescriptor()).isEqualTo(frame.payloadDescriptor());
}
+ @Test
+ public void testCrossLanguageWireFixture() throws Exception {
+ VideoFrameDescriptor expected = new
VideoFrameDescriptor("s3://bucket/视频.mp4", 7, 99, 42);
+ byte[] fixture =
+ fromHex(
+ new String(
+ IOUtils.readFully(
+ VideoFrameDescriptorTest.class
+ .getClassLoader()
+ .getResourceAsStream(
+
"org/apache/paimon/data/video-frame-descriptor-v1.hex"),
+ true),
+ StandardCharsets.UTF_8)
+ .trim());
+
+ assertThat(expected.serialize()).isEqualTo(fixture);
+ assertThat(BlobDescriptor.deserialize(fixture)).isEqualTo(expected);
+ assertThat(BlobDescriptor.isSerializedDescriptor(fixture)).isTrue();
+ }
+
@Test
public void testBlobFromBytesPreservesFrameDescriptor() {
VideoFrameDescriptor expected = new
VideoFrameDescriptor("file:/video.mp4", 0, 9, 7);
@@ -67,4 +90,17 @@ public class VideoFrameDescriptorTest {
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("non-negative");
}
+
+ private static byte[] fromHex(String hex) {
+ hex = hex.replaceAll("(?m)^#.*$", "").replaceAll("\\s", "");
+ byte[] bytes = new byte[hex.length() / 2];
+ for (int i = 0; i < bytes.length; i++) {
+ int offset = i * 2;
+ bytes[i] =
+ (byte)
+ ((Character.digit(hex.charAt(offset), 16) << 4)
+ + Character.digit(hex.charAt(offset + 1),
16));
+ }
+ return bytes;
+ }
}
diff --git
a/paimon-common/src/test/resources/org/apache/paimon/data/video-frame-descriptor-v1.hex
b/paimon-common/src/test/resources/org/apache/paimon/data/video-frame-descriptor-v1.hex
new file mode 100644
index 0000000000..9b45e36167
--- /dev/null
+++
b/paimon-common/src/test/resources/org/apache/paimon/data/video-frame-descriptor-v1.hex
@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+014d52464f454449561600000073333a2f2f6275636b65742fe8a786e9a2912e6d7034070000000000000063000000000000002a00000000000000
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 0c9fe566cb..1154385185 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,9 +18,7 @@
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.data.VideoFrameDescriptor;
import org.apache.paimon.fileindex.FileIndexOptions;
@@ -214,13 +212,7 @@ public class DedicatedFormatRollingFileWriter
asyncFileWrite,
statsDenseStore,
blobTargetFileSize,
- context.blobConsumer(),
- context.blobInlineFields(),
- context.videoFrameFields(),
- context.writeNullOnMissingFile(),
- context.writeNullOnFetchFailure(),
- context.blobFetchMetricReporter(),
- context.copyBufferSize());
+ context);
} else {
this.blobWriterFactory = null;
}
@@ -503,10 +495,10 @@ public class DedicatedFormatRollingFileWriter
private static int videoFrameFieldIndex(
RowType writeSchema, @Nullable BlobFileContext context) {
- if (context == null || context.videoFrameFields().isEmpty()) {
+ if (context == null || context.videoFrameField() == null) {
return -1;
}
- String field = context.videoFrameFields().iterator().next();
+ String field = context.videoFrameField();
return writeSchema.containsField(field) ?
writeSchema.getFieldIndex(field) : -1;
}
@@ -514,14 +506,7 @@ public class DedicatedFormatRollingFileWriter
if (videoFrameFieldIndex < 0 || row.isNullAt(videoFrameFieldIndex)) {
return null;
}
- Blob blob = row.getBlob(videoFrameFieldIndex);
- if (blob == null || blob.getClass() != BlobRef.class) {
- return null;
- }
- BlobDescriptor descriptor = blob.toDescriptor();
- return descriptor instanceof VideoFrameDescriptor
- ? ((VideoFrameDescriptor) descriptor).payloadDescriptor()
- : null;
+ return
VideoFrameDescriptor.payloadDescriptor(row.getBlob(videoFrameFieldIndex));
}
/** 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 582b417911..2581bba886 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
@@ -18,8 +18,6 @@
package org.apache.paimon.append;
-import org.apache.paimon.data.BlobConsumer;
-import org.apache.paimon.data.BlobFetchMetricReporter;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fileindex.FileIndexOptions;
import org.apache.paimon.format.FileFormat;
@@ -33,18 +31,16 @@ import org.apache.paimon.io.RollingFileWriter;
import org.apache.paimon.io.RollingFileWriterImpl;
import org.apache.paimon.io.RowDataFileWriter;
import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.operation.BlobFileContext;
import org.apache.paimon.statistics.NoneSimpleColStatsCollector;
import org.apache.paimon.statistics.SimpleColStatsCollector;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.LongCounter;
-import javax.annotation.Nullable;
-
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-import java.util.Set;
import java.util.function.Supplier;
import static java.util.Collections.singletonList;
@@ -65,36 +61,31 @@ public class MultipleBlobFileWriter implements Closeable {
boolean asyncFileWrite,
boolean statsDenseStore,
long targetFileSize,
- @Nullable BlobConsumer blobConsumer,
- Set<String> blobInlineFields,
- Set<String> videoFrameFields,
- boolean writeNullOnMissingFile,
- boolean writeNullOnFetchFailure,
- BlobFetchMetricReporter blobFetchMetricReporter,
- int copyBufferSize) {
- RowType blobRowType = new RowType(fieldsInBlobFile(writeSchema,
blobInlineFields));
+ BlobFileContext context) {
+ RowType blobRowType =
+ new RowType(fieldsInBlobFile(writeSchema,
context.blobInlineFields()));
this.blobWriters = new ArrayList<>();
for (String blobFieldName : blobRowType.getFieldNames()) {
- boolean video = videoFrameFields.contains(blobFieldName);
+ boolean video = blobFieldName.equals(context.videoFrameField());
FileFormat blobFileFormat;
if (video) {
- if (blobConsumer != null) {
+ if (context.blobConsumer() != null) {
throw new IllegalArgumentException(
"BlobConsumer is not supported for video frame
field '"
+ blobFieldName
+ "'.");
}
- VideoFileFormat format = new VideoFileFormat(copyBufferSize);
- format.setWriteNullOnMissingFile(writeNullOnMissingFile);
- format.setWriteNullOnFetchFailure(writeNullOnFetchFailure);
- format.setBlobFetchMetricReporter(blobFetchMetricReporter);
+ VideoFileFormat format = new
VideoFileFormat(context.copyBufferSize());
+
format.setWriteNullOnMissingFile(context.writeNullOnMissingFile());
+
format.setWriteNullOnFetchFailure(context.writeNullOnFetchFailure());
+
format.setBlobFetchMetricReporter(context.blobFetchMetricReporter());
blobFileFormat = format;
} else {
- BlobFileFormat format = new BlobFileFormat(false,
copyBufferSize);
- format.setWriteConsumer(blobConsumer);
- format.setWriteNullOnMissingFile(writeNullOnMissingFile);
- format.setWriteNullOnFetchFailure(writeNullOnFetchFailure);
- format.setBlobFetchMetricReporter(blobFetchMetricReporter);
+ BlobFileFormat format = new BlobFileFormat(false,
context.copyBufferSize());
+ format.setWriteConsumer(context.blobConsumer());
+
format.setWriteNullOnMissingFile(context.writeNullOnMissingFile());
+
format.setWriteNullOnFetchFailure(context.writeNullOnFetchFailure());
+
format.setBlobFetchMetricReporter(context.blobFetchMetricReporter());
blobFileFormat = format;
}
RowType fieldType = writeSchema.project(blobFieldName);
@@ -121,7 +112,7 @@ public class MultipleBlobFileWriter implements Closeable {
singletonList(blobFieldName),
null,
null);
- RollingFileWriter<InternalRow, DataFileMeta> rollingWriter =
+ RollingFileWriterImpl<InternalRow, DataFileMeta> rollingWriter =
video
? new VideoRollingFileWriter<>(writerFactory,
targetFileSize)
: new RollingFileWriterImpl<>(
@@ -170,20 +161,14 @@ public class MultipleBlobFileWriter implements Closeable {
private static class BlobProjectedFileWriter
extends ProjectedFileWriter<
- RollingFileWriter<InternalRow, DataFileMeta>,
List<DataFileMeta>> {
+ RollingFileWriterImpl<InternalRow, DataFileMeta>,
List<DataFileMeta>> {
public BlobProjectedFileWriter(
- RollingFileWriter<InternalRow, DataFileMeta> writer, int[]
projection) {
+ RollingFileWriterImpl<InternalRow, DataFileMeta> writer, int[]
projection) {
super(writer, projection);
}
- @SuppressWarnings("unchecked")
private List<FileWriterAbortExecutor> drainAbortExecutors() {
- RollingFileWriter<InternalRow, DataFileMeta> writer = writer();
- if (writer instanceof RollingFileWriterImpl) {
- return ((RollingFileWriterImpl<InternalRow, DataFileMeta>)
writer)
- .drainAbortExecutors();
- }
- return ((VideoRollingFileWriter<DataFileMeta>)
writer).drainAbortExecutors();
+ return writer().drainAbortExecutors();
}
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/VideoRollingFileWriter.java
b/paimon-core/src/main/java/org/apache/paimon/append/VideoRollingFileWriter.java
index 3c00f34ef9..c8eb1bbf83 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/VideoRollingFileWriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/VideoRollingFileWriter.java
@@ -18,25 +18,16 @@
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.data.VideoFrameDescriptor;
import org.apache.paimon.io.BundleRecords;
-import org.apache.paimon.io.FileWriterAbortExecutor;
-import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.io.RollingFileWriterImpl;
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;
@@ -47,130 +38,47 @@ import java.util.function.Supplier;
* encoded video remain in the current file. A different payload, NULL, or
placeholder starts a new
* file.
*/
-class VideoRollingFileWriter<R> implements RollingFileWriter<InternalRow, R> {
-
- private static final Logger LOG =
LoggerFactory.getLogger(VideoRollingFileWriter.class);
+class VideoRollingFileWriter<R> extends RollingFileWriterImpl<InternalRow, R> {
- private final Supplier<? extends SingleFileWriter<InternalRow, R>>
writerFactory;
- private final long targetFileSize;
- private final List<FileWriterAbortExecutor> closedWriters = new
ArrayList<>();
- private final List<R> results = new ArrayList<>();
-
- private @Nullable SingleFileWriter<InternalRow, R> currentWriter;
private @Nullable BlobDescriptor currentVideo;
- private long recordCount;
+ private @Nullable BlobDescriptor nextVideo;
private boolean pendingRoll;
- private boolean closed;
VideoRollingFileWriter(
Supplier<? extends SingleFileWriter<InternalRow, R>> writerFactory,
long targetFileSize) {
- this.writerFactory = writerFactory;
- this.targetFileSize = targetFileSize;
- }
-
- @Override
- public void write(InternalRow row) throws IOException {
- try {
- BlobDescriptor nextVideo = payloadDescriptor(row);
- if (currentWriter != null && pendingRoll &&
!Objects.equals(currentVideo, nextVideo)) {
- closeCurrentWriter();
- }
- if (currentWriter == null) {
- currentWriter = writerFactory.get();
- }
-
- currentWriter.write(row);
- recordCount++;
- currentVideo = nextVideo;
- if (currentWriter.reachTargetSize(
- recordCount % CHECK_ROLLING_RECORD_CNT == 0,
targetFileSize)) {
- pendingRoll = true;
- }
- } catch (Throwable e) {
- LOG.warn(
- "Exception occurs when writing video file {}. Cleaning
up.",
- currentWriter == null ? null : currentWriter.path(),
- e);
- abort();
- throw e;
- }
+ super(writerFactory, targetFileSize, Long.MAX_VALUE);
}
@Override
- public void writeBundle(BundleRecords records) throws IOException {
- for (InternalRow row : records) {
- write(row);
+ protected void beforeWrite(InternalRow row) throws IOException {
+ nextVideo = row.isNullAt(0) ? null :
VideoFrameDescriptor.payloadDescriptor(row.getBlob(0));
+ if (hasCurrentWriter() && pendingRoll && !Objects.equals(currentVideo,
nextVideo)) {
+ closeCurrentWriter();
}
}
@Override
- public long recordCount() {
- return recordCount;
- }
-
- @Override
- public void abort() {
- if (currentWriter != null) {
- currentWriter.abort();
- currentWriter = null;
- }
- for (FileWriterAbortExecutor abortExecutor : closedWriters) {
- abortExecutor.abort();
- }
+ protected void afterWrite(InternalRow row) {
+ currentVideo = nextVideo;
+ nextVideo = null;
}
@Override
- public List<R> result() {
- Preconditions.checkState(closed, "Cannot access the results unless
close all writers.");
- return results;
- }
-
- List<FileWriterAbortExecutor> drainAbortExecutors() {
- Preconditions.checkState(closed, "Cannot drain abort executors unless
close all writers.");
- List<FileWriterAbortExecutor> result = new ArrayList<>(closedWriters);
- closedWriters.clear();
- return result;
+ protected void onRollingCondition(InternalRow row) {
+ pendingRoll = true;
}
@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;
+ protected void onCurrentWriterClosed() {
currentVideo = null;
pendingRoll = false;
}
- private static @Nullable BlobDescriptor payloadDescriptor(InternalRow row)
{
- if (row.isNullAt(0)) {
- return null;
- }
- Blob blob = row.getBlob(0);
- if (blob == null || blob.getClass() != BlobRef.class) {
- return null;
+ @Override
+ public void writeBundle(BundleRecords records) throws IOException {
+ for (InternalRow row : records) {
+ write(row);
}
- BlobDescriptor descriptor = blob.toDescriptor();
- return descriptor instanceof VideoFrameDescriptor
- ? ((VideoFrameDescriptor) descriptor).payloadDescriptor()
- : 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 9dbedabe34..7a7153c967 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
@@ -130,7 +130,7 @@ public class DataEvolutionBlobCompactTask extends
DataEvolutionCompactTask {
RowType blobWriteType,
String blobFieldName,
DataFilePathFactory pathFactory) {
- boolean video = options.videoFrameField().contains(blobFieldName);
+ boolean video =
options.videoFrameField().map(blobFieldName::equals).orElse(false);
FileFormat blobFileFormat =
video
? new VideoFileFormat(options.blobCopyBufferSize())
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java
b/paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java
index fe35d51fdd..11c332b90f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/RollingFileWriterImpl.java
@@ -79,6 +79,7 @@ public class RollingFileWriterImpl<T, R> implements
RollingFileWriter<T, R> {
@Override
public void write(T row) throws IOException {
try {
+ beforeWrite(row);
// Open the current writer if write the first record or roll over
happen before.
if (currentWriter == null) {
openCurrentWriter();
@@ -87,9 +88,10 @@ public class RollingFileWriterImpl<T, R> implements
RollingFileWriter<T, R> {
currentWriter.write(row);
recordCount += 1;
currentFileRecordCount += 1;
+ afterWrite(row);
if (rollingFile(false)) {
- closeCurrentWriter();
+ onRollingCondition(row);
}
} catch (Throwable e) {
LOG.warn(
@@ -133,6 +135,22 @@ public class RollingFileWriterImpl<T, R> implements
RollingFileWriter<T, R> {
currentWriter = writerFactory.get();
}
+ /** Hook for rolling policies which need to close at a record boundary
before writing. */
+ protected void beforeWrite(T row) throws IOException {}
+
+ /** Hook for rolling policies which track the record most recently
written. */
+ protected void afterWrite(T row) {}
+
+ /** Handles a reached size or row target. The default policy rolls
immediately. */
+ protected void onRollingCondition(T row) throws IOException {
+ closeCurrentWriter();
+ }
+
+ /** Returns whether this rolling writer currently owns an open file
writer. */
+ protected boolean hasCurrentWriter() {
+ return currentWriter != null;
+ }
+
protected void closeCurrentWriter() throws IOException {
if (currentWriter == null) {
return;
@@ -146,8 +164,12 @@ public class RollingFileWriterImpl<T, R> implements
RollingFileWriter<T, R> {
results.add(currentWriter.result());
currentWriter = null;
currentFileRecordCount = 0;
+ onCurrentWriterClosed();
}
+ /** Hook for rolling policies to reset state tied to the file which was
just closed. */
+ protected void onCurrentWriterClosed() {}
+
@Override
public long recordCount() {
return recordCount;
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 b0a443e3e6..6d2e0da961 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,7 +34,7 @@ public class BlobFileContext {
private final Set<String> blobDescriptorFields;
private final Set<String> blobInlineFields;
- private final Set<String> videoFrameFields;
+ private final @Nullable String videoFrameField;
private final boolean writeNullOnMissingFile;
private final boolean writeNullOnFetchFailure;
private final int copyBufferSize;
@@ -45,13 +45,13 @@ public class BlobFileContext {
private BlobFileContext(
Set<String> blobDescriptorFields,
Set<String> blobInlineFields,
- Set<String> videoFrameFields,
+ @Nullable String videoFrameField,
boolean writeNullOnMissingFile,
boolean writeNullOnFetchFailure,
int copyBufferSize) {
this.blobDescriptorFields = blobDescriptorFields;
this.blobInlineFields = blobInlineFields;
- this.videoFrameFields = videoFrameFields;
+ this.videoFrameField = videoFrameField;
this.writeNullOnMissingFile = writeNullOnMissingFile;
this.writeNullOnFetchFailure = writeNullOnFetchFailure;
this.copyBufferSize = copyBufferSize;
@@ -77,7 +77,7 @@ public class BlobFileContext {
return new BlobFileContext(
descriptorFields,
inlineFields,
- options.videoFrameField(),
+ options.videoFrameField().orElse(null),
options.blobWriteNullOnMissingFile(),
options.blobWriteNullOnFetchFailure(),
options.blobCopyBufferSize());
@@ -109,8 +109,8 @@ public class BlobFileContext {
return blobInlineFields;
}
- public Set<String> videoFrameFields() {
- return videoFrameFields;
+ public @Nullable String videoFrameField() {
+ return videoFrameField;
}
@Nullable
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 5871482bbc..136b5dc4bb 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
@@ -1684,13 +1684,9 @@ public class SchemaValidation {
CoreOptions options,
Set<String> blobDescriptorFields,
Set<String> blobViewFields) {
- Set<String> configured = options.videoFrameField();
- checkArgument(
- configured.size() <= 1,
- "'%s' currently supports exactly one field, but found %s.",
- CoreOptions.VIDEO_FRAME_FIELD.key(),
- configured);
- for (String field : configured) {
+ Optional<String> configured = options.videoFrameField();
+ if (configured.isPresent()) {
+ String field = configured.get();
checkArgument(
rowType.containsField(field)
&&
rowType.getTypeAt(rowType.getFieldIndex(field)).getTypeRoot()
@@ -1712,7 +1708,7 @@ public class SchemaValidation {
CoreOptions.BLOB_VIEW_FIELD.key());
}
checkArgument(
- configured.isEmpty() || schema.primaryKeys().isEmpty(),
+ !configured.isPresent() || schema.primaryKeys().isEmpty(),
"'%s' only supports append-only tables.",
CoreOptions.VIDEO_FRAME_FIELD.key());
}
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 32ab4155e7..a2a2fe5c76 100644
--- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java
@@ -277,7 +277,7 @@ public class CoreOptionsTest {
options.set(CoreOptions.VIDEO_FRAME_FIELD, "video");
assertThat(CoreOptions.blobField(options.toMap())).containsExactly("image",
"video");
- assertThat(new
CoreOptions(options).videoFrameField()).containsExactly("video");
+ assertThat(new
CoreOptions(options).videoFrameField()).contains("video");
}
@Test
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 b910de8369..7fadf5942a 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
@@ -1557,6 +1557,60 @@ public class BlobTableTest extends TableTestBase {
assertVideoRows(table, firstBytes, secondBytes);
}
+ @Test
+ public void testVideoRollingByBlobTargetSize() 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 b");
+ schemaBuilder.option(CoreOptions.TARGET_FILE_ROW_NUM.key(), "1000");
+ schemaBuilder.option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ schemaBuilder.option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
+ schemaBuilder.option(CoreOptions.VIDEO_FRAME_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("size-first-source.mp4");
+ java.nio.file.Path secondSource =
tempPath.resolve("size-second-source.mp4");
+ java.nio.file.Files.write(firstSource, firstBytes);
+ java.nio.file.Files.write(secondSource, secondBytes);
+ UriReader sourceReader = UriReader.fromFile(LocalFileIO.create());
+ String firstUri = new Path(firstSource.toUri()).toString();
+ String secondUri = new Path(secondSource.toUri()).toString();
+
+ List<InternalRow> rows = new ArrayList<>();
+ for (int frame = 0; frame < 3; frame++) {
+ rows.add(
+ GenericRow.of(
+ frame,
+ Blob.fromDescriptor(
+ sourceReader,
+ new VideoFrameDescriptor(
+ firstUri, 0, firstBytes.length,
frame))));
+ }
+ for (int frame = 0; frame < 2; frame++) {
+ rows.add(
+ GenericRow.of(
+ frame + 3,
+ Blob.fromDescriptor(
+ sourceReader,
+ new VideoFrameDescriptor(
+ secondUri, 0, secondBytes.length,
frame))));
+ }
+ writeRows(getTableDefault(), rows);
+
+ List<DataFileMeta> videoFiles = liveVideoFiles(getTableDefault());
+ assertThat(videoFiles.size()).isEqualTo(2);
+ assertThat(
+ videoFiles.stream()
+ .map(DataFileMeta::rowCount)
+ .sorted()
+ .collect(Collectors.toList()))
+ .isEqualTo(Arrays.asList(2L, 3L));
+ }
+
private List<DataFileMeta> liveVideoFiles(FileStoreTable table) {
return table.store().newScan().plan().files().stream()
.map(ManifestEntry::file)
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/blob/VideoFormatWriter.java
b/paimon-format/src/main/java/org/apache/paimon/format/blob/VideoFormatWriter.java
index 1e3c747f0a..4bc3602b69 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/blob/VideoFormatWriter.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/blob/VideoFormatWriter.java
@@ -22,7 +22,6 @@ 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.data.VideoFrameDescriptor;
import org.apache.paimon.format.FileAwareFormatWriter;
@@ -116,13 +115,11 @@ public class VideoFormatWriter implements
FileAwareFormatWriter {
append(PLACEHOLDER_REFERENCE, 0);
return;
}
+ VideoFrameDescriptor frame = VideoFrameDescriptor.fromBlob(blob);
checkArgument(
- blob != null
- && blob.getClass() == BlobRef.class
- && blob.toDescriptor() instanceof VideoFrameDescriptor,
+ frame != null,
"Video fields require an exact BlobRef containing a
VideoFrameDescriptor.");
- VideoFrameDescriptor frame = (VideoFrameDescriptor)
blob.toDescriptor();
BlobDescriptor payload = frame.payloadDescriptor();
Integer ordinal = physicalVideos.get(payload);
if (ordinal == null) {
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/blob/VideoFileFormatTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/blob/VideoFileFormatTest.java
index 38b48ecd44..b2ece4a8b2 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/blob/VideoFileFormatTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/blob/VideoFileFormatTest.java
@@ -40,6 +40,7 @@ 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.IOUtils;
import org.apache.paimon.utils.RoaringBitmap32;
import org.junit.jupiter.api.BeforeEach;
@@ -47,6 +48,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.List;
@@ -118,6 +120,40 @@ public class VideoFileFormatTest {
assertThat(rows.get(5).getBlob(0)).isSameAs(BlobPlaceholder.INSTANCE);
}
+ @Test
+ public void testCrossLanguageV1Fixture() throws IOException {
+ byte[] fixture =
+ fromHex(
+ new String(
+ IOUtils.readFully(
+ VideoFileFormatTest.class
+ .getClassLoader()
+ .getResourceAsStream(
+
"org/apache/paimon/format/blob/video-v1.hex"),
+ true),
+ StandardCharsets.UTF_8)
+ .trim());
+
+ Blob a2 = sourceFrame("a.mp4", "abc".getBytes(StandardCharsets.UTF_8),
2);
+ Blob a3 = sourceFrame("a.mp4", "abc".getBytes(StandardCharsets.UTF_8),
3);
+ Blob b7 = sourceFrame("b.mp4",
"WXYZ".getBytes(StandardCharsets.UTF_8), 7);
+ Blob b8 = sourceFrame("b.mp4",
"WXYZ".getBytes(StandardCharsets.UTF_8), 8);
+ Blob a10 = sourceFrame("a.mp4",
"abc".getBytes(StandardCharsets.UTF_8), 10);
+ write(a2, a3, null, BlobPlaceholder.INSTANCE, b7, b8, a10);
+
+
assertThat(Files.readAllBytes(java.nio.file.Paths.get(file.toUri()))).isEqualTo(fixture);
+ try (SeekableInputStream in = fileIO.newInputStream(file)) {
+ VideoFileMeta meta = new VideoFileMeta(in, fixture.length, null);
+ assertThat(meta.recordNumber()).isEqualTo(7);
+ assertThat(meta.physicalVideoNumber()).isEqualTo(2);
+ assertThat(meta.frameIndex(0)).isEqualTo(2);
+ assertThat(meta.frameIndex(1)).isEqualTo(3);
+ assertThat(meta.frameIndex(4)).isEqualTo(7);
+ assertThat(meta.frameIndex(5)).isEqualTo(8);
+ assertThat(meta.frameIndex(6)).isEqualTo(10);
+ }
+ }
+
@Test
public void testSelectionKeepsLogicalRowPositions() throws IOException {
byte[] bytes = "first-mp4".getBytes();
@@ -269,4 +305,17 @@ public class VideoFileFormatTest {
private static int putInt(byte[] target, int position, int value) {
return put(target, position, intToLittleEndian(value));
}
+
+ private static byte[] fromHex(String hex) {
+ hex = hex.replaceAll("(?m)^#.*$", "").replaceAll("\\s", "");
+ byte[] bytes = new byte[hex.length() / 2];
+ for (int i = 0; i < bytes.length; i++) {
+ int offset = i * 2;
+ bytes[i] =
+ (byte)
+ ((Character.digit(hex.charAt(offset), 16) << 4)
+ + Character.digit(hex.charAt(offset + 1),
16));
+ }
+ return bytes;
+ }
}
diff --git
a/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-v1.hex
b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-v1.hex
new file mode 100644
index 0000000000..1d9b2533aa
--- /dev/null
+++
b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-v1.hex
@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+6162635758595a0602040100020100010106010403000e06020000000500000005000000050000004944454f01
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 2f063b7389..2c4b96d867 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -1247,9 +1247,20 @@ class CoreOptions:
value = self.options.get(CoreOptions.BLOB_FIELD, default)
return CoreOptions._parse_field_set(value)
- def video_frame_fields(self, default=None):
+ def video_frame_field(self, default=None) -> Optional[str]:
value = self.options.get(CoreOptions.VIDEO_FRAME_FIELD, default)
- return CoreOptions._parse_field_set(value)
+ fields = CoreOptions._parse_field_set(value)
+ if len(fields) > 1:
+ raise ValueError(
+ "'video-frame-field' currently supports exactly one field, "
+ f"but found {sorted(fields)}."
+ )
+ return next(iter(fields)) if fields else None
+
+ def video_frame_fields(self, default=None):
+ """Compatibility accessor; internal code should use
``video_frame_field``."""
+ field = self.video_frame_field(default)
+ return {field} if field is not None else set()
def blob_view_resolve_enabled(self, default=True):
return self.options.get(CoreOptions.BLOB_VIEW_RESOLVE_ENABLED, default)
diff --git a/paimon-python/pypaimon/multimodal/blob_read.py
b/paimon-python/pypaimon/multimodal/blob_read.py
index 675eeeb08c..e15a3b9b94 100644
--- a/paimon-python/pypaimon/multimodal/blob_read.py
+++ b/paimon-python/pypaimon/multimodal/blob_read.py
@@ -28,9 +28,8 @@ def fetch_blob_bodies(
row and MAP entry order and are grouped per column.
"""
from pypaimon.table.row.blob import (
- BlobDescriptor,
+ BlobDescriptorSerde,
BlobViewStruct,
- VideoFrameDescriptor,
)
ranges = []
@@ -50,11 +49,8 @@ def fetch_blob_bodies(
raise ValueError(
"read_blobs does not support unresolved blob-view columns;
"
"read such a column on its own, or enable blob-view
resolution.")
- if (
- VideoFrameDescriptor.is_video_frame_descriptor(raw)
- or BlobDescriptor.is_blob_descriptor(raw)
- ):
- descriptor = BlobDescriptor.deserialize(raw)
+ if BlobDescriptorSerde.is_descriptor(raw):
+ descriptor = BlobDescriptorSerde.deserialize(raw)
ranges.append(
(descriptor.uri, descriptor.offset, descriptor.length)
)
diff --git a/paimon-python/pypaimon/multimodal/table.py
b/paimon-python/pypaimon/multimodal/table.py
index 70cb3dd731..b0256249e0 100644
--- a/paimon-python/pypaimon/multimodal/table.py
+++ b/paimon-python/pypaimon/multimodal/table.py
@@ -171,13 +171,13 @@ class MultimodalTable:
return self.add_batches(frame_batches())
def _resolve_video_frame_column(self, requested):
- configured = self.raw_table.options.video_frame_fields()
- if not configured:
+ configured = self.raw_table.options.video_frame_field()
+ if configured is None:
raise ValueError(
"add_video requires table option 'video-frame-field'."
)
- column = requested or next(iter(configured))
- if column not in configured:
+ column = requested or configured
+ if column != configured:
raise ValueError(
"Video column %r is not configured by 'video-frame-field'."
% column
@@ -254,14 +254,13 @@ class MultimodalTable:
return self
def update(self, where, values):
- video_columns = self.raw_table.options.video_frame_fields()
+ video_column = self.raw_table.options.video_frame_field()
if isinstance(values, Mapping):
- updated_video_columns = video_columns.intersection(values)
- if updated_video_columns:
+ if video_column is not None and video_column in values:
raise ValueError(
"update() cannot write video-frame-field %r; use "
"replace_video() with a complete encoded video."
- % sorted(updated_video_columns)
+ % video_column
)
query = self.scan().where(where)
predicate = query._predicate
diff --git a/paimon-python/pypaimon/multimodal/video.py
b/paimon-python/pypaimon/multimodal/video.py
index bbc1f6b2de..ebaaab74f4 100644
--- a/paimon-python/pypaimon/multimodal/video.py
+++ b/paimon-python/pypaimon/multimodal/video.py
@@ -88,9 +88,17 @@ class VideoFrameCollator:
return self._collate(decoded_rows)
def close(self):
- while self._decoders:
- _, resource = self._decoders.popitem(last=False)
- self._close_resource(resource)
+ resources = list(self._decoders.values())
+ self._decoders.clear()
+ first_error = None
+ for resource in resources:
+ try:
+ self._close_resource(resource)
+ except Exception as error:
+ if first_error is None:
+ first_error = error
+ if first_error is not None:
+ raise first_error
def __getstate__(self):
# DataLoader spawn workers must never inherit non-picklable decoder or
@@ -180,8 +188,10 @@ class VideoFrameCollator:
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
+ try:
+ self.close()
+ finally:
+ self._owner_pid = pid
@staticmethod
def _close_resource(resource):
diff --git a/paimon-python/pypaimon/ray/ray_paimon.py
b/paimon-python/pypaimon/ray/ray_paimon.py
index 26968d899d..460344fb6e 100644
--- a/paimon-python/pypaimon/ray/ray_paimon.py
+++ b/paimon-python/pypaimon/ray/ray_paimon.py
@@ -270,7 +270,7 @@ def _unknown_blob_descriptor_columns(batch, scalar_cols):
def _looks_like_blob_descriptor(column):
import pyarrow as pa
- from pypaimon.table.row.blob import BlobDescriptor
+ from pypaimon.table.row.blob import BlobDescriptorSerde
if not (pa.types.is_binary(column.type) or
pa.types.is_large_binary(column.type)):
return False
@@ -278,7 +278,7 @@ def _looks_like_blob_descriptor(column):
for chunk in chunks:
for value in chunk:
if value.is_valid:
- return BlobDescriptor.is_blob_descriptor(value.as_py())
+ return BlobDescriptorSerde.is_descriptor(value.as_py())
return False
diff --git a/paimon-python/pypaimon/read/reader/concat_batch_reader.py
b/paimon-python/pypaimon/read/reader/concat_batch_reader.py
index 843f194031..b1ddd63043 100644
--- a/paimon-python/pypaimon/read/reader/concat_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/concat_batch_reader.py
@@ -23,10 +23,6 @@ import pyarrow as pa
from pyarrow import RecordBatch
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
-from pypaimon.read.reader.format_blob_reader import (
- BlobRecordIterator,
- VideoFrameRecordIterator,
-)
from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
from pypaimon.schema.data_types import DataField, PyarrowFieldParser
from pypaimon.table.row.blob import Blob
@@ -590,46 +586,18 @@ class BlobFallbackBatchReader(RecordBatchReader):
if reader is None:
return {}
- try:
- if getattr(reader, "_is_video", False):
- iterator = VideoFrameRecordIterator(
- reader._file_io,
- reader.file_path,
- reader._video_meta,
- self._data_field,
- )
- blobs = []
- for position, _ in positions_and_row_ids:
- iterator.current_position = position
- blobs.append(next(iterator).values[0])
- else:
- blob_lengths = [
- reader.blob_lengths[pos]
- for pos, _ in positions_and_row_ids
- ]
- blob_offsets = [
- reader.blob_offsets[pos]
- for pos, _ in positions_and_row_ids
- ]
- iterator = BlobRecordIterator(
- reader._file_io,
- reader.file_path,
- blob_lengths,
- blob_offsets,
- self._data_field,
- reader._input_stream,
- blob_as_descriptor=(
- self._blob_as_descriptor
- or self._blob_parallelism > 1
- ),
- )
- blobs = [row.values[0] for row in iterator]
- return {
- row_id: blob
- for (_, row_id), blob in zip(positions_and_row_ids, blobs)
- }
- except AttributeError as e:
- raise TypeError("Blob fallback reader expects FormatBlobReader
suppliers.") from e
+ read_values_at = getattr(reader, "read_values_at", None)
+ if not callable(read_values_at):
+ raise TypeError(
+ "Blob fallback reader expects readers with read_values_at()."
+ )
+ blobs = read_values_at(
+ [position for position, _ in positions_and_row_ids]
+ )
+ return {
+ row_id: blob
+ for (_, row_id), blob in zip(positions_and_row_ids, blobs)
+ }
@staticmethod
def _selected_positions_and_row_ids(
diff --git a/paimon-python/pypaimon/read/reader/format_blob_reader.py
b/paimon-python/pypaimon/read/reader/format_blob_reader.py
index d12b0ebf87..8446bfcb84 100644
--- a/paimon-python/pypaimon/read/reader/format_blob_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_blob_reader.py
@@ -15,7 +15,6 @@
# specific language governing permissions and limitations
# under the License.
-import bisect
import struct
from typing import List, Optional, Any, Iterator, BinaryIO
@@ -26,8 +25,11 @@ from pyarrow import RecordBatch
from pypaimon.common.delta_varint_compressor import DeltaVarintCompressor
from pypaimon.common.file_io import FileIO
from pypaimon.common.map_blob_key_serializer import
create_map_blob_key_serializer
-from pypaimon.common.uri_reader import UriReader
from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
+from pypaimon.read.reader.video_format_reader import (
+ VideoFileMeta,
+ VideoFrameRecordIterator,
+)
from pypaimon.schema.data_types import (
DataField,
PyarrowFieldParser,
@@ -36,7 +38,7 @@ from pypaimon.schema.data_types import (
is_array_blob_type,
is_map_blob_type,
)
-from pypaimon.table.row.blob import Blob, VideoFrameDescriptor
+from pypaimon.table.row.blob import Blob
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.table.row.row_kind import RowKind
@@ -311,6 +313,36 @@ class FormatBlobReader(RecordBatchReader):
return self._video_meta.record_count
return len(self.blob_lengths)
+ def read_values_at(self, positions: List[int]) -> List[object]:
+ """Read logical BLOB values by position without exposing format
internals."""
+ if self._is_video:
+ iterator = VideoFrameRecordIterator(
+ self._file_io,
+ self.file_path,
+ self._video_meta,
+ self._data_field,
+ )
+ values = []
+ for position in positions:
+ iterator.current_position = position
+ values.append(next(iterator).values[0])
+ return values
+
+ blob_lengths = [self.blob_lengths[position] for position in positions]
+ blob_offsets = [self.blob_offsets[position] for position in positions]
+ iterator = BlobRecordIterator(
+ self._file_io,
+ self.file_path,
+ blob_lengths,
+ blob_offsets,
+ self._data_field,
+ self._input_stream,
+ blob_as_descriptor=(
+ self._blob_as_descriptor or self._blob_parallelism > 1
+ ),
+ )
+ return [row.values[0] for row in iterator]
+
def _read_index(self) -> None:
if self._is_video:
self._video_meta = VideoFileMeta(
@@ -379,178 +411,6 @@ class FormatBlobReader(RecordBatchReader):
self.blob_offsets = selected_offsets
-class VideoFileMeta:
- """Validated embedded index of a ``.video`` file."""
-
- VERSION = 1
- MAGIC_NUMBER = 0x4F454449
- FOOTER_SIZE = 21
- NULL_REFERENCE = -1
- PLACE_HOLDER_REFERENCE = -2
-
- def __init__(self, stream, file_size: int):
- if file_size < self.FOOTER_SIZE:
- raise IOError(
- "Corrupt video file: file is smaller than its footer."
- )
- footer_start = file_size - self.FOOTER_SIZE
- stream.seek(footer_start)
- footer = stream.read(self.FOOTER_SIZE)
- if len(footer) != self.FOOTER_SIZE:
- raise IOError("Corrupt video file: cannot read footer.")
- lengths = struct.unpack('<IIIIIB', footer)
- index_lengths = lengths[:4]
- magic, version = lengths[4:]
- if magic != self.MAGIC_NUMBER:
- raise IOError(
- "Corrupt video file: invalid footer magic %s." % magic
- )
- if version != self.VERSION:
- raise IOError("Unsupported video format version: %s" % version)
-
- total_index_length = sum(index_lengths)
- if total_index_length > footer_start:
- raise IOError("Corrupt video file: indexes exceed the file size.")
- index_start = footer_start - total_index_length
- indexes = []
- offset = index_start
- for name, length in zip(
- ("physical video", "run length", "run reference", "first
frame"),
- index_lengths):
- stream.seek(offset)
- raw = stream.read(length)
- if len(raw) != length:
- raise IOError(
- "Corrupt video file: cannot read %s index." % name
- )
- indexes.append(DeltaVarintCompressor.decompress(raw))
- offset += length
-
- physical_lengths, run_lengths, references, first_frames = indexes
- physical_offsets = []
- payload_offset = 0
- for ordinal, length in enumerate(physical_lengths):
- if length <= 0 or length > index_start - payload_offset:
- raise IOError(
- "Corrupt video file: invalid physical video length %s "
- "at ordinal %s." % (length, ordinal)
- )
- physical_offsets.append(payload_offset)
- payload_offset += length
- if payload_offset != index_start:
- raise IOError(
- "Corrupt video file: indexed videos use %s bytes, but payload "
- "region contains %s bytes." % (payload_offset, index_start)
- )
-
- if not (len(run_lengths) == len(references) == len(first_frames)):
- raise IOError(
- "Corrupt video file: run indexes have different counts."
- )
- run_ends = []
- row_count = 0
- for run, (length, reference, first_frame) in enumerate(zip(
- run_lengths, references, first_frames)):
- if length <= 0:
- raise IOError(
- "Corrupt video file: invalid run length %s at run %s."
- % (length, run)
- )
- if (
- reference not in (
- self.NULL_REFERENCE, self.PLACE_HOLDER_REFERENCE
- )
- and (reference < 0 or reference >= len(physical_lengths))
- ):
- raise IOError(
- "Corrupt video file: run %s references physical video %s, "
- "but physical video count is %s."
- % (run, reference, len(physical_lengths))
- )
- if reference >= 0 and first_frame < 0:
- raise IOError(
- "Corrupt video file: run %s has negative first frame %s."
- % (run, first_frame)
- )
- row_count += length
- run_ends.append(row_count)
-
- self.physical_lengths = physical_lengths
- self.physical_offsets = physical_offsets
- self.run_ends = run_ends
- self.references = references
- self.first_frames = first_frames
- self.row_count = row_count
- self.selected_positions = None
-
- @property
- def record_count(self) -> int:
- return (
- self.row_count
- if self.selected_positions is None
- else len(self.selected_positions)
- )
-
- def select(self, row_indices) -> None:
- selected = []
- for value in row_indices:
- position = int(value)
- if position < 0 or position >= self.row_count:
- raise IndexError(
- "Video row index %s is out of range, record count: %s."
- % (position, self.row_count)
- )
- selected.append(position)
- self.selected_positions = selected
-
- def logical_position(self, returned_row: int) -> int:
- if self.selected_positions is None:
- return returned_row
- return self.selected_positions[returned_row]
-
- def frame(self, returned_row: int):
- logical = self.logical_position(returned_row)
- run = bisect.bisect_left(self.run_ends, logical + 1)
- reference = self.references[run]
- if reference == self.NULL_REFERENCE:
- return None
- if reference == self.PLACE_HOLDER_REFERENCE:
- return Blob.PLACE_HOLDER
- run_start = 0 if run == 0 else self.run_ends[run - 1]
- return (
- self.physical_offsets[reference],
- self.physical_lengths[reference],
- self.first_frames[run] + logical - run_start,
- )
-
-
-class VideoFrameRecordIterator:
-
- def __init__(self, file_io, file_path, meta, field):
- self.file_io = file_io
- self.file_path = file_path
- self.meta = meta
- self.field = field
- self.current_position = 0
- self._uri_reader = UriReader.from_file(file_io)
-
- def __iter__(self):
- return self
-
- def __next__(self):
- if self.current_position >= self.meta.record_count:
- raise StopIteration
- value = self.meta.frame(self.current_position)
- if isinstance(value, tuple):
- offset, length, frame_index = value
- descriptor = VideoFrameDescriptor(
- self.file_path, offset, length, frame_index
- )
- value = Blob.from_descriptor(self._uri_reader, descriptor)
- self.current_position += 1
- return GenericRow([value], [self.field], RowKind.INSERT)
-
-
class BlobRecordIterator:
MAGIC_NUMBER_SIZE = 4
METADATA_OVERHEAD = 16
diff --git a/paimon-python/pypaimon/read/reader/video_format_reader.py
b/paimon-python/pypaimon/read/reader/video_format_reader.py
new file mode 100644
index 0000000000..3c2b91efdf
--- /dev/null
+++ b/paimon-python/pypaimon/read/reader/video_format_reader.py
@@ -0,0 +1,197 @@
+# 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 bisect
+import struct
+
+from pypaimon.common.delta_varint_compressor import DeltaVarintCompressor
+from pypaimon.common.uri_reader import UriReader
+from pypaimon.table.row.blob import Blob, VideoFrameDescriptor
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.row.row_kind import RowKind
+
+
+class VideoFileMeta:
+ """Validated embedded index of a ``.video`` file."""
+
+ VERSION = 1
+ MAGIC_NUMBER = 0x4F454449
+ FOOTER_SIZE = 21
+ NULL_REFERENCE = -1
+ PLACE_HOLDER_REFERENCE = -2
+
+ def __init__(self, stream, file_size: int):
+ if file_size < self.FOOTER_SIZE:
+ raise IOError(
+ "Corrupt video file: file is smaller than its footer."
+ )
+ footer_start = file_size - self.FOOTER_SIZE
+ stream.seek(footer_start)
+ footer = stream.read(self.FOOTER_SIZE)
+ if len(footer) != self.FOOTER_SIZE:
+ raise IOError("Corrupt video file: cannot read footer.")
+ lengths = struct.unpack('<IIIIIB', footer)
+ index_lengths = lengths[:4]
+ magic, version = lengths[4:]
+ if magic != self.MAGIC_NUMBER:
+ raise IOError(
+ "Corrupt video file: invalid footer magic %s." % magic
+ )
+ if version != self.VERSION:
+ raise IOError("Unsupported video format version: %s" % version)
+
+ total_index_length = sum(index_lengths)
+ if total_index_length > footer_start:
+ raise IOError("Corrupt video file: indexes exceed the file size.")
+ index_start = footer_start - total_index_length
+ indexes = []
+ offset = index_start
+ for name, length in zip(
+ ("physical video", "run length", "run reference", "first
frame"),
+ index_lengths):
+ stream.seek(offset)
+ raw = stream.read(length)
+ if len(raw) != length:
+ raise IOError(
+ "Corrupt video file: cannot read %s index." % name
+ )
+ indexes.append(DeltaVarintCompressor.decompress(raw))
+ offset += length
+
+ physical_lengths, run_lengths, references, first_frames = indexes
+ physical_offsets = []
+ payload_offset = 0
+ for ordinal, length in enumerate(physical_lengths):
+ if length <= 0 or length > index_start - payload_offset:
+ raise IOError(
+ "Corrupt video file: invalid physical video length %s "
+ "at ordinal %s." % (length, ordinal)
+ )
+ physical_offsets.append(payload_offset)
+ payload_offset += length
+ if payload_offset != index_start:
+ raise IOError(
+ "Corrupt video file: indexed videos use %s bytes, but payload "
+ "region contains %s bytes." % (payload_offset, index_start)
+ )
+
+ if not (len(run_lengths) == len(references) == len(first_frames)):
+ raise IOError(
+ "Corrupt video file: run indexes have different counts."
+ )
+ run_ends = []
+ row_count = 0
+ for run, (length, reference, first_frame) in enumerate(zip(
+ run_lengths, references, first_frames)):
+ if length <= 0:
+ raise IOError(
+ "Corrupt video file: invalid run length %s at run %s."
+ % (length, run)
+ )
+ if (
+ reference not in (
+ self.NULL_REFERENCE, self.PLACE_HOLDER_REFERENCE
+ )
+ and (reference < 0 or reference >= len(physical_lengths))
+ ):
+ raise IOError(
+ "Corrupt video file: run %s references physical video %s, "
+ "but physical video count is %s."
+ % (run, reference, len(physical_lengths))
+ )
+ if reference >= 0 and first_frame < 0:
+ raise IOError(
+ "Corrupt video file: run %s has negative first frame %s."
+ % (run, first_frame)
+ )
+ row_count += length
+ run_ends.append(row_count)
+
+ self.physical_lengths = physical_lengths
+ self.physical_offsets = physical_offsets
+ self.run_ends = run_ends
+ self.references = references
+ self.first_frames = first_frames
+ self.row_count = row_count
+ self.selected_positions = None
+
+ @property
+ def record_count(self) -> int:
+ return (
+ self.row_count
+ if self.selected_positions is None
+ else len(self.selected_positions)
+ )
+
+ def select(self, row_indices) -> None:
+ selected = []
+ for value in row_indices:
+ position = int(value)
+ if position < 0 or position >= self.row_count:
+ raise IndexError(
+ "Video row index %s is out of range, record count: %s."
+ % (position, self.row_count)
+ )
+ selected.append(position)
+ self.selected_positions = selected
+
+ def logical_position(self, returned_row: int) -> int:
+ if self.selected_positions is None:
+ return returned_row
+ return self.selected_positions[returned_row]
+
+ def frame(self, returned_row: int):
+ logical = self.logical_position(returned_row)
+ run = bisect.bisect_left(self.run_ends, logical + 1)
+ reference = self.references[run]
+ if reference == self.NULL_REFERENCE:
+ return None
+ if reference == self.PLACE_HOLDER_REFERENCE:
+ return Blob.PLACE_HOLDER
+ run_start = 0 if run == 0 else self.run_ends[run - 1]
+ return (
+ self.physical_offsets[reference],
+ self.physical_lengths[reference],
+ self.first_frames[run] + logical - run_start,
+ )
+
+
+class VideoFrameRecordIterator:
+
+ def __init__(self, file_io, file_path, meta, field):
+ self.file_io = file_io
+ self.file_path = file_path
+ self.meta = meta
+ self.field = field
+ self.current_position = 0
+ self._uri_reader = UriReader.from_file(file_io)
+
+ def __iter__(self):
+ return self
+
+ def __next__(self):
+ if self.current_position >= self.meta.record_count:
+ raise StopIteration
+ value = self.meta.frame(self.current_position)
+ if isinstance(value, tuple):
+ offset, length, frame_index = value
+ descriptor = VideoFrameDescriptor(
+ self.file_path, offset, length, frame_index
+ )
+ value = Blob.from_descriptor(self._uri_reader, descriptor)
+ self.current_position += 1
+ return GenericRow([value], [self.field], RowKind.INSERT)
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index 01cd1aaf4a..5780219898 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -485,8 +485,8 @@ class SplitRead(ABC):
def _read_blob_as_descriptor(self, field_names: List[str]) -> bool:
if CoreOptions.blob_as_descriptor(self.table.options):
return True
- video_fields = CoreOptions.video_frame_fields(self.table.options)
- if any(field_name in video_fields for field_name in field_names):
+ video_field = CoreOptions.video_frame_field(self.table.options)
+ if video_field in field_names:
return True
deferred_fields = getattr(self, '_deferred_blob_fields', set())
return any(field_name in deferred_fields for field_name in field_names)
diff --git a/paimon-python/pypaimon/schema/schema_manager.py
b/paimon-python/pypaimon/schema/schema_manager.py
index 1e6c8d8a5b..59cac643a4 100644
--- a/paimon-python/pypaimon/schema/schema_manager.py
+++ b/paimon-python/pypaimon/schema/schema_manager.py
@@ -393,13 +393,8 @@ def _validate_blob_fields(
descriptor_fields = core_options.blob_descriptor_fields()
view_fields = core_options.blob_view_fields()
- video_fields = core_options.video_frame_fields()
-
- if len(video_fields) > 1:
- raise ValueError(
- "'video-frame-field' currently supports exactly one field, but
found "
- f"{sorted(video_fields)}."
- )
+ video_field = core_options.video_frame_field()
+ video_fields = {video_field} if video_field is not None else set()
non_scalar_video_fields = video_fields.difference(scalar_blob_field_names)
if non_scalar_video_fields:
raise ValueError(
diff --git a/paimon-python/pypaimon/table/row/blob.py
b/paimon-python/pypaimon/table/row/blob.py
index 9af8f43275..4988f61c24 100644
--- a/paimon-python/pypaimon/table/row/blob.py
+++ b/paimon-python/pypaimon/table/row/blob.py
@@ -65,13 +65,12 @@ class BlobDescriptor:
@classmethod
def deserialize(cls, data: bytes) -> 'BlobDescriptor':
- video_type = globals().get('VideoFrameDescriptor')
- if (
- cls is BlobDescriptor
- and video_type is not None
- and video_type.is_video_frame_descriptor(data)
- ):
- return video_type.deserialize(data)
+ if cls is BlobDescriptor:
+ return BlobDescriptorSerde.deserialize(data)
+ return cls._deserialize(data)
+
+ @classmethod
+ def _deserialize(cls, data: bytes) -> 'BlobDescriptor':
if len(data) < 5:
raise ValueError("Invalid BlobDescriptor data: too short")
@@ -281,6 +280,23 @@ class VideoFrameDescriptor(BlobDescriptor):
)
+class BlobDescriptorSerde:
+ """Single dispatch point for persisted BlobDescriptor wire types."""
+
+ @staticmethod
+ def is_descriptor(data: bytes) -> bool:
+ return (
+ VideoFrameDescriptor.is_video_frame_descriptor(data)
+ or BlobDescriptor.is_blob_descriptor(data)
+ )
+
+ @staticmethod
+ def deserialize(data: bytes) -> BlobDescriptor:
+ if VideoFrameDescriptor.is_video_frame_descriptor(data):
+ return VideoFrameDescriptor.deserialize(data)
+ return BlobDescriptor._deserialize(data)
+
+
class BlobViewStruct:
CURRENT_VERSION = 1
MAGIC = 0x424C4F4256494557 # "BLOBVIEW"
@@ -519,18 +535,13 @@ class Blob(ABC):
data = bytes(data)
if BlobViewStruct.is_blob_view_struct(data):
return Blob.from_view(BlobViewStruct.deserialize(data))
- is_video_frame = VideoFrameDescriptor.is_video_frame_descriptor(data)
- is_descriptor = is_video_frame or
BlobDescriptor.is_blob_descriptor(data)
+ is_descriptor = BlobDescriptorSerde.is_descriptor(data)
if not allow_blob_data and not is_descriptor:
raise ValueError(
"Expected BlobDescriptor bytes, got raw bytes
(allow_blob_data=False)"
)
if is_descriptor:
- descriptor = (
- VideoFrameDescriptor.deserialize(data)
- if is_video_frame
- else BlobDescriptor.deserialize(data)
- )
+ descriptor = BlobDescriptorSerde.deserialize(data)
if uri_reader_factory is None:
if file_io is None:
raise ValueError("file_io is required to resolve
BlobDescriptor bytes")
@@ -646,6 +657,26 @@ class BlobRef(Blob):
return hash(self._descriptor)
+def video_payload_descriptor(value) -> Optional[BlobDescriptor]:
+ """Return the physical payload identity represented by a video frame
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:
+ descriptor = value.to_descriptor()
+ return (
+ descriptor.payload_descriptor
+ if isinstance(descriptor, VideoFrameDescriptor)
+ else None
+ )
+ if isinstance(value, (bytes, bytearray)):
+ raw = bytes(value)
+ if VideoFrameDescriptor.is_video_frame_descriptor(raw):
+ return VideoFrameDescriptor.deserialize(raw).payload_descriptor
+ return None
+
+
BlobConsumer = Callable[[str, Optional[BlobDescriptor]], bool]
diff --git a/paimon-python/pypaimon/tests/blob_test.py
b/paimon-python/pypaimon/tests/blob_test.py
index bbc095cdf7..544e00cdee 100644
--- a/paimon-python/pypaimon/tests/blob_test.py
+++ b/paimon-python/pypaimon/tests/blob_test.py
@@ -84,6 +84,24 @@ def _to_url(path):
return str(path) if path else path
+def _fake_blob_values(reader, positions):
+ values = []
+ for position in positions:
+ length = reader.blob_lengths[position]
+ if length == -1:
+ values.append(None)
+ elif length == -2:
+ values.append(Blob.PLACE_HOLDER)
+ else:
+ values.append(Blob.from_file(
+ reader._file_io,
+ reader.file_path,
+ reader.blob_offsets[position] + 4,
+ length - 16,
+ ))
+ return values
+
+
class RowUtilsTest(unittest.TestCase):
def test_blob_validation_only_scans_blob_fields(self):
@@ -347,6 +365,9 @@ class BlobTest(unittest.TestCase):
self._input_stream = None
self.closed = False
+ def read_values_at(self, positions):
+ return _fake_blob_values(self, positions)
+
def close(self):
self.closed = True
@@ -413,6 +434,9 @@ class BlobTest(unittest.TestCase):
self._input_stream = None
self.closed = False
+ def read_values_at(self, positions):
+ return _fake_blob_values(self, positions)
+
def close(self):
self.closed = True
@@ -497,6 +521,9 @@ class BlobTest(unittest.TestCase):
self._input_stream = None
self.closed = False
+ def read_values_at(self, positions):
+ return _fake_blob_values(self, positions)
+
def close(self):
self.closed = True
@@ -590,6 +617,9 @@ class BlobTest(unittest.TestCase):
self.blob_offsets = [0]
self._input_stream = None
+ def read_values_at(self, positions):
+ return _fake_blob_values(self, positions)
+
def close(self):
pass
@@ -648,6 +678,9 @@ class BlobTest(unittest.TestCase):
self.blob_offsets = blob_offsets
self._input_stream = None
+ def read_values_at(self, positions):
+ return _fake_blob_values(self, positions)
+
def close(self):
pass
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 5e74114460..30b10d10ff 100644
--- a/paimon-python/pypaimon/tests/data_evolution_row_rolling_test.py
+++ b/paimon-python/pypaimon/tests/data_evolution_row_rolling_test.py
@@ -20,12 +20,15 @@ import shutil
import tempfile
import unittest
import uuid
+from unittest.mock import Mock
import pyarrow as pa
from pypaimon import CatalogFactory, Schema
from pypaimon.common.uri_reader import FileUriReader
from pypaimon.table.row.blob import Blob, BlobDescriptor, VideoFrameDescriptor
+from pypaimon.write.writer.dedicated_format_writer import DedicatedFormatWriter
+from pypaimon.write.writer.video_group import VideoGroupRollingPolicy
class DataEvolutionRowRollingTest(unittest.TestCase):
@@ -250,6 +253,42 @@ class DataEvolutionRowRollingTest(unittest.TestCase):
self.assertEqual([2, 3], normal_rows)
self.assertEqual(list(range(5)), self._read_ids(table))
+ def test_video_batches_are_preserved_at_payload_boundaries(self):
+ first = BlobDescriptor("file:/first.mp4", 0, 11)
+ second = BlobDescriptor("file:/second.mp4", 0, 12)
+ data = pa.Table.from_pydict(
+ {
+ 'id': list(range(5)),
+ 'payload': [
+ VideoFrameDescriptor(
+ first.uri, first.offset, first.length, frame
+ ).serialize()
+ for frame in range(3)
+ ] + [
+ VideoFrameDescriptor(
+ second.uri, second.offset, second.length, frame
+ ).serialize()
+ for frame in range(2)
+ ],
+ },
+ schema=self.blob_schema,
+ )
+ writer = object.__new__(DedicatedFormatWriter)
+ writer.video_frame_column = 'payload'
+ writer._video_group_policy = VideoGroupRollingPolicy()
+ writer._roll_before_video_group = Mock()
+ writer._write_batch = Mock()
+ writer._write_bounded_batches = Mock()
+
+ writer._write_video_batches(data.to_batches()[0])
+
+ self.assertEqual(2, writer._write_batch.call_count)
+ self.assertEqual(
+ [3, 2],
+ [call.args[0].num_rows for call in
writer._write_batch.call_args_list],
+ )
+ writer._write_bounded_batches.assert_not_called()
+
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_video_test.py
b/paimon-python/pypaimon/tests/multimodal_video_test.py
index 438e7f2556..080cbc2a9a 100644
--- a/paimon-python/pypaimon/tests/multimodal_video_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_video_test.py
@@ -42,6 +42,13 @@ class _Decoder:
self._calls.append("close")
+class _FailingCloseDecoder(_Decoder):
+
+ def close(self):
+ super().close()
+ raise RuntimeError("decoder close failed")
+
+
class VideoFrameCollatorTest(unittest.TestCase):
def setUp(self):
@@ -136,6 +143,42 @@ class VideoFrameCollatorTest(unittest.TestCase):
with self.assertRaisesRegex(ValueError,
"VideoFrameDescriptor"):
collator([{"episode_id": 0, "video": value}])
+ def test_close_releases_all_resources_after_decoder_failure(self):
+ descriptors = [
+ self._descriptor("episode-%d.mp4" % index, bytes([index]), index)
+ for index in range(2)
+ ]
+ calls = []
+ created = []
+
+ def factory(stream):
+ decoder = (
+ _FailingCloseDecoder(stream, calls)
+ if not created
+ else _Decoder(stream, calls)
+ )
+ created.append(decoder)
+ return decoder
+
+ collator = VideoFrameCollator(
+ self.table,
+ video_column="video",
+ decoder_factory=factory,
+ decode_fn=lambda decoder, frame, row: decoder.decode(frame),
+ collate_fn=lambda rows: rows,
+ )
+ collator([
+ {"episode_id": index, "video": descriptor}
+ for index, descriptor in enumerate(descriptors)
+ ])
+
+ with self.assertRaisesRegex(RuntimeError, "decoder close failed"):
+ collator.close()
+
+ self.assertEqual(2, calls.count("close"))
+ self.assertTrue(all(decoder.closed for decoder in created))
+ self.assertEqual(0, len(collator._decoders))
+
def test_reuses_resolved_table_file_io(self):
class ResolvedFileIO:
diff --git a/paimon-python/pypaimon/tests/video_format_test.py
b/paimon-python/pypaimon/tests/video_format_test.py
index 71ed6c94ae..33d26e325d 100644
--- a/paimon-python/pypaimon/tests/video_format_test.py
+++ b/paimon-python/pypaimon/tests/video_format_test.py
@@ -23,7 +23,8 @@ 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,
VideoFileMeta
+from pypaimon.read.reader.format_blob_reader import FormatBlobReader
+from pypaimon.read.reader.video_format_reader import VideoFileMeta
from pypaimon.schema.data_types import AtomicType, DataField
from pypaimon.table.row.blob import (
Blob,
@@ -38,6 +39,18 @@ from pypaimon.write.video_format_writer import
VideoFormatWriter
class VideoFormatTest(unittest.TestCase):
+ REPOSITORY_ROOT = Path(__file__).resolve().parents[3]
+ DESCRIPTOR_FIXTURE = (
+ REPOSITORY_ROOT
+ / "paimon-common/src/test/resources/org/apache/paimon/data/"
+ "video-frame-descriptor-v1.hex"
+ )
+ VIDEO_FIXTURE = (
+ REPOSITORY_ROOT
+ / "paimon-format/src/test/resources/org/apache/paimon/format/blob/"
+ "video-v1.hex"
+ )
+
def setUp(self):
self.temp_dir = tempfile.TemporaryDirectory()
self.root = Path(self.temp_dir.name)
@@ -69,6 +82,13 @@ class VideoFormatTest(unittest.TestCase):
with self.assertRaisesRegex(ValueError, "non-negative"):
VideoFrameDescriptor("x", 0, 1, -1)
+ def test_cross_language_descriptor_fixture(self):
+ fixture = self._fixture_bytes(self.DESCRIPTOR_FIXTURE)
+ expected = VideoFrameDescriptor("s3://bucket/视频.mp4", 7, 99, 42)
+
+ self.assertEqual(fixture, expected.serialize())
+ self.assertEqual(expected, BlobDescriptor.deserialize(fixture))
+
def test_pack_raw_videos_and_map_frame_runs(self):
first_bytes = b"first-mp4"
second_bytes = b"second-mp4"
@@ -111,6 +131,42 @@ class VideoFormatTest(unittest.TestCase):
self.assertNotEqual(frames[0].payload_descriptor,
frames[2].payload_descriptor)
self.assertIsNone(values[4])
+ def test_cross_language_video_v1_fixture(self):
+ fixture = self._fixture_bytes(self.VIDEO_FIXTURE)
+ fixture_path = self.root / "fixture.video"
+ fixture_path.write_bytes(fixture)
+ target = fixture_path.as_uri()
+
+ with self.file_io.new_input_stream(target) as stream:
+ meta = VideoFileMeta(stream, len(fixture))
+ self.assertEqual(7, meta.record_count)
+ self.assertEqual((0, 3, 2), meta.frame(0))
+ self.assertEqual((0, 3, 3), meta.frame(1))
+ self.assertIsNone(meta.frame(2))
+ self.assertIs(Blob.PLACE_HOLDER, meta.frame(3))
+ self.assertEqual((3, 4, 7), meta.frame(4))
+ self.assertEqual((3, 4, 8), meta.frame(5))
+ self.assertEqual((0, 3, 10), meta.frame(6))
+
+ written_target = (self.root / "written.video").as_uri()
+ writer = VideoFormatWriter(
+ self.file_io.new_output_stream(written_target),
+ file_path=written_target,
+ )
+ values = (
+ self._source_frame("a.mp4", b"abc", 2),
+ self._source_frame("a.mp4", b"abc", 3),
+ None,
+ Blob.PLACE_HOLDER,
+ self._source_frame("b.mp4", b"WXYZ", 7),
+ self._source_frame("b.mp4", b"WXYZ", 8),
+ self._source_frame("a.mp4", b"abc", 10),
+ )
+ for value in values:
+ writer.add_element(GenericRow([value], [self.field],
RowKind.INSERT))
+ writer.close()
+ self.assertEqual(fixture, (self.root / "written.video").read_bytes())
+
def test_selection_keeps_logical_frame_positions(self):
target = (self.root / "selection.video").as_uri()
writer = VideoFormatWriter(self.file_io.new_output_stream(target))
@@ -151,6 +207,11 @@ class VideoFormatTest(unittest.TestCase):
self.assertEqual(2, reader.record_count)
self.assertEqual([], reader.blob_lengths)
self.assertEqual([], reader.blob_offsets)
+ values = reader.read_values_at([0, 1])
+ self.assertEqual(
+ [1, 3],
+ [value.to_descriptor().frame_index for value in values],
+ )
finally:
reader.close()
@@ -227,6 +288,14 @@ class VideoFormatTest(unittest.TestCase):
self.file_io.uri_reader_factory.create(descriptor.uri), descriptor
)
+ @staticmethod
+ def _fixture_bytes(path):
+ hex_value = "".join(
+ line for line in path.read_text().splitlines()
+ if not line.startswith("#")
+ )
+ return bytes.fromhex(hex_value)
+
if __name__ == '__main__':
unittest.main()
diff --git a/paimon-python/pypaimon/tests/write/write_buffer_test.py
b/paimon-python/pypaimon/tests/write/write_buffer_test.py
index 434f328897..39f49b3d64 100644
--- a/paimon-python/pypaimon/tests/write/write_buffer_test.py
+++ b/paimon-python/pypaimon/tests/write/write_buffer_test.py
@@ -589,6 +589,7 @@ class CompositeFlushResumeTest(unittest.TestCase):
self._committed_files_to_delete_on_abort = []
self.file_io = _RecordingFileIO()
self.written = []
+ self._video_group_policy = None
def _write_normal_data_to_file(self, data: pa.Table):
self.written.append(data)
diff --git a/paimon-python/pypaimon/write/blob_format_writer.py
b/paimon-python/pypaimon/write/blob_format_writer.py
index 87e0a4d2d3..ee76b8074b 100644
--- a/paimon-python/pypaimon/write/blob_format_writer.py
+++ b/paimon-python/pypaimon/write/blob_format_writer.py
@@ -36,7 +36,7 @@ from pypaimon.table.row.blob import (
BlobConsumer,
BlobData,
BlobDescriptor,
- VideoFrameDescriptor,
+ BlobDescriptorSerde,
)
from pypaimon.common.delta_varint_compressor import DeltaVarintCompressor
@@ -353,16 +353,10 @@ class BlobFormatWriter:
if isinstance(col_data, Blob):
return col_data
if isinstance(col_data, bytes):
- if VideoFrameDescriptor.is_video_frame_descriptor(col_data):
- if uri_reader_factory is None:
- raise RuntimeError("uri_reader_factory is required for
descriptor bytes.")
- descriptor = VideoFrameDescriptor.deserialize(col_data)
- uri_reader = uri_reader_factory.create(descriptor.uri)
- return Blob.from_descriptor(uri_reader, descriptor)
- if BlobDescriptor.is_blob_descriptor(col_data):
+ if BlobDescriptorSerde.is_descriptor(col_data):
if uri_reader_factory is None:
raise RuntimeError("uri_reader_factory is required for
BlobDescriptor bytes.")
- descriptor = BlobDescriptor.deserialize(col_data)
+ descriptor = BlobDescriptorSerde.deserialize(col_data)
uri_reader = uri_reader_factory.create(descriptor.uri)
return Blob.from_descriptor(uri_reader, descriptor)
return BlobData(col_data)
diff --git a/paimon-python/pypaimon/write/table_update_by_row_id.py
b/paimon-python/pypaimon/write/table_update_by_row_id.py
index 783336d1d0..e2d25dc965 100644
--- a/paimon-python/pypaimon/write/table_update_by_row_id.py
+++ b/paimon-python/pypaimon/write/table_update_by_row_id.py
@@ -793,10 +793,7 @@ class TableUpdateByRowId:
0,
column_name,
self.table.options,
- video=(
- column_name
- in self.table.options.video_frame_fields()
- ),
+ video=column_name ==
self.table.options.video_frame_field(),
)
blob_writers.append(blob_writer)
arrow_type = original_data.schema.field(column_name).type
diff --git a/paimon-python/pypaimon/write/writer/blob_file_writer.py
b/paimon-python/pypaimon/write/writer/blob_file_writer.py
index 0d214b2afe..887d8cd792 100644
--- a/paimon-python/pypaimon/write/writer/blob_file_writer.py
+++ b/paimon-python/pypaimon/write/writer/blob_file_writer.py
@@ -27,8 +27,7 @@ from pypaimon.table.row.blob import (
Blob,
BlobConsumer,
BlobData,
- BlobDescriptor,
- VideoFrameDescriptor,
+ BlobDescriptorSerde,
)
from pypaimon.schema.data_types import (
DataField,
@@ -121,16 +120,11 @@ class BlobFileWriter:
return col_data
if isinstance(col_data, bytes):
- if VideoFrameDescriptor.is_video_frame_descriptor(col_data):
- descriptor = VideoFrameDescriptor.deserialize(col_data)
+ if BlobDescriptorSerde.is_descriptor(col_data):
+ descriptor = BlobDescriptorSerde.deserialize(col_data)
uri_reader =
self.file_io.uri_reader_factory.create(descriptor.uri)
return Blob.from_descriptor(uri_reader, descriptor)
- elif BlobDescriptor.is_blob_descriptor(col_data):
- descriptor = BlobDescriptor.deserialize(col_data)
- uri_reader =
self.file_io.uri_reader_factory.create(descriptor.uri)
- return Blob.from_descriptor(uri_reader, descriptor)
- else:
- return BlobData(col_data)
+ return BlobData(col_data)
raise ValueError(
"Blob field value must be bytes/blob or serialized BlobDescriptor
bytes, "
@@ -187,9 +181,9 @@ class BlobFileWriter:
@staticmethod
def _deserialize_descriptor_or_none(raw: bytes):
- if not BlobDescriptor.is_blob_descriptor(raw):
+ if not BlobDescriptorSerde.is_descriptor(raw):
return None
- return BlobDescriptor.deserialize(raw)
+ return BlobDescriptorSerde.deserialize(raw)
def reach_target_size(self, target_size: int) -> bool:
return self.writer.reach_target_size(target_size)
diff --git a/paimon-python/pypaimon/write/writer/blob_writer.py
b/paimon-python/pypaimon/write/writer/blob_writer.py
index 2c4be59275..c32205b2c9 100644
--- a/paimon-python/pypaimon/write/writer/blob_writer.py
+++ b/paimon-python/pypaimon/write/writer/blob_writer.py
@@ -23,13 +23,12 @@ from typing import Optional, Tuple, Dict
from pypaimon.common.options.core_options import CoreOptions
from pypaimon.data.timestamp import Timestamp
from pypaimon.table.row.blob import (
- Blob,
BlobConsumer,
- BlobRef,
- VideoFrameDescriptor,
+ video_payload_descriptor,
)
from pypaimon.write.writer.append_only_data_writer import AppendOnlyDataWriter
from pypaimon.write.writer.blob_file_writer import BlobFileWriter
+from pypaimon.write.writer.video_group import VideoGroupRollingPolicy
logger = logging.getLogger(__name__)
@@ -68,8 +67,7 @@ class BlobWriter(AppendOnlyDataWriter):
self.file_uuid = str(uuid.uuid4())
self.file_count = 0
- self._current_video_group = None
- self._pending_video_roll = False
+ self._video_group_policy = VideoGroupRollingPolicy() if video else None
logger.info(f"Initialized BlobWriter with blob file format,
blob_target_file_size={self.blob_target_file_size}")
@@ -85,16 +83,18 @@ class BlobWriter(AppendOnlyDataWriter):
# in-memory serialized descriptor size.
for i in range(pending.num_rows):
row_data = pending.slice(i, 1)
- next_group = self._video_payload_descriptor(row_data.column(0)[0])
+ next_group = video_payload_descriptor(row_data.column(0)[0])
self._roll_before_video_group(next_group)
self._write_row_to_file(row_data)
self.record_count += 1
- self._current_video_group = next_group
+ if self._video_group_policy is not None:
+ self._video_group_policy.record(next_group)
if self.rolling_file():
- if self.video and next_group is not None:
- self._pending_video_roll = True
- else:
+ if (
+ self._video_group_policy is None
+ or not self._video_group_policy.defer_roll()
+ ):
self.close_current_writer()
def _write_row_to_file(self, row_data: pa.Table):
@@ -110,7 +110,7 @@ class BlobWriter(AppendOnlyDataWriter):
self.sequence_generator.next()
def write_blob(self, value, arrow_type=pa.large_binary()):
- next_group = self._video_payload_descriptor(value)
+ next_group = video_payload_descriptor(value)
self._roll_before_video_group(next_group)
if self.current_writer is None:
self.open_current_writer()
@@ -118,12 +118,14 @@ class BlobWriter(AppendOnlyDataWriter):
self.current_writer.write_blob(self.blob_column, arrow_type, value)
self.sequence_generator.next()
self.record_count += 1
- self._current_video_group = next_group
+ if self._video_group_policy is not None:
+ self._video_group_policy.record(next_group)
if self.rolling_file():
- if self.video and next_group is not None:
- self._pending_video_roll = True
- else:
+ if (
+ self._video_group_policy is None
+ or not self._video_group_policy.defer_roll()
+ ):
self.close_current_writer()
def open_current_writer(self):
@@ -166,35 +168,17 @@ class BlobWriter(AppendOnlyDataWriter):
self.current_writer = None
self.current_file_path = None
- self._current_video_group = None
- self._pending_video_roll = False
+ if self._video_group_policy is not None:
+ self._video_group_policy.reset()
def _roll_before_video_group(self, next_group):
if (
- self.video
+ self._video_group_policy is not None
and self.current_writer is not None
- and self._pending_video_roll
- and self._current_video_group != next_group
+ and self._video_group_policy.should_roll_before(next_group)
):
self.close_current_writer()
- @staticmethod
- def _video_payload_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:
- descriptor = value.to_descriptor()
- if isinstance(descriptor, VideoFrameDescriptor):
- return descriptor.payload_descriptor
- return None
- if isinstance(value, (bytes, bytearray)):
- raw = bytes(value)
- if VideoFrameDescriptor.is_video_frame_descriptor(raw):
- return VideoFrameDescriptor.deserialize(raw).payload_descriptor
- return None
-
def _write_data_to_file(self, data):
"""
Keep a fallback path for direct blob table writes while preserving the
writer uuid+counter
@@ -310,8 +294,8 @@ class BlobWriter(AppendOnlyDataWriter):
logger.warning(f"Error aborting blob writer: {e}", exc_info=e)
self.current_writer = None
self.current_file_path = None
- self._current_video_group = None
- self._pending_video_roll = False
+ if self._video_group_policy is not None:
+ self._video_group_policy.reset()
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 e6303541ee..95472cf5a0 100644
--- a/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
+++ b/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
@@ -34,8 +34,7 @@ from pypaimon.schema.data_types import (
from pypaimon.table.row.blob import (
Blob,
BlobConsumer,
- BlobRef,
- VideoFrameDescriptor,
+ video_payload_descriptor,
)
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.write.row_utils import (
@@ -44,6 +43,7 @@ from pypaimon.write.row_utils import (
row_values_to_arrow_table,
)
from pypaimon.write.writer.data_writer import DataWriter
+from pypaimon.write.writer.video_group import VideoGroupRollingPolicy
from pypaimon.write.writer.write_buffer import WriteBuffer
logger = logging.getLogger(__name__)
@@ -76,15 +76,9 @@ class DedicatedFormatWriter(DataWriter):
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.video_frame_fields = CoreOptions.video_frame_fields(self.options)
+ self.video_frame_column = CoreOptions.video_frame_field(self.options)
self.blob_inline_fields =
self.blob_descriptor_fields.union(self.blob_view_fields)
- if len(self.video_frame_fields) > 1:
- raise ValueError("'video-frame-field' currently supports exactly
one field.")
- self.video_frame_column = (
- next(iter(self.video_frame_fields)) if self.video_frame_fields
else None
- )
-
unknown_descriptor_fields = self.blob_descriptor_fields.difference(
set(self.blob_column_names)
)
@@ -148,8 +142,11 @@ class DedicatedFormatWriter(DataWriter):
# State management for blob writer
self.record_count = 0
self.closed = False
- self._current_video_group = None
- self._pending_video_group_roll = False
+ self._video_group_policy = (
+ VideoGroupRollingPolicy()
+ if self.video_frame_column is not None
+ else None
+ )
# Normal columns are buffered separately from the blob and vector
# columns, which their own writers own.
@@ -172,7 +169,7 @@ class DedicatedFormatWriter(DataWriter):
blob_column=blob_column,
options=options,
blob_consumer=blob_consumer,
- video=blob_column in self.video_frame_fields,
+ video=blob_column == self.video_frame_column,
)
# Initialize vector writer when vector.file.format is configured.
@@ -228,25 +225,10 @@ class DedicatedFormatWriter(DataWriter):
self._require_finished_flush()
try:
if self.video_frame_column is not None:
- for index in range(data.num_rows):
- row = data.slice(index, 1)
- next_group = self._video_payload_descriptor_from_batch(row)
- self._roll_before_video_group(next_group)
- self._current_video_group = next_group
- self._write_batch(row)
+ self._write_video_batches(data)
return
- offset = 0
- # _write_batch keeps normal/blob/vector pending rows in lockstep
- # and closes all writers when the common row limit is reached.
- while offset < data.num_rows:
- capacity = self.target_file_row_num - self.pending_row_count
- if capacity <= 0:
- self._close_current_writers()
- capacity = self.target_file_row_num
- length = min(capacity, data.num_rows - offset)
- self._write_batch(data.slice(offset, length))
- offset += length
+ self._write_bounded_batches(data)
except Exception as e:
logger.error("Exception occurs when writing data. Cleaning up.",
exc_info=e)
@@ -294,11 +276,11 @@ class DedicatedFormatWriter(DataWriter):
require_columns(values_by_name, required_columns, "write_row")
if self.video_frame_column is not None:
- next_group = self._video_payload_descriptor(
+ next_group = video_payload_descriptor(
values_by_name[self.video_frame_column]
)
self._roll_before_video_group(next_group)
- self._current_video_group = next_group
+ self._video_group_policy.record(next_group)
if self.normal_column_names:
normal_values = dict(values_by_name)
@@ -340,8 +322,6 @@ class DedicatedFormatWriter(DataWriter):
def _normal_row_value(self, field_name: str, value):
if field_name in self.blob_descriptor_fields and value is not None:
- from pypaimon.table.row.blob import Blob
-
if isinstance(value, Blob):
try:
return value.to_descriptor().serialize()
@@ -518,45 +498,58 @@ class DedicatedFormatWriter(DataWriter):
return self._normal_buffer.nbytes > self.target_file_size
def _roll_or_defer_for_video_group(self):
- if (
- self.video_frame_column is not None
- and self._current_video_group is not None
- ):
- self._pending_video_group_roll = True
- else:
- self._close_current_writers()
+ if self._video_group_policy is not None and
self._video_group_policy.defer_roll():
+ return
+ self._close_current_writers()
def _roll_before_video_group(self, next_group):
if (
- self._pending_video_group_roll
- and self._current_video_group != next_group
+ self._video_group_policy is not None
+ and self._video_group_policy.should_roll_before(next_group)
):
self._close_current_writers()
- def _video_payload_descriptor_from_batch(self, data: pa.RecordBatch):
+ def _write_video_batches(self, data: pa.RecordBatch):
+ for batch, group in self._video_group_runs(data):
+ self._roll_before_video_group(group)
+ self._video_group_policy.record(group)
+ if group is None:
+ self._write_bounded_batches(batch)
+ else:
+ self._write_batch(batch)
+
+ def _write_bounded_batches(self, data: pa.RecordBatch):
+ offset = 0
+ # _write_batch keeps normal/blob/vector pending rows in lockstep
+ # and closes all writers when the common row limit is reached.
+ while offset < data.num_rows:
+ capacity = self.target_file_row_num - self.pending_row_count
+ if capacity <= 0:
+ self._close_current_writers()
+ capacity = self.target_file_row_num
+ length = min(capacity, data.num_rows - offset)
+ self._write_batch(data.slice(offset, length))
+ offset += length
+
+ def _video_group_runs(self, data: pa.RecordBatch):
column_index = data.schema.get_field_index(self.video_frame_column)
if column_index < 0:
raise KeyError(
f"Column '{self.video_frame_column}' was not found in the
record batch."
)
- return self._video_payload_descriptor(data.column(column_index)[0])
+ if data.num_rows == 0:
+ return
- @staticmethod
- def _video_payload_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:
- descriptor = value.to_descriptor()
- if isinstance(descriptor, VideoFrameDescriptor):
- return descriptor.payload_descriptor
- return None
- if isinstance(value, (bytes, bytearray)):
- raw = bytes(value)
- if VideoFrameDescriptor.is_video_frame_descriptor(raw):
- return VideoFrameDescriptor.deserialize(raw).payload_descriptor
- return None
+ column = data.column(column_index)
+ start = 0
+ current_group = video_payload_descriptor(column[0])
+ for index in range(1, data.num_rows):
+ next_group = video_payload_descriptor(column[index])
+ if next_group != current_group:
+ yield data.slice(start, index - start), current_group
+ start = index
+ current_group = next_group
+ yield data.slice(start, data.num_rows - start), current_group
@property
def pending_row_count(self) -> int:
@@ -628,8 +621,8 @@ class DedicatedFormatWriter(DataWriter):
self._pending_normal_meta = None
self.record_count = 0
- self._current_video_group = None
- self._pending_video_group_roll = False
+ if self._video_group_policy is not None:
+ self._video_group_policy.reset()
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 '<none>'
diff --git a/paimon-python/pypaimon/write/writer/video_group.py
b/paimon-python/pypaimon/write/writer/video_group.py
new file mode 100644
index 0000000000..23309778ef
--- /dev/null
+++ b/paimon-python/pypaimon/write/writer/video_group.py
@@ -0,0 +1,39 @@
+# 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.
+
+class VideoGroupRollingPolicy:
+ """Tracks deferred rolling at physical-video group boundaries."""
+
+ def __init__(self):
+ self.current_group = None
+ self.pending_roll = False
+
+ def should_roll_before(self, next_group) -> bool:
+ return self.pending_roll and self.current_group != next_group
+
+ def record(self, group) -> None:
+ self.current_group = group
+
+ def defer_roll(self) -> bool:
+ if self.current_group is None:
+ return False
+ self.pending_roll = True
+ return True
+
+ def reset(self) -> None:
+ self.current_group = None
+ self.pending_roll = False
