Akash3121 commented on code in PR #10114:
URL: https://github.com/apache/paimon/pull/10114#discussion_r4089606735
##########
paimon-python/pypaimon/write/file_store_write.py:
##########
@@ -332,6 +336,7 @@ def prepare_commit(self, commit_identifier) ->
List[CommitMessage]:
new_files=committed_files,
changelog_files=changelog_files,
total_buckets=self._runtime_total_buckets.get(partition),
+ preserve_blob_files_on_abort=preserve_blob_files,
Review Comment:
This ownership flag is only present on the Python `CommitMessage` ; the
Java-v14 serializer used by native commit does not encode it. With
`commit.native.enabled=true` , `TableCommit.abort()` converts this message to
a native message and calls `native.abort()` , so the native side receives no
indication that the `BlobConsumer` owns these `.blob` packs and performs
normal abort cleanup. That can delete data already handed to the consumer. The
smallest safe fix is to make native commit/abort unsupported whenever any
message has `preserve_blob_files_on_abort=True`, so abort falls back to
`_abort_commit_messages` ; alternatively the native wire format and abort
implementation must carry and honor this ownership state.
Please add a `TableCommit.abort()` test with native commit enabled and a
preservation-marked message, asserting the native path is not used and the pack
remains.
##########
paimon-python/pypaimon/blob/managed_blob_reference_file.py:
##########
@@ -0,0 +1,173 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import io
+import struct
+import zlib
+from typing import List, Optional, Tuple
+
+from pypaimon.index.pk.primary_key_index_source_meta import (
+ _decode_modified_utf8,
+ _encode_modified_utf8,
+)
+
+
+class ManagedBlobReferenceFile:
+ """Versioned metadata listing managed BLOB packs referenced by one data
file."""
+
+ MAGIC = 0x50424C52
+ VERSION = 1
+ MANAGED_BLOB_SUFFIX = ".managed.blob"
+ REFERENCE_FILE_SUFFIX = ".blobref"
+
+ @staticmethod
+ def from_descriptor_uri(descriptor_uri: str) -> Optional["Reference"]:
+ if not
descriptor_uri.endswith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX):
+ return None
+ parent, _, name = descriptor_uri.rpartition("/")
+ if not parent or not name:
+ return None
+ return Reference(parent, name)
+
+ @staticmethod
+ def sidecar_path(data_file_path: str) -> str:
+ return data_file_path + ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX
+
+ @staticmethod
+ def sidecar_name(data_file_name: str) -> str:
+ return data_file_name + ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX
+
+ @staticmethod
+ def write(file_io, path: str, references: List["Reference"]) -> None:
+ normalized = sorted(
+ references,
+ key=lambda ref: (ref.storage_root_id, ref.relative_path),
+ )
+ unique: List[Reference] = []
+ for reference in normalized:
+ if not unique or reference != unique[-1]:
+ unique.append(reference)
+
+ payload = io.BytesIO()
+ payload.write(struct.pack(">B", ManagedBlobReferenceFile.VERSION))
+ payload.write(struct.pack(">i", len(unique)))
+ for reference in unique:
+ _write_modified_utf(payload, reference.storage_root_id)
+ _write_modified_utf(payload, reference.relative_path)
+
+ payload_bytes = payload.getvalue()
+ checksum = zlib.crc32(payload_bytes) & 0xFFFFFFFF
+ checksum_signed = (
+ checksum if checksum < 0x80000000 else checksum - 0x100000000
+ )
+ try:
+ with file_io.new_output_stream(path) as out:
+ out.write(struct.pack(">i", ManagedBlobReferenceFile.MAGIC))
+ out.write(payload_bytes)
+ out.write(struct.pack(">i", checksum_signed))
+ except Exception:
+ file_io.delete_quietly(path)
+ raise
+
+ @staticmethod
+ def read(file_io, path: str) -> List["Reference"]:
+ with file_io.new_input_stream(path) as stream:
+ data = stream.read()
+ # magic 4 + version 1 + count 4 + crc 4. Shorter input must not
+ # surface as struct.error from the checksum unpack.
+ if len(data) < 13:
+ raise IOError("Invalid managed BLOB reference file: too short")
+ magic = struct.unpack_from(">i", data, 0)[0]
+ if magic != ManagedBlobReferenceFile.MAGIC:
+ raise IOError("Invalid managed BLOB reference file magic: %s" %
magic)
+
+ offset = 4
+ version = data[offset]
+ offset += 1
+ if version != ManagedBlobReferenceFile.VERSION:
+ raise IOError("Unsupported managed BLOB reference file version:
%s" % version)
+
+ count = struct.unpack_from(">i", data, offset)[0]
+ offset += 4
+ if count < 0:
+ raise IOError("Invalid managed BLOB reference count: %s" % count)
+
+ references: List[Reference] = []
+ for _ in range(count):
+ root, offset = _read_modified_utf(data, offset)
+ rel, offset = _read_modified_utf(data, offset)
+ references.append(Reference(root, rel))
+
+ if offset + 4 > len(data):
+ raise IOError("Invalid managed BLOB reference file checksum")
+ expected_checksum = struct.unpack_from(">i", data, offset)[0] &
0xFFFFFFFF
+ actual_checksum = zlib.crc32(data[4:offset]) & 0xFFFFFFFF
+ if expected_checksum != actual_checksum:
+ raise IOError(
+ "Invalid managed BLOB reference file checksum. Expected %s but
computed %s."
+ % (expected_checksum, actual_checksum))
+ if offset + 4 != len(data):
+ raise IOError("Unexpected trailing bytes in managed BLOB reference
file.")
+ return references
+
+
+class Reference:
+ """Exact identity of a managed BLOB payload pack."""
+
+ __slots__ = ("storage_root_id", "relative_path")
+
+ def __init__(self, storage_root_id: str, relative_path: str):
+ if not storage_root_id:
+ raise ValueError("Managed BLOB storage root must not be empty.")
+ if not relative_path or "/" in relative_path or relative_path in (".",
".."):
+ raise ValueError(
+ "Managed BLOB relative path must be a file name: %s." %
relative_path)
+ self.storage_root_id = storage_root_id
+ self.relative_path = relative_path
+
+ def __eq__(self, other) -> bool:
+ if not isinstance(other, Reference):
+ return False
+ return (
+ self.storage_root_id == other.storage_root_id
+ and self.relative_path == other.relative_path
+ )
+
+ def __hash__(self) -> int:
+ return hash((self.storage_root_id, self.relative_path))
+
+ def __repr__(self) -> str:
+ return "%s/%s" % (self.storage_root_id, self.relative_path)
+
+
+def _write_modified_utf(out: io.BytesIO, value: str) -> None:
+ encoded = _encode_modified_utf8(value)
+ if len(encoded) > 65535:
+ raise ValueError("Modified UTF-8 string is too long.")
+ out.write(struct.pack(">H", len(encoded)))
+ out.write(encoded)
+
+
+def _read_modified_utf(data: bytes, offset: int) -> Tuple[str, int]:
+ if offset + 2 > len(data):
+ raise IOError("Truncated modified UTF-8 length.")
+ length = struct.unpack_from(">H", data, offset)[0]
+ offset += 2
+ end = offset + length
+ if end > len(data):
+ raise IOError("Truncated modified UTF-8 payload.")
+ return _decode_modified_utf8(data[offset:end]), end
Review Comment:
non-blocking
The reused decoder does not validate that continuation bytes have the
`10xxxxxx` form. For example, a CRC-valid `.blobref ` containing `b"\xc0A"`
is accepted here and decoded as `"\x01"` , while Java
`DataInputStream.readUTF` rejects that sequence. Since this format is intended
to be Java-compatible, malformed files can be interpreted differently by Java
and Python. Could we make the shared decoder strict (or validate locally),
convert the decode failure to `IOError` , and add a CRC-valid
malformed-continuation fixture that both implementations reject?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
