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 88d349219b [core][python] Support adding video columns to existing
tables (#10506)
88d349219b is described below
commit 88d349219be527e9d6127f75bbf8508947b9220b
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Oct 12 10:18:41 2026 +0800
[core][python] Support adding video columns to existing tables (#10506)
---
docs/docs/pypaimon/video.md | 17 +++
.../apache/paimon/schema/ColumnDirectiveUtils.java | 29 ++--
.../paimon/table/AbstractFileStoreTable.java | 5 +-
.../org/apache/paimon/append/BlobTableTest.java | 157 +++++++++++++++++++++
.../DataEvolutionRowIdReassignerTest.java | 118 ++++++++++++++++
.../paimon/schema/ColumnDirectiveUtilsTest.java | 60 ++++++++
paimon-python/pypaimon/read/split_read.py | 6 +-
.../pypaimon/schema/column_directive_utils.py | 12 +-
paimon-python/pypaimon/table/file_store_table.py | 6 +-
.../pypaimon/tests/column_directive_utils_test.py | 36 +++++
.../pypaimon/tests/multimodal_table_test.py | 107 ++++++++++++++
.../pypaimon/tests/native_merge_into_test.py | 11 +-
.../table/time_travel_schema_evolution_test.py | 37 +++++
.../pypaimon/tests/video_column_reassign_test.py | 140 ++++++++++++++++++
paimon-python/pypaimon/write/file_store_commit.py | 2 +-
15 files changed, 723 insertions(+), 20 deletions(-)
diff --git a/docs/docs/pypaimon/video.md b/docs/docs/pypaimon/video.md
index 2c2b9961f3..01e3521e80 100644
--- a/docs/docs/pypaimon/video.md
+++ b/docs/docs/pypaimon/video.md
@@ -123,6 +123,23 @@ file. Its video grouping policy coordinates normal, BLOB,
and vector rolling
at payload boundaries. A file may exceed its target before the next boundary.
The normal `.blob` format is unchanged.
+### Add a Video Column
+
+Use `__VIDEO_FRAME_FIELD` to add and automatically register a nullable video
+column. Existing rows read NULL in the new column:
+
+```python
+from pypaimon.schema.data_types import AtomicType
+from pypaimon.schema.schema_change import SchemaChange
+
+conn.catalog.alter_table("default.images", [
+ SchemaChange.add_column("image_video", AtomicType("BYTES"),
"__VIDEO_FRAME_FIELD"),
+])
+images = conn.get_table("images")
+```
+
+Existing video fields are preserved automatically.
+
### Update frame rows and replace a video
Frame rows use the ordinary table update API for non-video columns. This writes
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/ColumnDirectiveUtils.java
b/paimon-core/src/main/java/org/apache/paimon/schema/ColumnDirectiveUtils.java
index 6d9c77d2f0..6cf4c1e7f2 100644
---
a/paimon-core/src/main/java/org/apache/paimon/schema/ColumnDirectiveUtils.java
+++
b/paimon-core/src/main/java/org/apache/paimon/schema/ColumnDirectiveUtils.java
@@ -48,6 +48,7 @@ public final class ColumnDirectiveUtils {
public static final String BLOB_FIELD_DIRECTIVE = "__BLOB_FIELD";
public static final String BLOB_DESCRIPTOR_FIELD_DIRECTIVE =
"__BLOB_DESCRIPTOR_FIELD";
public static final String BLOB_VIEW_FIELD_DIRECTIVE = "__BLOB_VIEW_FIELD";
+ public static final String VIDEO_FRAME_FIELD_DIRECTIVE =
"__VIDEO_FRAME_FIELD";
public static final String VECTOR_FIELD_DIRECTIVE = "__VECTOR_FIELD";
private ColumnDirectiveUtils() {}
@@ -56,7 +57,8 @@ public final class ColumnDirectiveUtils {
* Parses the comment of an {@code ALTER TABLE ADD COLUMN} statement.
Returns {@code null} when
* the comment is a regular user comment; returns a {@link
ParsedDirective} when the comment
* begins with a supported directive. Throws {@link
IllegalArgumentException} when the comment
- * begins with {@code __BLOB} or {@code __VECTOR} but is not one of the
supported directives.
+ * begins with {@code __BLOB}, {@code __VIDEO} or {@code __VECTOR} but is
not one of the
+ * supported directives.
*/
@Nullable
public static ParsedDirective parseAddColumnComment(@Nullable String
comment) {
@@ -67,11 +69,15 @@ public final class ColumnDirectiveUtils {
if (comment.startsWith("__VECTOR")) {
return parseVectorDirective(comment);
}
- if (!comment.startsWith("__BLOB")) {
+ if (!comment.startsWith("__BLOB") && !comment.startsWith("__VIDEO")) {
return null;
}
- String optionKey = matchDirective(comment, BLOB_VIEW_FIELD_DIRECTIVE);
- String marker = BLOB_VIEW_FIELD_DIRECTIVE;
+ String optionKey = matchDirective(comment,
VIDEO_FRAME_FIELD_DIRECTIVE);
+ String marker = VIDEO_FRAME_FIELD_DIRECTIVE;
+ if (optionKey == null) {
+ optionKey = matchDirective(comment, BLOB_VIEW_FIELD_DIRECTIVE);
+ marker = BLOB_VIEW_FIELD_DIRECTIVE;
+ }
if (optionKey == null) {
optionKey = matchDirective(comment,
BLOB_DESCRIPTOR_FIELD_DIRECTIVE);
marker = BLOB_DESCRIPTOR_FIELD_DIRECTIVE;
@@ -82,12 +88,14 @@ public final class ColumnDirectiveUtils {
}
Preconditions.checkArgument(
optionKey != null,
- "Unsupported BLOB directive in column comment: '%s'. Supported
directives are "
- + "'%s', '%s' and '%s'.",
+ "Unsupported %s directive in column comment: '%s'. Supported
directives are "
+ + "'%s', '%s', '%s' and '%s'.",
+ comment.startsWith("__VIDEO") ? "VIDEO" : "BLOB",
comment,
BLOB_FIELD_DIRECTIVE,
BLOB_DESCRIPTOR_FIELD_DIRECTIVE,
- BLOB_VIEW_FIELD_DIRECTIVE);
+ BLOB_VIEW_FIELD_DIRECTIVE,
+ VIDEO_FRAME_FIELD_DIRECTIVE);
String realComment =
comment.length() == marker.length()
? null
@@ -156,6 +164,8 @@ public final class ColumnDirectiveUtils {
return CoreOptions.BLOB_DESCRIPTOR_FIELD.key();
} else if (BLOB_VIEW_FIELD_DIRECTIVE.equals(marker)) {
return CoreOptions.BLOB_VIEW_FIELD.key();
+ } else if (VIDEO_FRAME_FIELD_DIRECTIVE.equals(marker)) {
+ return CoreOptions.VIDEO_FRAME_FIELD.key();
} else if (VECTOR_FIELD_DIRECTIVE.equals(marker)) {
return CoreOptions.VECTOR_FIELD.key();
} else {
@@ -298,6 +308,8 @@ public final class ColumnDirectiveUtils {
option = CoreOptions.BLOB_DESCRIPTOR_FIELD;
} else if (CoreOptions.BLOB_VIEW_FIELD.key().equals(optionKey)) {
option = CoreOptions.BLOB_VIEW_FIELD;
+ } else if (CoreOptions.VIDEO_FRAME_FIELD.key().equals(optionKey)) {
+ option = CoreOptions.VIDEO_FRAME_FIELD;
} else if (CoreOptions.VECTOR_FIELD.key().equals(optionKey)) {
option = CoreOptions.VECTOR_FIELD;
} else {
@@ -333,7 +345,8 @@ public final class ColumnDirectiveUtils {
new ConfigOption[] {
CoreOptions.BLOB_FIELD,
CoreOptions.BLOB_DESCRIPTOR_FIELD,
- CoreOptions.BLOB_VIEW_FIELD
+ CoreOptions.BLOB_VIEW_FIELD,
+ CoreOptions.VIDEO_FRAME_FIELD
};
private static final ConfigOption<String>[] VECTOR_OPTIONS =
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
index 5a04aeb116..cb4349a59e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
@@ -584,7 +584,7 @@ abstract class AbstractFileStoreTable implements
FileStoreTable {
/** Prevents current column declarations from overriding a historical
schema's field options. */
private static Map<String, String> excludeCurrentSchemaFieldOptions(
TableSchema historicalSchema, Options options, Set<String>
dynamicOptionKeys) {
- // Keep scan and runtime options. Only these directive-managed column
declarations
+ // Keep scan and runtime options. These column storage declarations
// must follow the historical schema, since columns may have been
added or dropped.
Map<String, String> historicalOptions = new HashMap<>(options.toMap());
for (ConfigOption<String> option :
@@ -592,7 +592,8 @@ abstract class AbstractFileStoreTable implements
FileStoreTable {
CoreOptions.VECTOR_FIELD,
CoreOptions.BLOB_FIELD,
CoreOptions.BLOB_DESCRIPTOR_FIELD,
- CoreOptions.BLOB_VIEW_FIELD)) {
+ CoreOptions.BLOB_VIEW_FIELD,
+ CoreOptions.VIDEO_FRAME_FIELD)) {
// Restore the canonical key and aliases together, or a stale
alias may take effect
// when the historical schema has no canonical value.
List<String> keys = new ArrayList<>();
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 4660b0f998..f67f3fc582 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
@@ -1529,6 +1529,163 @@ public class BlobTableTest extends TableTestBase {
assertThat(tasks2.stream().anyMatch(task -> task.type() ==
BLOB)).isFalse();
}
+ @Test
+ public void testAddVideoColumnToExistingImages() throws Exception {
+ catalog.createTable(
+ identifier(),
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("image", DataTypes.BLOB())
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .build(),
+ false);
+ FileStoreTable oldTable = getTableDefault();
+ byte[] image = "original-image".getBytes();
+ writeRows(oldTable, Collections.singletonList(GenericRow.of(0, new
BlobData(image))));
+ long oldSnapshot = oldTable.latestSnapshot().get().id();
+ catalog.alterTable(
+ identifier(),
+ SchemaChange.addColumn("video", DataTypes.BYTES(),
"__VIDEO_FRAME_FIELD"),
+ false);
+ FileStoreTable table = getTableDefault();
+ List<InternalRow> rows = read(table);
+ assertThat(rows.size()).isEqualTo(1);
+ assertThat(rows.get(0).getBlob(1).toData()).isEqualTo(image);
+ assertThat(rows.get(0).isNullAt(2)).isTrue();
+
+ // An already running writer can still append rows using the old
schema.
+ writeRows(oldTable, Collections.singletonList(GenericRow.of(1, new
BlobData(image))));
+ byte[] payload = "complete-video".getBytes();
+ java.nio.file.Path source = tempPath.resolve("evolved-video.mp4");
+ java.nio.file.Files.write(source, payload);
+ writeRows(
+ table,
+ Collections.singletonList(
+ GenericRow.of(
+ 2,
+ new BlobData(image),
+ Blob.fromDescriptor(
+
UriReader.fromFile(LocalFileIO.create()),
+ new VideoFrameDescriptor(
+ new
Path(source.toUri()).toString(),
+ 0,
+ payload.length,
+ 7,
+ -1,
+ 0)))));
+ rows = read(table,
org.apache.paimon.utils.Pair.of(CoreOptions.BLOB_AS_DESCRIPTOR, "true"));
+ rows.sort((a, b) -> Integer.compare(a.getInt(0), b.getInt(0)));
+ assertThat(rows.size()).isEqualTo(3);
+ assertThat(rows.get(0).isNullAt(2)).isTrue();
+ assertThat(rows.get(1).isNullAt(2)).isTrue();
+ assertThat(rows.get(2).getBlob(2).toData()).isEqualTo(payload);
+ assertThat(((VideoFrameDescriptor)
rows.get(2).getBlob(2).toDescriptor()).frameIndex())
+ .isEqualTo(7);
+ assertThat(liveVideoFiles(table).size()).isEqualTo(1);
+ assertThat(
+ read(table.copy(
+ Collections.singletonMap(
+ "scan.snapshot-id", "" +
oldSnapshot)))
+ .size())
+ .isEqualTo(1);
+
+ // Adding another video column automatically preserves the existing
video field.
+ catalog.alterTable(
+ identifier(),
+ SchemaChange.addColumn("second", DataTypes.BLOB(),
"__VIDEO_FRAME_FIELD"),
+ false);
+
assertThat(getTableDefault().schema().options().get("video-frame-field"))
+ .isEqualTo("video,second");
+ for (InternalRow row : read(getTableDefault())) {
+ assertThat(row.isNullAt(3)).isTrue();
+ }
+
+ catalog.alterTable(identifier(), SchemaChange.dropColumn("video"),
false);
+
assertThat(getTableDefault().schema().options().get("video-frame-field"))
+ .isEqualTo("second");
+ assertThat(read(getTableDefault()).size()).isEqualTo(3);
+ catalog.alterTable(identifier(), SchemaChange.dropColumn("second"),
false);
+
assertThat(getTableDefault().schema().options().containsKey("video-frame-field")).isFalse();
+ assertThat(read(getTableDefault()).size()).isEqualTo(3);
+ }
+
+ @Test
+ public void testRejectUnsafeVideoColumnEvolution() throws Exception {
+ catalog.createTable(
+ identifier(),
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("image", DataTypes.BLOB())
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .build(),
+ false);
+ writeRows(
+ getTableDefault(),
+ Collections.singletonList(GenericRow.of(0, new BlobData(new
byte[] {1}))));
+ for (List<SchemaChange> changes :
+ Arrays.asList(
+ Collections.singletonList(
+ SchemaChange.setOption("video-frame-field",
"image")),
+ Arrays.asList(
+ SchemaChange.addColumn("bad",
DataTypes.BLOB()),
+ SchemaChange.setOption("video-frame-field",
"bad")))) {
+ assertThatThrownBy(() -> catalog.alterTable(identifier(), changes,
false))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThat(getTableDefault().schema().id()).isZero();
+ }
+ for (org.apache.paimon.types.DataType type :
+ Arrays.asList(DataTypes.INT(), DataTypes.BLOB().notNull())) {
+ assertThatThrownBy(
+ () ->
+ catalog.alterTable(
+ identifier(),
+ SchemaChange.addColumn(
+ "bad", type,
"__VIDEO_FRAME_FIELD"),
+ false))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThat(getTableDefault().schema().id()).isZero();
+ }
+ // Editing an existing column's comment must not change its storage
mode.
+ assertThatThrownBy(
+ () ->
+ catalog.alterTable(
+ identifier(),
+ SchemaChange.updateColumnComment(
+ "image",
"__VIDEO_FRAME_FIELD"),
+ false))
+ .isInstanceOf(IllegalArgumentException.class);
+
assertThat(getTableDefault().schema().options().containsKey("video-frame-field")).isFalse();
+ catalog.alterTable(
+ identifier(),
+ SchemaChange.addColumn("video", DataTypes.BLOB(),
"__VIDEO_FRAME_FIELD"),
+ false);
+ for (String value : Arrays.asList("", "image", "video,image")) {
+ assertThatThrownBy(
+ () ->
+ catalog.alterTable(
+ identifier(),
+
SchemaChange.setOption("video-frame-field", value),
+ false))
+ .isInstanceOf(UnsupportedOperationException.class);
+ }
+ assertThatThrownBy(
+ () ->
+ catalog.alterTable(
+ identifier(),
+
SchemaChange.removeOption("video-frame-field"),
+ false))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(
+ () ->
+ getTableDefault()
+ .copy(
+ Collections.singletonMap(
+ "video-frame-field",
"video,image")))
+ .isInstanceOf(UnsupportedOperationException.class);
+ }
+
@Test
public void testVideoRollingAndCompaction() throws Exception {
Schema.Builder schemaBuilder = Schema.newBuilder();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
index 13ece4e019..5c8bda6fdf 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
@@ -25,8 +25,10 @@ import org.apache.paimon.catalog.RenamingSnapshotCommit;
import org.apache.paimon.catalog.SnapshotCommit;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Blob;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.VideoFrameDescriptor;
import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
@@ -61,6 +63,7 @@ import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.stats.SimpleStats;
import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.CatalogEnvironment;
@@ -81,8 +84,11 @@ import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.SegmentsCache;
import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.UriReader;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import javax.annotation.Nullable;
@@ -124,6 +130,8 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link DataEvolutionRowIdReassigner}. */
public class DataEvolutionRowIdReassignerTest extends TableTestBase {
+ private static final byte[] VIDEO_PAYLOAD = new byte[] {11, 22, 33, 44};
+
private static final int LARGE_ENTRY_COUNT = 10_000;
private static final int LARGE_MANIFEST_FILE_COUNT = 20;
private static final long LARGE_ENTRY_ROW_COUNT = 2L;
@@ -1309,6 +1317,116 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
.containsExactly("0|a|v0", "100|c|v100", "1|b|v1", "2|a|v2",
"3|b|v3", "4|a|v4");
}
+ @ParameterizedTest
+ @ValueSource(strings = {"a", "b"})
+ void testReassignBeforeVideoBackfillCommit(String reassignedPartition)
throws Exception {
+ FileStoreTable table = createTableAndAddVideo();
+ String originalFile = currentEntriesAtRange(table, "pt=a/", 0,
0).get(0).file().fileName();
+ List<CommitMessage> pending = prepareVideo(table, 0);
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(
+ table, partitionPredicate(table,
reassignedPartition))
+ .reassign();
+
assertThat(result.newSnapshotId).isGreaterThan(result.previousSnapshotId);
+ long afterReassign = table.latestSnapshot().get().id();
+ if (reassignedPartition.equals("a")) {
+ assertThatThrownBy(() -> commitDefault(pending))
+ .hasMessageContaining("Row ID existence conflict");
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(afterReassign);
+ assertVideoBackfillRows(table, false);
+ long newRowId =
+ currentEntries(table).stream()
+ .map(ManifestEntry::file)
+ .filter(file ->
file.fileName().equals(originalFile))
+ .findFirst()
+ .get()
+ .nonNullFirstRowId();
+ assertThat(newRowId).isGreaterThanOrEqualTo(5);
+ commitDefault(prepareVideo(table, newRowId));
+ } else {
+ commitDefault(pending);
+ }
+ assertVideoBackfillRows(table, true);
+ }
+
+ @Test
+ void testVideoBackfillCommitsAfterReassignPlanning() throws Exception {
+ FileStoreTable table = createTableAndAddVideo();
+ List<CommitMessage> pending = prepareVideo(table, 0);
+ AtomicBoolean injected = new AtomicBoolean();
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(
+ table,
+ partitionPredicate(table, "a"),
+ () -> {
+ if (injected.compareAndSet(false, true)) {
+ try {
+ commitDefault(pending);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ })
+ .reassign();
+ assertThat(injected).isTrue();
+
assertThat(result.newSnapshotId).isGreaterThan(result.previousSnapshotId);
+ assertVideoBackfillRows(table, true);
+ }
+
+ private FileStoreTable createTableAndAddVideo() throws Exception {
+ createTableWithInterleavedPartitions();
+ catalog.alterTable(
+ identifier(),
+ SchemaChange.addColumn("video", DataTypes.BYTES(),
"__VIDEO_FRAME_FIELD"),
+ false);
+ return getTableDefault();
+ }
+
+ private List<CommitMessage> prepareVideo(FileStoreTable table, long
firstRowId)
+ throws Exception {
+ long snapshot = table.latestSnapshot().get().id();
+ java.nio.file.Path source = tempPath.resolve("source.mp4");
+ java.nio.file.Files.write(source, VIDEO_PAYLOAD);
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
+ builder.newWrite()
+
.withWriteType(table.rowType().project(Arrays.asList("pt", "video")))) {
+ VideoFrameDescriptor frame =
+ new VideoFrameDescriptor(
+ new Path(source.toUri()).toString(), 0,
VIDEO_PAYLOAD.length, 7, -1, 0);
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString("a"),
+
Blob.fromDescriptor(UriReader.fromFile(LocalFileIO.create()), frame)));
+ List<CommitMessage> messages = write.prepareCommit();
+ assignFirstRowId(messages, firstRowId);
+ messages.replaceAll(
+ message -> ((CommitMessageImpl)
message).withCheckFromSnapshot(snapshot));
+ return messages;
+ }
+ }
+
+ private void assertVideoBackfillRows(FileStoreTable table, boolean
backfilled)
+ throws Exception {
+ List<InternalRow> rows = read(table,
Pair.of(CoreOptions.BLOB_AS_DESCRIPTOR, "true"));
+ rows.sort((a, b) -> Integer.compare(a.getInt(1), b.getInt(1)));
+ assertThat(rows).hasSize(5);
+ for (int id = 0; id < 5; id++) {
+ InternalRow row = rows.get(id);
+ assertThat(row.getInt(1)).isEqualTo(id);
+ assertThat(row.getString(2).toString()).isEqualTo("v" + id);
+ if (backfilled && id == 0) {
+ VideoFrameDescriptor descriptor =
+ (VideoFrameDescriptor) row.getBlob(3).toDescriptor();
+ assertThat(descriptor.frameIndex()).isEqualTo(7 + id);
+ assertThat(descriptor.uri()).endsWith(".video");
+ assertThat(row.getBlob(3).toData()).isEqualTo(VIDEO_PAYLOAD);
+ } else {
+ assertThat(row.isNullAt(3)).isTrue();
+ }
+ }
+ }
+
@Test
public void testReassignRetriesAfterConcurrentPartialColumnAppend() throws
Exception {
createTableDefault();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/ColumnDirectiveUtilsTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/ColumnDirectiveUtilsTest.java
index c8ab49525d..64f734e7fd 100644
---
a/paimon-core/src/test/java/org/apache/paimon/schema/ColumnDirectiveUtilsTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/schema/ColumnDirectiveUtilsTest.java
@@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.BlobType;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypeRoot;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.MapType;
@@ -58,6 +59,53 @@ public class ColumnDirectiveUtilsTest {
assertThat(opts).isEmpty();
}
+ @Test
+ public void testVideoFrameFieldDirective() {
+ for (DataType type :
+ new DataType[] {DataTypes.BYTES(), DataTypes.BINARY(8),
DataTypes.BLOB()}) {
+ Map<String, String> opts = new HashMap<>();
+ opts.put(CoreOptions.VIDEO_FRAME_FIELD.key(), "existing");
+ ColumnDirectiveUtils.ConvertedColumn result =
+ ColumnDirectiveUtils.applyAddColumnDirective(
+ "__VIDEO_FRAME_FIELD; camera frame", "video",
type, opts);
+ assertThat(result.type()).isEqualTo(DataTypes.BLOB());
+ assertThat(result.comment()).isEqualTo("camera frame");
+
assertThat(opts).containsEntry(CoreOptions.VIDEO_FRAME_FIELD.key(),
"existing,video");
+ result =
+ ColumnDirectiveUtils.applyAddColumnDirective(
+ "__VIDEO_FRAME_FIELD", "second", type, opts);
+ assertThat(result.comment()).isNull();
+ assertThat(opts)
+ .containsEntry(CoreOptions.VIDEO_FRAME_FIELD.key(),
"existing,video,second");
+ }
+ }
+
+ @Test
+ public void testInvalidVideoFrameFieldDirective() {
+ Map<String, String> opts = new HashMap<>();
+ for (String comment : new String[] {"__VIDEO_UNKNOWN",
"__VIDEO_FRAME_FIELD_typo"}) {
+ assertThatThrownBy(
+ () ->
+
ColumnDirectiveUtils.applyAddColumnDirective(
+ comment, "video",
DataTypes.BYTES(), opts))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThat(opts).isEmpty();
+ }
+ for (DataType type :
+ new DataType[] {
+ DataTypes.INT(),
+ DataTypes.ARRAY(DataTypes.BYTES()),
+ DataTypes.MAP(DataTypes.STRING(), DataTypes.BYTES())
+ }) {
+ assertThatThrownBy(
+ () ->
+
ColumnDirectiveUtils.applyAddColumnDirective(
+ "__VIDEO_FRAME_FIELD", "video",
type, opts))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThat(opts).isEmpty();
+ }
+ }
+
@Test
public void testBlobFieldDirective() {
Map<String, String> opts = new HashMap<>();
@@ -426,6 +474,18 @@ public class ColumnDirectiveUtilsTest {
assertThat(opts).containsEntry(CoreOptions.VECTOR_FIELD.key(), "v");
}
+ @Test
+ public void testRemoveDroppedVideoOptions() {
+ Map<String, String> opts = new HashMap<>();
+ opts.put(CoreOptions.VIDEO_FRAME_FIELD.key(), "video,second");
+ opts.put(CoreOptions.BLOB_FIELD.key(), "image");
+ ColumnDirectiveUtils.removeDroppedDirectiveOptions("video",
DataTypes.BLOB(), opts);
+ assertThat(opts).containsEntry(CoreOptions.VIDEO_FRAME_FIELD.key(),
"second");
+ ColumnDirectiveUtils.removeDroppedDirectiveOptions("second",
DataTypes.BLOB(), opts);
+
assertThat(opts).doesNotContainKey(CoreOptions.VIDEO_FRAME_FIELD.key());
+ assertThat(opts).containsEntry(CoreOptions.BLOB_FIELD.key(), "image");
+ }
+
@Test
public void testRemoveDroppedArrayBlobOptions() {
Map<String, String> opts = new HashMap<>();
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index d0e875426d..e6faeba3f1 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -1836,12 +1836,12 @@ class DataEvolutionSplitRead(SplitRead):
if not file.write_cols or len(file.write_cols) == 0:
raise ValueError("Blob/vector file must have write columns")
- # Find the field by name in the table schema
+ # Resolve the original field ID even after the column is dropped or
re-added.
field_name = file.write_cols[0]
- for field in self.table.fields:
+ for field in self._resolve_schema(file.schema_id).fields:
if field.name == field_name:
return field.id
- raise ValueError(f"Field {field_name} not found in table schema")
+ raise ValueError(f"Field {field_name} not found in file schema
{file.schema_id}")
def _get_field_ids_from_write_cols(self, write_cols: List[str]) ->
List[int]:
field_ids = []
diff --git a/paimon-python/pypaimon/schema/column_directive_utils.py
b/paimon-python/pypaimon/schema/column_directive_utils.py
index 005664b09a..4059a0ef28 100644
--- a/paimon-python/pypaimon/schema/column_directive_utils.py
+++ b/paimon-python/pypaimon/schema/column_directive_utils.py
@@ -21,6 +21,7 @@ Mirrors Java's ColumnDirectiveUtils. Supported directives:
__BLOB_FIELD -> blob-field
__BLOB_DESCRIPTOR_FIELD -> blob-descriptor-field
__BLOB_VIEW_FIELD -> blob-view-field
+ __VIDEO_FRAME_FIELD -> video-frame-field
__VECTOR_FIELD;dim -> vector-field
"""
@@ -34,9 +35,11 @@ from pypaimon.schema.data_types import (ArrayType,
AtomicType, DataField,
BLOB_FIELD_DIRECTIVE = "__BLOB_FIELD"
BLOB_DESCRIPTOR_FIELD_DIRECTIVE = "__BLOB_DESCRIPTOR_FIELD"
BLOB_VIEW_FIELD_DIRECTIVE = "__BLOB_VIEW_FIELD"
+VIDEO_FRAME_FIELD_DIRECTIVE = "__VIDEO_FRAME_FIELD"
VECTOR_FIELD_DIRECTIVE = "__VECTOR_FIELD"
_BLOB_DIRECTIVES = [
+ (VIDEO_FRAME_FIELD_DIRECTIVE, CoreOptions.VIDEO_FRAME_FIELD.key()),
(BLOB_VIEW_FIELD_DIRECTIVE, CoreOptions.BLOB_VIEW_FIELD.key()),
(BLOB_DESCRIPTOR_FIELD_DIRECTIVE, CoreOptions.BLOB_DESCRIPTOR_FIELD.key()),
(BLOB_FIELD_DIRECTIVE, CoreOptions.BLOB_FIELD.key()),
@@ -46,6 +49,7 @@ _BLOB_OPTIONS = [
CoreOptions.BLOB_FIELD,
CoreOptions.BLOB_DESCRIPTOR_FIELD,
CoreOptions.BLOB_VIEW_FIELD,
+ CoreOptions.VIDEO_FRAME_FIELD,
]
_VECTOR_OPTIONS = [
@@ -82,7 +86,7 @@ def parse_add_column_comment(comment: Optional[str]) ->
Optional[ParsedDirective
if comment.startswith("__VECTOR"):
return _parse_vector_directive(comment)
- if not comment.startswith("__BLOB"):
+ if not comment.startswith(("__BLOB", "__VIDEO")):
return None
for marker, option_key in _BLOB_DIRECTIVES:
@@ -96,10 +100,12 @@ def parse_add_column_comment(comment: Optional[str]) ->
Optional[ParsedDirective
continue
return ParsedDirective(option_key, real_comment, False, 0)
+ family = "VIDEO" if comment.startswith("__VIDEO") else "BLOB"
raise ValueError(
- f"Unsupported BLOB directive in column comment: '{comment}'. "
+ f"Unsupported {family} directive in column comment: '{comment}'. "
f"Supported directives are '{BLOB_FIELD_DIRECTIVE}', "
- f"'{BLOB_DESCRIPTOR_FIELD_DIRECTIVE}' and
'{BLOB_VIEW_FIELD_DIRECTIVE}'."
+ f"'{BLOB_DESCRIPTOR_FIELD_DIRECTIVE}', '{BLOB_VIEW_FIELD_DIRECTIVE}' "
+ f"and '{VIDEO_FRAME_FIELD_DIRECTIVE}'."
)
diff --git a/paimon-python/pypaimon/table/file_store_table.py
b/paimon-python/pypaimon/table/file_store_table.py
index 220f3b56f2..9527111d28 100644
--- a/paimon-python/pypaimon/table/file_store_table.py
+++ b/paimon-python/pypaimon/table/file_store_table.py
@@ -536,6 +536,9 @@ class FileStoreTable(Table):
return table
def _copy(self, options: dict, resolve_time_travel: bool) ->
'FileStoreTable':
+ video_key = CoreOptions.VIDEO_FRAME_FIELD.key()
+ if video_key in options and options[video_key] !=
self.table_schema.options.get(video_key):
+ raise ValueError("Cannot change immutable option " + video_key)
directory_key = CoreOptions.DATA_FILE_PATH_DIRECTORY.key()
if directory_key in options and options[directory_key] !=
self.options.data_file_path_directory():
raise ValueError('Cannot change immutable option ' + directory_key)
@@ -619,7 +622,8 @@ class FileStoreTable(Table):
(CoreOptions.VECTOR_FIELD.key(),),
(CoreOptions.BLOB_FIELD.key(),),
(CoreOptions.BLOB_DESCRIPTOR_FIELD.key(),
'blob.stored-descriptor-fields'),
- (CoreOptions.BLOB_VIEW_FIELD.key(),)):
+ (CoreOptions.BLOB_VIEW_FIELD.key(),),
+ (CoreOptions.VIDEO_FRAME_FIELD.key(),)):
if any(key in dynamic_option_keys for key in keys):
# Preserve explicit overrides and removals, including those
from earlier copies.
continue
diff --git a/paimon-python/pypaimon/tests/column_directive_utils_test.py
b/paimon-python/pypaimon/tests/column_directive_utils_test.py
index 5e0c11730f..2b93c942ac 100644
--- a/paimon-python/pypaimon/tests/column_directive_utils_test.py
+++ b/paimon-python/pypaimon/tests/column_directive_utils_test.py
@@ -81,6 +81,33 @@ class TestParseAddColumnComment(unittest.TestCase):
class TestApplyAddColumnDirective(unittest.TestCase):
+ def test_video_frame_field(self):
+ for source in ("BYTES", "BINARY(8)", "BLOB"):
+ with self.subTest(source=source):
+ opts = {CoreOptions.VIDEO_FRAME_FIELD.key(): "existing"}
+ result = apply_add_column_directive(
+ "__VIDEO_FRAME_FIELD; camera frame", "video",
AtomicType(source), opts)
+ self.assertEqual(result.type.type, "BLOB")
+ self.assertTrue(result.type.nullable)
+ self.assertEqual(result.comment, "camera frame")
+ self.assertEqual(opts[CoreOptions.VIDEO_FRAME_FIELD.key()],
"existing,video")
+ result = apply_add_column_directive(
+ "__VIDEO_FRAME_FIELD", "second", AtomicType(source), opts)
+ self.assertIsNone(result.comment)
+ self.assertEqual(opts[CoreOptions.VIDEO_FRAME_FIELD.key()],
"existing,video,second")
+
+ def test_invalid_video_frame_field(self):
+ opts = {}
+ for comment in ("__VIDEO_UNKNOWN", "__VIDEO_FRAME_FIELD_typo"):
+ with self.subTest(comment=comment), self.assertRaises(ValueError):
+ apply_add_column_directive(comment, "video",
AtomicType("BYTES"), opts)
+ self.assertEqual(opts, {})
+ for source in (AtomicType("INT"), ArrayType(True, AtomicType("BYTES")),
+ MapType(True, AtomicType("STRING"),
AtomicType("BYTES"))):
+ with self.subTest(source=source), self.assertRaises(ValueError):
+ apply_add_column_directive("__VIDEO_FRAME_FIELD", "video",
source, opts)
+ self.assertEqual(opts, {})
+
def test_non_directive_returns_none(self):
opts = {}
result = apply_add_column_directive(
@@ -259,6 +286,15 @@ class TestRemoveDroppedDirectiveOptions(unittest.TestCase):
self.assertNotIn(CoreOptions.BLOB_DESCRIPTOR_FIELD.key(), opts)
self.assertEqual(opts[CoreOptions.VECTOR_FIELD.key()], "v")
+ def test_drop_video(self):
+ opts = {CoreOptions.VIDEO_FRAME_FIELD.key(): "video,second",
+ CoreOptions.BLOB_FIELD.key(): "image"}
+ remove_dropped_directive_options("video", "BLOB", opts)
+ self.assertEqual(opts[CoreOptions.VIDEO_FRAME_FIELD.key()], "second")
+ remove_dropped_directive_options("second", "BLOB", opts)
+ self.assertNotIn(CoreOptions.VIDEO_FRAME_FIELD.key(), opts)
+ self.assertEqual(opts[CoreOptions.BLOB_FIELD.key()], "image")
+
def test_drop_vector(self):
opts = {
CoreOptions.VECTOR_FIELD.key(): "emb,emb2",
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
index e2f665e8b1..13d306269c 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -144,6 +144,113 @@ class MultimodalTableTest(unittest.TestCase):
self.assertEqual(["id", "content", "embedding", "payload"],
[field.name for field in table.raw_table.fields])
+ def test_add_video_column_to_existing_images(self):
+ from pypaimon.schema.data_types import AtomicType
+ from pypaimon.schema.schema_change import SchemaChange
+
+ table = self.conn.create_table(
+ "evolved_video", schema=_schema({"id": pa.int32(), "image":
pa.large_binary()}),
+ options=_PARQUET_OPTIONS)
+ table.add([{"id": 0, "image": b"image-0"}, {"id": 1, "image":
b"image-1"}])
+ table.raw_table.create_tag("before_video")
+ old_table = table
+ self.conn.catalog.alter_table("default.evolved_video", [
+ SchemaChange.add_column("video", AtomicType("BYTES"),
"__VIDEO_FRAME_FIELD"),
+ ])
+ table = self.conn.get_table("evolved_video")
+ self.assertEqual([None, None], [r["video"] for r in
table.scan().to_list()])
+ old_table.add([{"id": 2, "image": b"image-2"}])
+ video_path = os.path.join(self.temp_dir, "backfill.mp4")
+ with open(video_path, "wb") as out:
+ out.write(b"complete-video-payload")
+ table.replace_video("id < 2", pmm.Blob.from_local(video_path),
first_frame=7)
+ # A retained writer must not stamp its old schema onto a new snapshot.
+ old_table.add([{"id": 3, "image": b"image-3"}])
+ snapshot = table.raw_table.snapshot_manager().get_latest_snapshot()
+ self.assertEqual(table.raw_table.table_schema.id, snapshot.schema_id)
+ table.raw_table.create_tag("after_old_append")
+ rows = sorted(table.scan().select(["id", "image", "video",
"_ROW_ID"]).to_list(),
+ key=lambda row: row["id"])
+ self.assertEqual([0, 1, 2, 3], [row["_ROW_ID"] for row in rows])
+ self.assertEqual([7, 8],
[pmm.VideoFrameDescriptor.deserialize(row["video"]).frame_index
+ for row in rows[:2]])
+ self.assertTrue(all(row["video"] is None for row in rows[2:]))
+ _, images = table.scan().read_blobs("image")
+ self.assertEqual([b"image-0", b"image-1", b"image-2", b"image-3"],
images["image"])
+ _, videos = table.scan().where("id < 2").read_blobs("video")
+ self.assertEqual([b"complete-video-payload"] * 2, videos["video"])
+ for options in ({"scan.snapshot-id": str(snapshot.id)},
{"scan.tag-name": "after_old_append"}):
+ with self.subTest(options=options):
+ historical = table.raw_table.copy(options)
+ self.assertEqual(["id", "image", "video"], [field.name for
field in historical.fields])
+ read = historical.new_read_builder()
+ restored =
sorted(read.new_read().to_arrow(read.new_scan().plan().splits()).to_pylist(),
+ key=lambda row: row["id"])
+ self.assertEqual([0, 1, 2, 3], [row["id"] for row in restored])
+ self.assertEqual([row["video"] for row in rows], [row["video"]
for row in restored])
+ historical =
self.conn.catalog.get_table("default.evolved_video").copy({"scan.tag-name":
"before_video"})
+ self.assertNotIn("video-frame-field", historical.table_schema.options)
+ # The retained old schema still reads the original rows after the
backfill.
+ read = historical.new_read_builder()
+ historical_rows =
read.new_read().to_arrow(read.new_scan().plan().splits()).to_pylist()
+ self.assertEqual([0, 1], sorted(row["id"] for row in historical_rows))
+ self.assertTrue(all(row.get("video") is None for row in
historical_rows))
+
+ self.conn.catalog.alter_table("default.evolved_video", [
+ SchemaChange.add_column("second", AtomicType("BYTES"),
"__VIDEO_FRAME_FIELD")])
+ for column, remaining in (("video", "second"), ("second", None)):
+ self.conn.catalog.alter_table("default.evolved_video",
[SchemaChange.drop_column(column)])
+ current = self.conn.get_table("evolved_video")
+ self.assertEqual(remaining,
current.raw_table.table_schema.options.get("video-frame-field"))
+ self.assertEqual(4, len(current.scan().to_list()))
+ self.conn.catalog.alter_table("default.evolved_video", [
+ SchemaChange.add_column("video", AtomicType("BYTES"),
"__VIDEO_FRAME_FIELD")])
+ current = self.conn.get_table("evolved_video")
+ self.assertEqual([None] * 4, [row["video"] for row in
current.scan().to_list()])
+
+ def test_add_video_column_validation(self):
+ from pypaimon.schema.data_types import AtomicType
+ from pypaimon.schema.schema_change import SchemaChange as C
+
+ table = self.conn.create_table(
+ "video_evolution_validation",
+ schema=_schema({"id": pa.int32(), "image": pa.large_binary()}),
+ options=_PARQUET_OPTIONS)
+ table.add([{"id": 0, "image": b"original"}])
+ identifier = "default.video_evolution_validation"
+
+ with self.assertRaisesRegex(ValueError, "immutable option"):
+ table.raw_table.copy({"video-frame-field": "image"})
+
+ def alter(changes):
+ self.conn.catalog.alter_table(identifier, changes)
+
+ for changes in [
+ [C.set_option("video-frame-field", "image")],
+ [C.set_option("video-frame-field", "missing")],
+ [C.add_column("bad", AtomicType("INT"), "__VIDEO_FRAME_FIELD")],
+ [C.add_column("bad", AtomicType("BLOB", False),
"__VIDEO_FRAME_FIELD")],
+ [C.add_column("bad", AtomicType("BLOB")),
C.set_option("video-frame-field", "image,bad")],
+ ]:
+ with self.subTest(changes=changes):
+ with self.assertRaises(RuntimeError):
+ alter(changes)
+ self.assertEqual(0,
self.conn.catalog.get_table(identifier).table_schema.id)
+ alter([C.update_column_comment("image", "__VIDEO_FRAME_FIELD")])
+ self.assertNotIn("video-frame-field",
self.conn.catalog.get_table(identifier).table_schema.options)
+ alter([C.add_column("video", AtomicType("BYTES"),
"__VIDEO_FRAME_FIELD")])
+ alter([C.add_column("second", AtomicType("BLOB"),
"__VIDEO_FRAME_FIELD")])
+ self.assertEqual(
+ "video,second",
self.conn.catalog.get_table(identifier).table_schema.options["video-frame-field"])
+ for change in [C.set_option("video-frame-field", "second"),
+ C.set_option("video-frame-field", ""),
C.remove_option("video-frame-field")]:
+ with self.assertRaises(RuntimeError):
+ alter([change])
+ # A column added by a previous commit may already contain ordinary
BLOB data.
+ alter([C.add_column("later", AtomicType("BLOB"))])
+ with self.assertRaisesRegex(RuntimeError, "not supported yet"):
+ alter([C.set_option("video-frame-field", "video,second,later")])
+
def test_add_video_embeds_frame_ordinals_outside_normal_data(self):
table = self.conn.create_table(
"video_frames",
diff --git a/paimon-python/pypaimon/tests/native_merge_into_test.py
b/paimon-python/pypaimon/tests/native_merge_into_test.py
index fe15bac7ca..738cf43cea 100644
--- a/paimon-python/pypaimon/tests/native_merge_into_test.py
+++ b/paimon-python/pypaimon/tests/native_merge_into_test.py
@@ -195,8 +195,15 @@ def
test_native_merge_keeps_video_inserts_on_python_path(tmp_path):
from pypaimon.write.native_merge_into import create_native_merge_into
from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
- table = _table(tmp_path).copy({'video-frame-field': 'name'})
- source = pa.Table.from_pylist([dict(id=1, value=10, name='frame')],
schema=_SCHEMA)
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('db', True)
+ schema = pa.schema([('id', pa.int32()), ('video', pa.large_binary())])
+ catalog.create_table('db.t', Schema.from_pyarrow_schema(schema, options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'video-frame-field': 'video', 'write.native.enabled': 'true',
+ }), False)
+ table = catalog.get_table('db.t')
+ source = pa.Table.from_pylist([dict(id=1, video=None)], schema=schema)
assert create_native_merge_into(table, source, ['id'], [],
[WhenNotMatched('*')],
'merge', BATCH_COMMIT_IDENTIFIER) is None
diff --git
a/paimon-python/pypaimon/tests/table/time_travel_schema_evolution_test.py
b/paimon-python/pypaimon/tests/table/time_travel_schema_evolution_test.py
index 17ba35de6a..45fdfe4683 100644
--- a/paimon-python/pypaimon/tests/table/time_travel_schema_evolution_test.py
+++ b/paimon-python/pypaimon/tests/table/time_travel_schema_evolution_test.py
@@ -63,6 +63,43 @@ class TestTimeTravelSchemaEvolution:
builder = table.new_read_builder()
return
builder.new_read().to_arrow(builder.new_scan().plan().splits()).to_pylist()
+ def test_commit_retry_refreshes_snapshot_schema(self, monkeypatch):
+ original = self._create([DataField(0, 'id', AtomicType('INT'))])
+ self._write(original, {'id': 1})
+ builder = original.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ snapshot_commit = commit.file_store_commit.snapshot_commit
+ publish = snapshot_commit.commit
+ attempted_schemas = []
+
+ def publish_after_schema_change(*args):
+ attempted_schemas.append(args[1].schema_id)
+ if len(attempted_schemas) == 1:
+ self.catalog.alter_table('test.t', [
+ SchemaChange.add_column('new_column',
AtomicType('STRING'))])
+ return False
+ return publish(*args)
+
+ monkeypatch.setattr(snapshot_commit, 'commit',
publish_after_schema_change)
+ try:
+ writer.write_arrow(pa.Table.from_pylist([{'id': 2}],
schema=pa.schema([('id', pa.int32())])))
+ messages = writer.prepare_commit()
+ commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+
+ current = self.catalog.get_table('test.t')
+ snapshot = current.snapshot_manager().get_latest_snapshot()
+ assert attempted_schemas == [0, 1]
+ assert snapshot.schema_id == current.table_schema.id == 1
+ historical = current.copy({'scan.snapshot-id': str(snapshot.id)})
+ assert sorted(self._read(historical), key=lambda row: row['id']) == [
+ {'id': 1, 'new_column': None}, {'id': 2, 'new_column': None}]
+ assert all(file.schema_id == 0
+ for split in
current.new_read_builder().new_scan().plan().splits()
+ for file in split.files)
+
@pytest.mark.parametrize('key,directive,field_type', [
('vector-field', '__VECTOR_FIELD;3', ArrayType(True,
AtomicType('FLOAT'))),
('blob-field', '__BLOB_FIELD', AtomicType('BYTES')),
diff --git a/paimon-python/pypaimon/tests/video_column_reassign_test.py
b/paimon-python/pypaimon/tests/video_column_reassign_test.py
new file mode 100644
index 0000000000..bcd8cd3ba0
--- /dev/null
+++ b/paimon-python/pypaimon/tests/video_column_reassign_test.py
@@ -0,0 +1,140 @@
+# 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.
+
+from dataclasses import replace
+from uuid import uuid4
+
+import pyarrow as pa
+import pytest
+import pypaimon.multimodal as pm
+
+from pypaimon.manifest.manifest_file_manager import ManifestFileManager
+from pypaimon.manifest.manifest_list_manager import ManifestListManager
+from pypaimon.schema.data_types import AtomicType
+from pypaimon.schema.schema_change import SchemaChange
+from pypaimon.snapshot.renaming_snapshot_commit import RenamingSnapshotCommit
+from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
+
+
+def _publish_reassigned_partition(table, partition_value):
+ """Publish the metadata-only OVERWRITE shape emitted by Java reassign.
+
+ The real Java reassign algorithm is exercised by
DataEvolutionRowIdReassignerTest.
+ Preserve shared row ranges across normal/image column files in this
fixture.
+ """
+ latest = table.snapshot_manager().get_latest_snapshot()
+ manifest_file = ManifestFileManager(table)
+ manifest_list = ManifestListManager(table)
+ entries =
manifest_file.read_entries_parallel(manifest_list.read_all(latest),
drop_stats=False)
+ next_row_id = latest.next_row_id
+ mapping = {}
+ reassigned = []
+ for entry in entries:
+ if entry.partition.values[0] == partition_value:
+ key = (entry.file.first_row_id, entry.file.row_count)
+ if key not in mapping:
+ mapping[key] = next_row_id
+ next_row_id += entry.file.row_count
+ entry = entry.assign_first_row_id(mapping[key])
+ reassigned.append(entry)
+ token = str(uuid4())
+ manifests = manifest_file.rolling_write(reassigned, 1 << 30, 'manifest-' +
token)
+ base, delta = 'manifest-list-' + token + '-0', 'manifest-list-' + token +
'-1'
+ manifest_list.write(base, manifests)
+ manifest_list.write(delta, [])
+ snapshot = replace(latest, id=latest.id + 1, base_manifest_list=base,
+ delta_manifest_list=delta, delta_record_count=0,
+ commit_user='video-reassign-test', commit_identifier=0,
+ commit_kind='OVERWRITE', time_millis=latest.time_millis
+ 1,
+ next_row_id=next_row_id, uuid=str(uuid4()))
+ assert
RenamingSnapshotCommit(table.snapshot_manager()).commit(latest.uuid, snapshot,
[])
+
+
[email protected]('phase', ['after_mapping', 'commit_cas'])
[email protected]('concurrent_change', ['same_partition',
'other_partition', 'append'])
+def test_video_backfill_reassign_interleavings(tmp_path, monkeypatch, phase,
concurrent_change):
+ conn = pm.connect(options={'warehouse': str(tmp_path / 'warehouse')})
+ table = conn.create_table('images', schema=pa.schema([
+ ('pt', pa.string()), ('id', pa.int32()), ('image',
pa.large_binary())]),
+ partitioned=['pt'], options={'file.format': 'parquet'})
+ for group in range(4):
+ table.add([{'pt': 'a' if group % 2 == 0 else 'b', 'id': i, 'image':
bytes([i])}
+ for i in range(group * 2, group * 2 + 2)])
+ conn.catalog.alter_table('default.images', [
+ SchemaChange.add_column('video', AtomicType('BYTES'),
'__VIDEO_FRAME_FIELD')])
+ table = conn.get_table('images')
+ raw = table.raw_table
+ source = tmp_path / 'source.mp4'
+ source.write_bytes(b'complete-video-payload')
+ video = pm.Blob.from_local(str(source))
+ injected = []
+
+ def concurrent_commit():
+ injected.append(True)
+ if concurrent_change == 'append':
+ table.add([{'pt': 'b', 'id': 8, 'image': bytes([8]), 'video':
None}])
+ else:
+ _publish_reassigned_partition(raw, 'a' if concurrent_change ==
'same_partition' else 'b')
+ injected.append(raw.snapshot_manager().get_latest_snapshot().id)
+
+ real_init = TableUpdateByRowId.__init__
+ real_commit = RenamingSnapshotCommit.commit
+
+ def init_after_reassign(updater, *args, **kwargs):
+ if not injected:
+ concurrent_commit()
+ real_init(updater, *args, **kwargs)
+
+ def commit_after_reassign(committer, base_uuid, snapshot, statistics):
+ if not injected and snapshot.commit_kind == 'APPEND':
+ concurrent_commit()
+ # Let the real compare-and-swap lose to the concurrent snapshot,
+ # then exercise the updater's retry validation.
+ return real_commit(committer, base_uuid, snapshot, statistics)
+
+ with monkeypatch.context() as patch:
+ if phase == 'after_mapping':
+ patch.setattr(TableUpdateByRowId, '__init__', init_after_reassign)
+ else:
+ patch.setattr(RenamingSnapshotCommit, 'commit',
commit_after_reassign)
+ if concurrent_change == 'same_partition':
+ with pytest.raises((ValueError, RuntimeError), match='valid
range|Row ID existence conflict'):
+ table.replace_video('id < 2', video, first_frame=7)
+ else:
+ table.replace_video('id < 2', video, first_frame=7)
+ assert injected
+
+ if concurrent_change == 'same_partition':
+ assert raw.snapshot_manager().get_latest_snapshot().id == injected[1]
+ assert all(row['video'] is None for row in table.scan().to_list())
+ # A fresh read rebuilds row-ID -> frame mapping; never reuse stale
commit messages.
+ table.replace_video('id < 2', video, first_frame=7)
+
+ rows = sorted(table.scan().select(['id', 'video']).to_list(), key=lambda
row: row['id'])
+ assert [r['id'] for r in rows] == list(range(9 if concurrent_change ==
'append' else 8))
+ for row in rows:
+ if row['id'] < 2:
+ descriptor = pm.VideoFrameDescriptor.deserialize(row['video'])
+ assert descriptor.frame_index == row['id'] + 7
+ assert descriptor.uri.endswith('.video')
+ else:
+ assert row['video'] is None
+ scalar, images = table.scan().read_blobs('image')
+ assert dict(zip(scalar['id'].to_pylist(), images['image'])) == {
+ row['id']: bytes([row['id']]) for row in rows}
+ _, videos = table.scan().where('id < 2').read_blobs('video')
+ assert videos['video'] == [b'complete-video-payload'] * 2
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index c00f138921..d5390088e3 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -689,7 +689,7 @@ class FileStoreCommit:
snapshot_data = Snapshot(
version=3,
id=new_snapshot_id,
- schema_id=self.table.table_schema.id,
+ schema_id=self.table.schema_manager.latest().id,
base_manifest_list=base_manifest_list,
base_manifest_list_size=base_manifest_list_size,
delta_manifest_list=delta_manifest_list,
