[python] Externalize primary-key BLOBs and keep shared packs reachable.

paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java

@@ -59,6 +59,7 @@
 import org.apache.paimon.manifest.FileKind;
 import org.apache.paimon.manifest.IndexManifestEntry;
 import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.operation.LocalManagedBlobOrphanFilesClean;
 import org.apache.paimon.options.MemorySize;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.predicate.Predicate;
@@ -70,6 +71,7 @@
 import org.apache.paimon.table.AppendOnlyFileStoreTable;
 import org.apache.paimon.table.CatalogEnvironment;
 import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
 import org.apache.paimon.table.PrimaryKeyFileStoreTable;
 import org.apache.paimon.table.Table;
 import org.apache.paimon.table.sink.BatchTableCommit;
@@ -104,6 +106,7 @@
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Paths;
+import java.nio.file.attribute.FileTime;
 import java.time.Instant;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
@@ -116,6 +119,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -732,6 +736,64 @@ public void testReadPythonDynamicBucketHashIndex() throws Exception {
                         "hello-java, 42, java-new", "python-only, 7, python-only");
     }
 
+    @Test
+    @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+    public void testReadPythonManagedBlobAndClean() throws Exception {
+        java.nio.file.Path tablePath =
+                tempDir.resolve("warehouse/default.db/managed_blob_python_to_java");
+        FileStoreTable table =
+                FileStoreTableFactory.create(LocalFileIO.create(), new Path(tablePath.toString()));
+        assertPythonManagedBlobRows(table);
+        Long snapshotId = table.snapshotManager().latestSnapshotId();
+
+        List<java.nio.file.Path> packs;
+        try (Stream<java.nio.file.Path> files = Files.walk(tablePath)) {
+            packs =
+                    files.filter(path -> path.toString().endsWith(".managed.blob"))
+                            .collect(Collectors.toList());
+        }
+        assertThat(packs).hasSize(3);
+        long now = System.currentTimeMillis();
+        FileTime oldTime = FileTime.fromMillis(now - TimeUnit.HOURS.toMillis(72));
+        for (java.nio.file.Path pack : packs) {
+            Files.setLastModifiedTime(pack, oldTime);
+        }
+        java.nio.file.Path orphan = packs.get(0).resolveSibling("orphan.managed.blob");
+        Files.copy(packs.get(0), orphan);
+        Files.setLastModifiedTime(orphan, oldTime);
+
+        try (LocalManagedBlobOrphanFilesClean cleaner =
+                new LocalManagedBlobOrphanFilesClean(
+                        table, now - TimeUnit.HOURS.toMillis(24), false)) {
+            assertThat(cleaner.clean().getDeletedFileCount()).isEqualTo(1);
+        }
+        assertThat(Files.exists(orphan)).isFalse();
+        for (java.nio.file.Path pack : packs) {
+            assertThat(Files.exists(pack)).isTrue();
+        }
+        assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId);
+        assertPythonManagedBlobRows(table);
+    }
+
+    private void assertPythonManagedBlobRows(FileStoreTable table) throws Exception {
+        List<String> rows =
+                getResult(
+                        table.newRead(),
+                        table.newScan().plan().splits(),
+                        row -> {
+                            assertThat(row.getBlob(1).toData())
+                                    .isEqualTo("python-payload".getBytes(StandardCharsets.UTF_8));
+                            assertThat(row.getArray(2).getBlob(0).toData())
+                                    .isEqualTo("python-item".getBytes(StandardCharsets.UTF_8));
+                            InternalMap map = row.getMap(3);
+                            assertThat(map.keyArray().getString(0).toString()).isEqualTo("k");
+                            assertThat(map.valueArray().getBlob(0).toData())
+                                    .isEqualTo("python-map".getBytes(StandardCharsets.UTF_8));
+                            return Integer.toString(row.getInt(0));
+                        });
+        assertThat(rows).containsExactly("1");
+    }
+
     private int assignDynamicBucket(FileStoreTable table, InternalRow row) {
         InternalRowSerializer serializer =
                 new InternalRowSerializer(DataTypes.STRING(), DataTypes.BIGINT());

paimon-python/README.md

@@ -300,9 +300,10 @@ mixing Arrow batches and Python `Blob` objects in one writer.
 
 A successful `prepare_commit()` transfers file ownership to the caller.
 Writer `abort()` and `close()` preserve those prepared files and may clean only
-outputs still owned by the writer. Explicit `commit.abort(messages)` deletes
-newly written data, changelog, sidecars and added indexes, matching Java's
-Paimon Table committer. Call it only when those messages are known to be
+outputs still owned by the writer. A BlobConsumer keeps that writer from
+deleting packs it has already exposed. Explicit `commit.abort(messages)` deletes
+newly written data, BLOB packs, changelog, sidecars and added indexes, matching
+Java's Paimon Table committer. Call it only when those messages are known to be
 uncommitted and will never be submitted. Never call it after a commit with an
 unknown outcome: publication may have succeeded before its response failed.
 Internal failure paths preserve prepared files instead of calling commit abort.

paimon-python/dev/run_mixed_tests.sh

@@ -242,7 +242,7 @@ run_python_write_test() {
 
     # Run the parameterized Python test method for writing data (pk table, includes bucket num assertion)
     echo "Running Python test for JavaPyReadWriteTest (test_py_write_read_pk_table)..."
-    if python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest -k "test_py_write_read_pk_table or test_py_write_read_append_table or test_py_write_dynamic_bucket_hash_index or test_py_write_floating_sequence" -v; then
+    if python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest -k "test_py_write_read_pk_table or test_py_write_read_append_table or test_py_write_dynamic_bucket_hash_index or test_py_write_floating_sequence or test_py_write_managed_blob" -v; then
         echo -e "${GREEN}✓ Python write test completed successfully${NC}"
         return 0
     else
@@ -261,13 +261,21 @@ run_java_read_test() {
     echo "Running Maven test for JavaPyE2ETest.testReadPkTable (Java Read Parquet/Orc/Avro)..."
     echo "Note: Maven may download dependencies on first run, this may take a while..."
     local parquet_result=0
-    if mvn test -Dtest=org.apache.paimon.JavaPyE2ETest#testReadPkTable+testReadAppendTable+testReadPythonDynamicBucketHashIndex+testReadPythonFloatingSequence -pl paimon-core -Drun.e2e.tests=true -Dpython.version="$PYTHON_VERSION"; then
+    if mvn test -Dtest=org.apache.paimon.JavaPyE2ETest#testReadPkTable+testReadAppendTable+testReadPythonDynamicBucketHashIndex+testReadPythonFloatingSequence+testReadPythonManagedBlobAndClean -pl paimon-core -Drun.e2e.tests=true -Dpython.version="$PYTHON_VERSION"; then
         echo -e "${GREEN}✓ Java read Parquet/Orc/Avro test completed successfully${NC}"
     else
         echo -e "${RED}✗ Java read Parquet/Orc/Avro test failed${NC}"
         parquet_result=1
     fi
 
+    if [[ $parquet_result -eq 0 ]]; then
+        cd "$PAIMON_PYTHON_DIR"
+        if ! python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest::test_read_python_managed_blob_after_java_gc -v; then
+            parquet_result=1
+        fi
+        cd "$PROJECT_ROOT"
+    fi
+
     echo ""
 
     if [[ $parquet_result -eq 0 ]]; then

paimon-python/pypaimon/blob/__init__.py

@@ -0,0 +1,16 @@
+# 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.

paimon-python/pypaimon/blob/managed_blob_cleanup.py

@@ -0,0 +1,132 @@
+# 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.
+
+"""Reference-aware cleanup of shared primary-key BLOB packs."""
+
+import logging
+
+from pypaimon.blob.managed_blob_payload import resolve_table_blob_uri
+from pypaimon.blob.managed_blob_reference_file import ManagedBlobReferenceFile
+from pypaimon.utils.path import path_identity, to_file_io_path
+
+logger = logging.getLogger(__name__)
+
+
+def _reference_identities(table, reference):
+    uri = reference.storage_root_id + '/' + reference.relative_path
+    # Java Path strings contain literal URI characters. Older Python references
+    # may instead contain encoded file URIs or table-relative paths. A surviving
+    # reference must protect every possible interpretation of that path.
+    paths = {uri, resolve_table_blob_uri(table.table_path, uri)}
+    return {path_identity(candidate, table.file_io)
+            for path in paths for candidate in (path, to_file_io_path(path))}
+
+
+def managed_blob_deletion_paths(table, paths):
+    """Keep shared packs while deleting exclusively owned data files and sidecars."""
+    deleting = list(paths)
+    deleting.extend(_managed_pack_paths(table, deleting))
+    protected = _protected_managed_packs(table, deleting)
+    return [path for path in deleting if path_identity(path, table.file_io) not in protected]
+
+
+def _managed_pack_paths(table, paths):
+    """Collect colocated pack candidates before deleting aborted sidecars."""
+    packs = set()
+    for path in paths:
+        path = str(path)
+        if not path.endswith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX):
+            continue
+        try:
+            references = ManagedBlobReferenceFile.read(table.file_io, path)
+            parent = path_identity(path, table.file_io).rpartition('/')[0]
+            for reference in references:
+                for pack in _reference_identities(table, reference):
+                    if (pack.rpartition('/')[0] == parent
+                            and pack.endswith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)):
+                        packs.add(pack)
+        except Exception as error:
+            logger.warning(
+                "Failed to read managed BLOB references from %s during abort: %s",
+                path, error)
+    return packs
+
+
+def _protected_managed_packs(table, paths):
+    """Keep a pack when a blobref outside this abort still names it.
+
+    One pack can be referenced by several data files. Deleting it
+    with the first aborted file would make the other files unreadable.
+    """
+    pack_suffix = ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX
+    ref_suffix = ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX
+    deleting = {path_identity(path, table.file_io) for path in paths}
+    packs_by_dir = {}
+    for path in deleting:
+        if not path.endswith(pack_suffix):
+            continue
+        parent, _, name = path.rpartition("/")
+        if parent and name:
+            packs_by_dir.setdefault(parent, set()).add(path)
+    if not packs_by_dir:
+        return set()
+
+    list_status = getattr(table.file_io, "list_status", None)
+    if list_status is None:
+        return set(path for packs in packs_by_dir.values() for path in packs)
+
+    protected = set()
+    for parent, packs in packs_by_dir.items():
+        try:
+            statuses = list(list_status(parent))
+        except Exception as error:
+            logger.warning(
+                "Failed to list %s while checking managed BLOB packs: %s",
+                parent,
+                error,
+            )
+            # Discovery did not finish. Deleting the packs could remove
+            # payloads that a committed file still references.
+            protected.update(packs)
+            continue
+        for status in statuses:
+            name = getattr(status, "base_name", None)
+            if not name:
+                path = getattr(status, "path", None) or getattr(status, "original_path", None)
+                name = str(path).replace("\\", "/").rsplit("/", 1)[-1] if path else None
+            if not name:
+                protected.update(packs)
+                break
+            if not str(name).endswith(ref_suffix):
+                continue
+            blobref_path = parent + "/" + str(name)
+            if blobref_path in deleting:
+                continue
+            try:
+                references = ManagedBlobReferenceFile.read(table.file_io, blobref_path)
+                referenced = {identity for reference in references
+                              for identity in _reference_identities(table, reference)}
+            except Exception as error:
+                logger.warning(
+                    "Failed to read %s while checking managed BLOB packs: %s",
+                    blobref_path,
+                    error,
+                )
+                protected.update(packs)
+                break
+            protected.update(packs & referenced)
+    return protected